# state

> Distributed key-value state management with reactive change triggers — registers the `state` trigger type and the `state::*` functions.

| field | value |
|-------|-------|
| version | 0.21.0 |
| type | binary |
| repo | https://github.com/iii-hq/workers |
| supported_targets | x86_64-apple-darwin, aarch64-apple-darwin, i686-pc-windows-msvc, x86_64-pc-windows-msvc, aarch64-pc-windows-msvc, x86_64-unknown-linux-gnu, aarch64-unknown-linux-gnu, x86_64-unknown-linux-musl, armv7-unknown-linux-gnueabihf |
| author | iii |

## installation

```sh
iii worker add state@0.21.0
```

## configuration

```yaml
- adapter:
    config:
      file_path: ./data/state_store.db
      store_method: file_based
    name: kv
```

## dependencies

- `configuration` @ `^0.21.6`

## readme

# state

Distributed key-value state storage with scope-based organization and
reactive change triggers. Values are addressed by a `scope` (namespace) and a
`key`, shared across every worker connected to the engine, and persisted
through a pluggable adapter (`kv` or `redis`). Callers reach the store
through six functions — `state::set`, `state::get`, `state::delete`,
`state::update`, `state::list`, `state::list_groups` — and this worker also
registers the `state` trigger type, which fires `state:created`,
`state:updated`, or `state:deleted` after every successful mutation so
downstream functions can react to data changes without polling. This worker
is the standalone migration of the engine's built-in `iii-state`.

## Install

```bash
iii worker add state
```

`iii worker add` fetches the binary, writes a config block into
`~/.iii/config.yaml`, and the engine starts the worker on the next
`iii start`.

## Functions

| Function | Input | Returns | Fires |
|---|---|---|---|
| `state::set` | `{ scope, key, value }` (`data` accepted as an alias for `value`) | `{ old_value, new_value }` | `state:created` (new key) or `state:updated` |
| `state::get` | `{ scope, key }` | the value, or `null` | — |
| `state::delete` | `{ scope, key }` | the deleted value, or `null` | `state:deleted` (even when the key did not exist) |
| `state::update` | `{ scope, key, ops }` — ordered atomic ops: `set`, `merge`, `increment`, `decrement`, `append`, `remove` | `{ old_value, new_value, errors }` | `state:created` or `state:updated` |
| `state::list` | `{ scope }` | flat array of every value in the scope | — |
| `state::list_groups` | `{}` | `{ groups }` — sorted, deduplicated scope names | — |

## Trigger type

This worker always registers the `state` trigger type. Bind a function to it
with:

| Field | Required | Default | Description |
|---|---|---|---|
| `scope` | no | any scope | Only fire for writes in this scope. |
| `key` | no | any key | Only fire for writes to this key. |
| `condition_function_id` | no | — | Function invoked first with the event; only an explicit `false` return skips the handler (null/no result passes, an error skips and logs). |

Trigger delivery is asynchronous: handlers run after the write completes and
a handler failure never rolls the write back. Duplicate trigger ids replace
the previous binding silently (builtin parity).

```typescript
iii.registerTrigger({
  type: 'state',
  function_id: 'orders::on-status-change',
  config: { scope: 'orders', key: 'status' },
})
```

The handler receives the event payload:

```json
{
  "type": "state",
  "event_type": "state:updated",
  "scope": "orders",
  "key": "status",
  "old_value": { "status": "pending" },
  "new_value": { "status": "shipped" }
}
```

`event_type` is one of `state:created`, `state:updated`, `state:deleted`;
`old_value` is `null` for created keys and `new_value` is `null` for deleted
keys.

## Configuration

