Stream

Durable streams for real-time data subscriptions.

iii-stream

Architecture

Data Flow

When a worker triggers stream::set, the engine:

Persists the data via the configured adapter (Redis or KvStore)
Publishes a notification to all WebSocket clients subscribed to that stream and group
Evaluates registered stream triggers and fires matching handlers

A single stream::set handles persistence, real-time delivery, and reactive logic in one operation.

Groups

Streams organize data hierarchically: stream_name > group_id > item_id.

stream_name identifies the top-level stream (e.g. chat, presence, dashboard)
group_id partitions data within a stream (e.g. room-1, team-alpha)
item_id uniquely identifies a record within a group (e.g. user-123, msg-456)

Clients subscribe at the group level by connecting to ws://host:port/stream/{stream_name}/{group_id}/. They receive all item-level changes within that group.

Sample Configuration

- name: iii-stream
  config:
    port: ${STREAM_PORT:3112}
    host: 0.0.0.0
    adapter:
      name: redis
      config:
        redis_url: ${REDIS_URL:redis://localhost:6379}

Configuration

number

The port to listen on. Defaults to 3112.

string

The host to listen on. Defaults to 0.0.0.0.

string

The authentication function to use. It’s a path to a function that will be used to authenticate the client. You can register the function using the iii SDK and then use the path to the function here.

Adapter

The adapter to use. It’s the adapter that will be used to store the streams. You can register the adapter using the iii SDK and then use the path to the adapter here.

Adapters

redis

Uses Redis as the backend for the streams. Stores stream data in Redis and leverages Redis Pub/Sub for real-time event delivery.

name: redis
config:
  redis_url: ${REDIS_URL:redis://localhost:6379}

Configuration

string

The URL of the Redis instance to use.

kv

Built-in key-value store. Supports in-memory or file-based persistence. No external dependencies required.

name: kv
config:
  store_method: file_based
  file_path: ./data/streams_store.db

Configuration

string

Storage method. Options: in_memory (lost on restart) or file_based (persisted to disk).

string

Directory path for file-based storage. Each stream is stored as a separate file.

Functions

function

Sets a value in the stream.

Parameters

string

required

The ID of the stream to set the value in.

string

required

The group ID of the stream to set the value in.

string

required

The item ID of the stream to set the value in.

any

required

The value to set in the stream.

Returns

any

The previous value, or null if the item was newly created.

any

required

The value now stored in the stream.

function

Gets a value from the stream.

Parameters

string

required

The ID of the stream to retrieve the value from.

string

required

The group ID in the stream to retrieve the value from.

string

required

The item ID in the stream to retrieve.

Returns

any

required

The value retrieved from the stream.

function

Deletes a value from the stream.

Parameters

string

required

The ID of the stream to delete the value from.

string

required

The group ID in the stream to delete the value from.

string

required

The item ID in the stream to delete.

Returns

any

The value that was deleted, or null if the item did not exist.

function

Retrieves a group from the stream. This function will return all the items in the group.

Parameters

string

required

The ID of the stream to retrieve the group from.

string

required

The group ID in the stream to retrieve the group from.

Returns

any[]

required

The group retrieved from the stream. It’s an array of items in the group.

function

List all groups in a stream.

Parameters

string

required

The ID of the stream to list groups from.

Returns

string[]

required

An array of group IDs in the stream.

function

List all streams with their group metadata.

Parameters

This function takes no parameters.

Returns

object[]

required

An array of stream metadata objects. Each object has an id (string) and a groups (string[]) field.

number

required

The total number of streams.

function

Send a custom event to all subscribers of a stream group.

Parameters

string

required

The ID of the stream to send the event to.

string

required

The group ID in the stream to send the event to.

string

required

The event type string delivered to subscribers.

string

Optional item ID to associate the event with.

any

required

The event payload delivered to subscribers.

Returns

null

Returns null on success.

function

Atomically update an item in the stream using a list of operations.

Parameters

string

required

The ID of the stream containing the item to update.

string

required

The group ID in the stream containing the item to update.

string

required

The item ID in the stream to update.

UpdateOp[]

required

The list of atomic operations to apply. Each operation is a tagged object with a type field (set, merge, increment, decrement, append, or remove) and associated fields (path, value, by). For set / increment / decrement / append / remove, paths are first-level field names. For merge, path accepts either a single string (legacy / first-level field) or an array of literal segments for nested merge — see the state worker docs for the full contract and validation rules. Use path: "" (or omit path) to target the root value.

Returns

any

The previous value, or null if the item was newly created.

any

required

The value now stored in the stream after applying the operations.

UpdateOpError[]

Per-op validation errors. Field is omitted when empty. Each entry has op_index, code, message, and an optional doc_url.

Error codes

stream::update uses the same update engine as state::update. Each op may add an entry to the response errors array. Operations are best-effort: successfully applied ops still reflect in new_value, and failed ops are skipped.

Code	Triggered when	Fix
`set.target_not_object`	`set` tried to write a field while the current value is not an object	Set the root to an object first, or use `path: ""` to replace the root.
`append.target_not_object`	`append` used a field path while the current value is not an object	Set the root to an object first, or append at `path: ""`.
`append.type_mismatch`	`append` targeted an incompatible existing value, such as appending to a number or appending a non-string to a string	Match the appended value to the existing field type, or initialize the field to an array, string, or null.
`increment.target_not_object`	`increment` used a field path while the current value is not an object	Set the root to an object first.
`increment.not_number`	`increment` targeted an existing field that is not a number	Initialize the field as a number first, for example with `set` to `0`.
`decrement.target_not_object`	`decrement` used a field path while the current value is not an object	Set the root to an object first.
`decrement.not_number`	`decrement` targeted an existing field that is not a number	Initialize the field as a number first, for example with `set` to `0`.
`remove.target_not_object`	`remove` used a field path while the current value is not an object	Set the root to an object first. Removing a missing field from an object remains silent.
`<op>.path.proto_polluted`	A path segment is `__proto__`, `constructor`, or `prototype`	Use a different field name.
`<op>.path.segment_too_long`	A path segment is longer than 256 bytes	Shorten the field name or merge path segment.
`merge.path.too_deep`	A nested merge path has more than 32 segments	Reduce the nested path depth.
`merge.path.empty_segment`	A nested merge path array contains an empty segment	Remove the empty segment.
`merge.value.not_an_object`	`merge` value is not a JSON object	Pass an object as the merge value.
`merge.value.too_deep`	`merge` value has JSON nesting deeper than 16 levels	Flatten the value.
`merge.value.too_many_keys`	`merge` value has more than 1024 top-level keys	Split the write into smaller updates.
`merge.value.proto_polluted`	A top-level key in the merge value is `__proto__`, `constructor`, or `prototype`	Use a different key name.

Each error includes op_index, code, and message; doc_url is optional.

{
  "old_value": { "count": 1 },
  "new_value": { "count": 1 },
  "errors": [
    {
      "op_index": 0,
      "code": "append.type_mismatch",
      "message": "Cannot append at path 'count': target is number, expected array, string, null, or missing field.",
      "doc_url": "https://iii.dev/docs/workers/iii-state#error-codes"
    }
  ]
}

Authentication

It’s possible to implement a function to handle authentication.

Define a function to handle the authentication. It received one single argument with the request data.

Show StreamAuthInput

Record<string, string>

required

The HTTP headers sent with the request.

string

required

The request path.

Record<string, string[]>

required

Query parameters in the request, as a map from key to array of string values.

string

required

The remote address (IP) of the request.

Node / TypeScript
Python
Rust

iii.registerFunction('onAuth', (input) => ({
  context: { name: 'John Doe' },
}))

def on_auth(input):
    return {'context': {'name': 'John Doe'}}

iii.register_function("onAuth", on_auth)

iii.register_function((RegisterFunctionMessage::with_id("onAuth".into()), |_input| async move {
    Ok(json!({ "context": { "name": "John Doe" } }))
}));

Make sure you add the function to the configuration file.

- name: iii-stream
  config:
    auth_function: onAuth

Now whenever someone opens a websocket connection, the function onAuth will be called with the request data.

Trigger Types

This worker adds three trigger types: stream (item changes), stream:join (WebSocket connect), and stream:leave (WebSocket disconnect).

stream:join and stream:leave

Fire when a client connects or disconnects via WebSocket. Both trigger types deliver the same payload to the handler:

string

required

The subscription ID, used for uniqueness and logging.

string

required

The stream name of the subscription.

string

required

The group ID of the subscription.

string

The item ID of the subscription, if provided by the client.

object

The context generated by the authentication layer.

Node / TypeScript
Python
Rust

import { StreamJoinLeaveEvent } from 'iii-sdk/stream'

const fn = iii.registerFunction('onJoin', (input: StreamJoinLeaveEvent) => {
  console.log(`Joined ${input.stream_name}/${input.group_id}`, input.context)
  return {}
})

iii.registerTrigger({
  type: 'stream:join',
  function_id: fn.id,
  config: {},
})

from iii import StreamJoinLeaveEvent

def on_join(input: StreamJoinLeaveEvent):
    print(f"Joined {input.stream_name}/{input.group_id}", input.context)
    return {}

iii.register_function("onJoin", on_join)
iii.register_trigger({'type': 'stream:join', 'function_id': 'onJoin', 'config': {}})

use iii_sdk::{IIITrigger, StreamJoinLeaveCallRequest, StreamJoinLeaveTriggerConfig};

iii.register_function((RegisterFunctionMessage::with_id("onJoin".into()), |input| async move {
    let event: StreamJoinLeaveCallRequest = serde_json::from_value(input)?;
    println!("Joined {}/{}", event.stream_name, event.group_id);
    Ok(json!({}))
});

iii.register_trigger(
    IIITrigger::StreamJoin(StreamJoinLeaveTriggerConfig::new()).for_function("onJoin"),
)?;

Fires when an item changes in the stream (via stream::set, stream::update, or stream::delete). Register with a config object to filter which stream, group, or item triggers the handler:

string

required

The stream name to watch. Only changes on this stream fire the handler.

string

If set, only changes within this group fire the handler.

string

If set, only changes to this specific item fire the handler.

string

Function ID for conditional execution. The engine invokes it with the event payload; if it returns false, the handler function is not called.

The handler receives a payload with the following shape:

string

required

The event type: create, update, or delete.

number

required

Unix timestamp of the event.

string

required

The stream where the change occurred.

string

required

The group where the change occurred.

string

The item ID that changed.

object

required

The event detail object containing type and data fields.

Node / TypeScript
Python
Rust

import { StreamChangeEvent } from 'iii-sdk/stream'

const fn = iii.registerFunction('onChatMessage', (input: StreamChangeEvent) => {
  console.log(`[${input.event.type}] ${input.streamName}/${input.groupId}/${input.id}`, input.event.data)
  return {}
})

iii.registerTrigger({
  type: 'stream',
  function_id: fn.id,
  config: { stream_name: 'chat' },
})

from iii import StreamChangeEvent

def on_chat_message(input: StreamChangeEvent):
    print(f"[{input.event.type}] {input.streamName}/{input.groupId}/{input.id}", input.event.data)
    return {}

iii.register_function("onChatMessage", on_chat_message)
iii.register_trigger({
    'type': 'stream',
    'function_id': 'onChatMessage',
    'config': {'stream_name': 'chat'},
})

use iii_sdk::{IIITrigger, StreamCallRequest, StreamEventType, StreamTriggerConfig};

iii.register_function((RegisterFunctionMessage::with_id("onChatMessage".into()), |input| async move {
    let event: StreamCallRequest = serde_json::from_value(input)?;
    println!("[{:?}] {}/{}/{}: {:?}", event.event.event_type, event.stream_name, event.group_id, event.id.as_deref().unwrap_or(""), event.event.data);
    Ok(json!({}))
});

iii.register_trigger(
    IIITrigger::Stream(StreamTriggerConfig::new().stream_name("chat")).for_function("onChatMessage"),
)?;

Usage Example: Real-Time Presence

Streams organize data by stream_name, group_id, and item_id. Use for live presence, collaborative docs, or dashboards:

Node / TypeScript
Python
Rust

import { registerWorker, TriggerAction } from 'iii-sdk'

const iii = registerWorker('ws://localhost:49134')

iii.trigger({
  function_id: 'stream::set',
  payload: {
    stream_name: 'presence',
    group_id: 'room-1',
    item_id: 'user-123',
    data: { name: 'Alice', online: true, lastSeen: new Date().toISOString() },
  },
  action: TriggerAction.Void(),
})

const user = await iii.trigger({
  function_id: 'stream::get',
  payload: {
    stream_name: 'presence',
    group_id: 'room-1',
    item_id: 'user-123',
  },
})

const roomMembers = await iii.trigger({
  function_id: 'stream::list',
  payload: {
    stream_name: 'presence',
    group_id: 'room-1',
  },
})

iii.trigger({
  function_id: 'stream::delete',
  payload: {
    stream_name: 'presence',
    group_id: 'room-1',
    item_id: 'user-123',
  },
  action: TriggerAction.Void(),
})

from iii import register_worker, TriggerAction

iii = register_worker('ws://localhost:49134')

iii.trigger({
    'function_id': 'stream::set',
    'payload': {
        'stream_name': 'presence',
        'group_id': 'room-1',
        'item_id': 'user-123',
        'data': {'name': 'Alice', 'online': True, 'lastSeen': '2026-01-01T00:00:00Z'},
    },
    'action': TriggerAction.Void(),
})

user = iii.trigger({
    'function_id': 'stream::get',
    'payload': {
        'stream_name': 'presence',
        'group_id': 'room-1',
        'item_id': 'user-123',
    },
})

room_members = iii.trigger({
    'function_id': 'stream::list',
    'payload': {
        'stream_name': 'presence',
        'group_id': 'room-1',
    },
})

iii.trigger({
    'function_id': 'stream::delete',
    'payload': {
        'stream_name': 'presence',
        'group_id': 'room-1',
        'item_id': 'user-123',
    },
    'action': TriggerAction.Void(),
})

use iii_sdk::{register_worker, InitOptions, TriggerRequest, TriggerAction};
use serde_json::json;

let iii = register_worker("ws://localhost:49134", InitOptions::default());

iii.trigger(TriggerRequest::new("stream::set", json!({
    "stream_name": "presence",
    "group_id": "room-1",
    "item_id": "user-123",
    "data": { "name": "Alice", "online": true }
})).action(TriggerAction::void())).await?;

let user = iii.trigger(TriggerRequest::new("stream::get", json!({
    "stream_name": "presence",
    "group_id": "room-1",
    "item_id": "user-123"
}))).await?;

let room_members = iii.trigger(TriggerRequest::new("stream::list", json!({
    "stream_name": "presence",
    "group_id": "room-1"
}))).await?;

iii.trigger(TriggerRequest::new("stream::delete", json!({
    "stream_name": "presence",
    "group_id": "room-1",
    "item_id": "user-123"
})).action(TriggerAction::void())).await?;

Clients connect via WebSocket to ws://host:3112/stream/presence/room-1/ and receive real-time updates when items change.

Usage Example: Join with Auth Context

Configure the stream worker with an auth function:

- name: iii-stream
  config:
    port: 3112
    host: 0.0.0.0
    auth_function: stream::auth
    adapter:
      name: kv
      config:
        store_method: file_based
        file_path: ./data/stream_store

Register the auth function. Clients may send the token via Authorization: Bearer <token> (Node.js) or Sec-WebSocket-Protocol: Authorization,<token> (browser stream-client):

Node / TypeScript
Python
Rust

iii.registerFunction('stream::auth', (input) => {
  const auth = input.headers?.['authorization']?.replace(/^Bearer\s+/i, '')
  const proto = input.headers?.['sec-websocket-protocol']
  const token = auth ?? (proto?.startsWith('Authorization,') ? proto.slice(13) : null)
  return token ? { context: { userId: 'user-from-token' } } : { context: null }
})

def stream_auth(input):
    auth = input.get('headers', {}).get('authorization', '')
    token = auth.replace('Bearer ', '', 1) if auth.startswith('Bearer ') else None
    if token:
        return {'context': {'userId': 'user-from-token'}}
    return {'context': None}

iii.register_function("stream::auth", stream_auth)

iii.register_function((RegisterFunctionMessage::with_id("stream::auth".into()), |input| async move {
    let headers = input.get("headers").and_then(|h| h.as_object()));

    let token = headers
        .and_then(|h| h.get("authorization"))
        .and_then(|v| v.as_str())
        .and_then(|s| s.strip_prefix("Bearer "));

    match token {
        Some(_) => Ok(json!({ "context": { "userId": "user-from-token" } })),
        None => Ok(json!({ "context": null })),
    }
});

Join/leave triggers receive the auth context:

Node / TypeScript
Python
Rust

import { StreamJoinLeaveEvent } from 'iii-sdk/stream'

const fn = iii.registerFunction('onJoin', (input: StreamJoinLeaveEvent) => {
  if (input.context?.userId) {
    console.log(`User ${input.context.userId} joined ${input.stream_name}/${input.group_id}/${input.id}`)
  }
  return {}
})

iii.registerTrigger({
  type: 'stream:join',
  function_id: fn.id,
  config: {},
})

from iii import StreamJoinLeaveEvent

def on_join(input: StreamJoinLeaveEvent):
    if input.context and input.context.get('userId'):
        print(f"User {input.context['userId']} joined {input.stream_name}/{input.group_id}/{input.id}")
    return {}

iii.register_function("onJoin", on_join)
iii.register_trigger({'type': 'stream:join', 'function_id': 'onJoin', 'config': {}})

use iii_sdk::{IIITrigger, StreamJoinLeaveCallRequest, StreamJoinLeaveTriggerConfig};

iii.register_function((RegisterFunctionMessage::with_id("onJoin".into()), |input| async move {
    let event: StreamJoinLeaveCallRequest = serde_json::from_value(input)?;
    if let Some(user_id) = event.context.as_ref().and_then(|c| c.get("userId")).and_then(|u| u.as_str()) {
        println!("User {} joined {}/{}/{}", user_id, event.stream_name, event.group_id, event.id.as_deref().unwrap_or(""));
    }
    Ok(json!({}))
});

iii.register_trigger(
    IIITrigger::StreamJoin(StreamJoinLeaveTriggerConfig::new()).for_function("onJoin"),
)?;

Usage Example: Conditional Join

Node / TypeScript
Python
Rust

import { StreamJoinLeaveEvent } from 'iii-sdk/stream'

const conditionFn = iii.registerFunction(
  { id: 'conditions::requireContext' },
  async (input: StreamJoinLeaveEvent) => input.context?.userId != null,
)

const fn = iii.registerFunction('onJoin', (input: StreamJoinLeaveEvent) => {
  console.log('User joined:', input.context?.userId, input.stream_name)
  return {}
})

iii.registerTrigger({
  type: 'stream:join',
  function_id: fn.id,
  config: { condition_function_id: conditionFn.id },
})

from iii import StreamJoinLeaveEvent

def require_context(input: StreamJoinLeaveEvent):
    return input.context is not None and input.context.get('userId') is not None

iii.register_function("conditions::requireContext", require_context)

def on_join(input: StreamJoinLeaveEvent):
    print('User joined:', input.context.get('userId') if input.context else None, input.stream_name)
    return {}

iii.register_function("onJoin", on_join)
iii.register_trigger({
    'type': 'stream:join',
    'function_id': 'onJoin',
    'config': {'condition_function_id': 'conditions::requireContext'},
})

use iii_sdk::{IIITrigger, StreamJoinLeaveCallRequest, StreamJoinLeaveTriggerConfig};

iii.register_function((RegisterFunctionMessage::with_id("conditions::requireContext".into()), |input| async move {
    let event: StreamJoinLeaveCallRequest = serde_json::from_value(input)?;
    let has_user = event.context
        .as_ref()
        .and_then(|c| c.get("userId"))
        .is_some();
    Ok(json!(has_user))
});

iii.register_function((RegisterFunctionMessage::with_id("onJoin".into()), |input| async move {
    let event: StreamJoinLeaveCallRequest = serde_json::from_value(input)?;
    let user_id = event.context.as_ref().and_then(|c| c.get("userId")).and_then(|u| u.as_str()).unwrap_or("");
    println!("User joined: {} {}", user_id, event.stream_name);
    Ok(json!({}))
});

iii.register_trigger(
    IIITrigger::StreamJoin(StreamJoinLeaveTriggerConfig::new().condition("conditions::requireContext"))
        .for_function("onJoin"),
)?;

Getting Started

Primitives & Concepts

How-to Guides

Architecture

Workers

Console

Advanced

Examples

Stream

Architecture

Data Flow

Groups

Sample Configuration

Configuration

Adapters

redis

Configuration

kv

Configuration

Functions

Error codes

Authentication

Trigger Types

stream:join and stream:leave

stream

Usage Example: Real-Time Presence

Usage Example: Join with Auth Context

Usage Example: Conditional Join

​Architecture

​Data Flow

​Groups

​Sample Configuration

​Configuration

​Adapters

​redis

​Configuration

​kv

​Configuration

​Functions

​Error codes

​Authentication

​Trigger Types

​stream:join and stream:leave

​stream

​Usage Example: Real-Time Presence

​Usage Example: Join with Auth Context

​Usage Example: Conditional Join

Architecture

Data Flow

Groups

Sample Configuration

Configuration

Adapters

redis

Configuration

kv

Configuration

Functions

Error codes

Authentication

Trigger Types

stream:join and stream:leave

stream

Usage Example: Real-Time Presence

Usage Example: Join with Auth Context

Usage Example: Conditional Join