| Field | Default | Description |
|---|---|---|
| `adapter` | `kv` | Storage adapter: `kv` (in-process; `store_method: in_memory` or `file_based` with `file_path` and `save_interval_ms`) or `redis` (`redis_url`, default `redis://localhost:6379`). Restart-tier: a runtime change is logged and takes effect at the next worker start. |
| `triggers_enabled` | `true` | Globally enable/disable state change-trigger fan-out. Applied live. |
| `max_value_bytes` | unset (no limit) | Reject `state::set` writes whose JSON-serialized value exceeds this many bytes (`VALUE_TOO_LARGE`). Minimum `1`. Applied live. |
| `save_interval_ms` | `5000` | Persistence flush cadence (ms) for the file-backed `kv` adapter. `100`–`3600000`. Applied live by respawning the adapter's save loop (hot-retune; no-op for in-memory/redis). |

Configuration is owned by the `configuration` worker — edit it from the
console (**Configuration → Workers → state**) or seed it once via the
worker's config block in `config.yaml` on first boot. `triggers_enabled` and
`max_value_bytes` apply on the next write, `save_interval_ms` hot-retunes
the save loop, and `adapter` takes effect on the next restart.

### Requires removing the built-in `iii-state` worker

The built-in `iii-state` worker also owns the `state` trigger type and the
`state::*` functions. Two owners of the same surface on one engine collide —
whichever registers last wins — so this worker requires `iii-state` to be
absent: omit it from the engine's `config.yaml` (a config that doesn't list
a worker won't run it).

On boot, this worker queries the engine for connected workers and refuses to
start with a clear error if `iii-state` is still active, so a stale config
fails loudly instead of silently racing the built-in worker for ownership of
`state`.

**Store migration:** if the builtin used a file-based `kv` store, point this
worker's `adapter.config.file_path` at the builtin's existing directory
(the default engine config used `./data/state_store.db`). The on-disk format
(one rkyv `.bin` file per scope) is identical, so the existing data loads
as-is — no export/import step.

## Latency

The builtin ran in-process inside the engine, so a `state::*` call cost
microseconds; as a standalone worker every call crosses the engine⇄worker
WebSocket, which puts round-trips in the low-millisecond range. That
order-of-magnitude delta is inherent to the migration and applies to every
externalized builtin. A formal benchmark was waived by project decision
(2026-07-06).

## Parity vs builtin

| Behavior | Builtin | This worker |
|---|---|---|
| Function ids | `state::set/get/delete/update/list/list_groups` | same (exact) |
| `set` input | `{scope, key, value}` (`data` alias) | same |
| Events | `state:created`/`updated`/`deleted`, payload `{type:"state", event_type, scope, key, old_value, new_value}` | same |
| Trigger config | `{scope?, key?, condition_function_id?}`; only explicit `false` blocks | same |
| Duplicate trigger id | silent replace | same |
| Trigger `metadata` | forwarded to handlers via call_with_metadata | not forwarded (iii-sdk 0.20 TriggerRequest has no metadata field; same limitation as the http worker) |
| Store adapters | kv (in_memory/file_based), redis, bridge | kv, redis (bridge not ported — see docs/adr/0001) |
| kv on-disk format | rkyv `.bin` per scope | identical — builtin data loads as-is |
| `save_interval_ms` | default 5000ms, floor 100ms, hot-retune | same |
| `max_value_bytes` | guards `set` only, `VALUE_TOO_LARGE` | same (code is the message prefix) |
| Error codes | coded `ErrorBody` (`SET_ERROR`, ...) | code as message prefix (SDK handler errors carry a message) |
| Durability | store dies with the ENGINE process | store dies with the WORKER process (ADR 0001; file_based/redis unchanged) |
| Latency | in-process µs | engine⇄worker WS round-trip (low ms) — formal benchmark waived by project decision 2026-07-06 |
| Telemetry (`track_state_*`) | engine-internal counters | none — out of scope for this migration; lands with the shared worker observability story |

## api reference

```json
{
  "functions": [
    {
      "description": "Delete a value from state",
      "metadata": {},
      "name": "state::delete",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value to delete within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "scope"
        ],
        "title": "StateDeleteInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The value that was deleted (read before delete), or null if it did not exist.",
        "title": "StateDeleteResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "Get a value from state",
      "metadata": {},
      "name": "state::get",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "scope"
        ],
        "title": "StateGetInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The raw value stored at scope/key, or null if absent.",
        "title": "StateGetResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "Get a group from state",
      "metadata": {},
      "name": "state::list",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "scope": {
            "description": "Namespace whose keys should be listed as a group.",
            "type": "string"
          }
        },
        "required": [
          "scope"
        ],
        "title": "StateGetGroupInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The values in the given scope, as a JSON array (or null on the rare serialization failure).",
        "title": "StateListResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "List all state groups",
      "metadata": {},
      "name": "state::list_groups",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "StateListGroupsInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "groups": {
            "items": {
              "type": "string"
            },
            "type": "array"
          }
        },
        "required": [
          "groups"
        ],
        "title": "StateListGroupsResult",
        "type": "object"
      }
    },
    {
      "description": "Internal: reload state configuration from the authoritative store on change.",
      "metadata": {},
      "name": "state::on-config-change",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "ConfigChangeRequest",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "ok": {
            "type": "boolean"
          }
        },
        "required": [
          "ok"
        ],
        "title": "ConfigChangeAck",
        "type": "object"
      }
    },
    {
      "description": "Set a value in state",
      "metadata": {},
      "name": "state::set",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys (e.g. `users`, `orders`).",
            "type": "string"
          },
          "value": {
            "description": "Arbitrary JSON value to store. Replaces any existing value at `scope`/`key`."
          }
        },
        "required": [
          "key",
          "scope",
          "value"
        ],
        "title": "StateSetInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "new_value": {
            "description": "The value after the update"
          },
          "old_value": {
            "description": "The value before the update (None if key didn't exist)"
          }
        },
        "required": [
          "new_value"
        ],
        "title": "StreamSetResult",
        "type": "object"
      }
    },
    {
      "description": "Update a value in state",
      "metadata": {},
      "name": "state::update",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "MergePath": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "items": {
                  "type": "string"
                },
                "type": "array"
              }
            ],
            "description": "Path target for a [`UpdateOp::Merge`] operation. Accepts either a single string (legacy / first-level field) or an array of literal segments (nested path).\n\nPath normalization rules applied by the engine: - absent / `Single(\"\")` / `Segments(vec![])` → root merge - `Single(\"foo\")` is equivalent to `Segments(vec![\"foo\".into()])` - `Segments([\"a\", \"b\", \"c\"])` walks three literal keys, never interpreting dots specially. `Segments(vec![\"a.b\".into()])` is a single literal key named `\"a.b\"`.\n\n**Variant ordering is load-bearing.** `#[serde(untagged)]` tries variants in declaration order, `Single` MUST come before `Segments` so a JSON string deserializes into `Single` rather than failing the array match first."
          },
          "UpdateOp": {
            "description": "Operations that can be performed atomically on a stream value",
            "oneOf": [
              {
                "description": "Set a value at path (overwrite)",
                "properties": {
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "set"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Merge object into existing value (object-only). Path may be omitted (root merge), a single first-level key, or an array of literal segments for nested merge. See [`MergePath`].",
                "properties": {
                  "path": {
                    "anyOf": [
                      {
                        "$ref": "#/definitions/MergePath"
                      },
                      {
                        "type": "null"
                      }
                    ]
                  },
                  "type": {
                    "enum": [
                      "merge"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "type",
                  "value"
                ],
                "type": "object"
              },
              {
                "description": "Increment numeric value",
                "properties": {
                  "by": {
                    "format": "int64",
                    "type": "integer"
                  },
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "increment"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "by",
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Decrement numeric value",
                "properties": {
                  "by": {
                    "format": "int64",
                    "type": "integer"
                  },
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "decrement"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "by",
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Append an element to an array or concatenate a string at the optional path. Path may be omitted (root append), a single first-level key, or an array of literal segments for nested append. See [`MergePath`] for the variant shape.",
                "properties": {
                  "path": {
                    "anyOf": [
                      {
                        "$ref": "#/definitions/MergePath"
                      },
                      {
                        "type": "null"
                      }
                    ]
                  },
                  "type": {
                    "enum": [
                      "append"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "type",
                  "value"
                ],
                "type": "object"
              },
              {
                "description": "Remove a field",
                "properties": {
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "remove"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "path",
                  "type"
                ],
                "type": "object"
              }
            ]
          }
        },
        "properties": {
          "key": {
            "description": "Identifier for the value to update within the scope.",
            "type": "string"
          },
          "ops": {
            "description": "Ordered list of update operations applied atomically to the existing value.",
            "items": {
              "$ref": "#/definitions/UpdateOp"
            },
            "type": "array"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "ops",
          "scope"
        ],
        "title": "StateUpdateInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "UpdateOpError": {
            "description": "Per-op error reported by an atomic update operation.",
            "properties": {
              "code": {
                "description": "Stable error code, e.g. `\"merge.path.too_deep\"`.",
                "type": "string"
              },
              "doc_url": {
                "description": "Optional documentation URL for this error class.",
                "type": [
                  "string",
                  "null"
                ]
              },
              "message": {
                "description": "Human-readable description with concrete numbers when applicable.",
                "type": "string"
              },
              "op_index": {
                "description": "Index of the offending op within the original `ops` array.",
                "format": "uint",
                "minimum": 0,
                "type": "integer"
              }
            },
            "required": [
              "code",
              "message",
              "op_index"
            ],
            "type": "object"
          }
        },
        "description": "Result of an atomic update operation",
        "properties": {
          "errors": {
            "description": "Errors encountered while applying ops. Successfully applied ops are still reflected in `new_value`. Field is omitted from JSON when empty for backward compatibility.",
            "items": {
              "$ref": "#/definitions/UpdateOpError"
            },
            "type": "array"
          },
          "new_value": {
            "description": "The value after the update"
          },
          "old_value": {
            "description": "The value before the update (None if key didn't exist)"
          }
        },
        "required": [
          "new_value"
        ],
        "title": "StreamUpdateResult",
        "type": "object"
      }
    }
  ],
  "triggers": [
    {
      "description": "State trigger",
      "invocation_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Trigger config schema — field-parity with the engine's StateTriggerConfig (engine/src/workers/state/trigger.rs:18-23).",
        "properties": {
          "condition_function_id": {
            "description": "Optional function evaluated per event; only an explicit `false` skips.",
            "type": [
              "string",
              "null"
            ]
          },
          "key": {
            "description": "Only fire for writes to this key (any key when unset).",
            "type": [
              "string",
              "null"
            ]
          },
          "scope": {
            "description": "Only fire for writes in this scope (any scope when unset).",
            "type": [
              "string",
              "null"
            ]
          }
        },
        "title": "StateTriggerSpec",
        "type": "object"
      },
      "metadata": {},
      "name": "state",
      "return_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "StateEventType": {
            "enum": [
              "state:created",
              "state:updated",
              "state:deleted"
            ],
            "type": "string"
          }
        },
        "properties": {
          "event_type": {
            "allOf": [
              {
                "$ref": "#/definitions/StateEventType"
              }
            ],
            "description": "Type of state change"
          },
          "key": {
            "description": "State key",
            "type": "string"
          },
          "new_value": {
            "description": "New value"
          },
          "old_value": {
            "description": "Previous value (null for created events)"
          },
          "scope": {
            "description": "State scope",
            "type": "string"
          },
          "type": {
            "description": "Always \"state\"",
            "type": "string"
          }
        },
        "required": [
          "event_type",
          "key",
          "new_value",
          "scope",
          "type"
        ],
        "title": "StateCallRequest",
        "type": "object"
      }
    }
  ]
}
```