Stream

Build durable streams for real-time data subscriptions.

Durable streams for real-time data subscriptions.

iii-stream

Architecture

Data Flow

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

1. Persists the data via the configured adapter (Redis or KvStore)
2. Publishes a notification to all WebSocket clients subscribed to that stream and group
3. 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

port · number

The port to listen on. Defaults to 3112.

host · string

The host to listen on. Defaults to 0.0.0.0.

auth_function · 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 · 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

redis_url · 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

store_method · string

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

file_path · string

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

Functions

stream::set · function

Sets a value in the stream.

Parameters

stream_name · string · required

The ID of the stream to set the value in.

group_id · string · required

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

item_id · string · required

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

data · any · required

The value to set in the stream.

Returns

old_value · any

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

new_value · any · required

The value now stored in the stream.

stream::get · function

Gets a value from the stream.

Parameters

stream_name · string · required

The ID of the stream to retrieve the value from.

group_id · string · required

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

item_id · string · required

The item ID in the stream to retrieve.

Returns

value · any · required

The value retrieved from the stream.

stream::delete · function

Deletes a value from the stream.

Parameters

stream_name · string · required

The ID of the stream to delete the value from.

group_id · string · required

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

item_id · string · required

The item ID in the stream to delete.

Returns

old_value · any

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

stream::list · function

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

Parameters

stream_name · string · required

The ID of the stream to retrieve the group from.

group_id · string · required

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

Returns

group · any[] · required

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

stream::list_groups · function

List all groups in a stream.

Parameters

stream_name · string · required

The ID of the stream to list groups from.

Returns

groups · string[] · required

An array of group IDs in the stream.

stream::list_all · function

List all streams with their group metadata.

Parameters

This function takes no parameters.

Returns

stream · object[] · required

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

count · number · required

The total number of streams.

stream::send · function

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

Parameters

stream_name · string · required

The ID of the stream to send the event to.

group_id · string · required

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

type · string · required

The event type string delivered to subscribers.

id · string

Optional item ID to associate the event with.

data · any · required

The event payload delivered to subscribers.

Returns

result · null

Returns null on success.

stream::update · function

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

Parameters

stream_name · string · required

The ID of the stream containing the item to update.

group_id · string · required

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

item_id · string · required

The item ID in the stream to update.

ops · 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

old_value · any

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

new_value · any · required

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

errors · 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.

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

Show StreamAuthInput

headers · Record<string, string> · required

The HTTP headers sent with the request.

path · string · required

The request path.

query_params · Record<string, string[]> · required

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

addr · string · required

The remote address (IP) of the request.

Node / TypeScript

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

Python

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

iii.register_function("onAuth", on_auth)

Rust

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

2. Make sure you add the function to the configuration file.

- name: iii-stream
  config:
    auth_function: onAuth

3. 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:

subscription_id · string · required

The subscription ID, used for uniqueness and logging.

stream_name · string · required

The stream name of the subscription.

group_id · string · required

The group ID of the subscription.

id · string

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

context · object

The context generated by the authentication layer.

Node / TypeScript

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: {},
})

Python

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': {}})

Rust

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"),
)?;

stream

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:

stream_name · string · required

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

group_id · string

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

item_id · string

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

condition_function_id · 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:

type · string · required

The event type: create, update, or delete.

timestamp · number · required

Unix timestamp of the event.

streamName · string · required

The stream where the change occurred.

groupId · string · required

The group where the change occurred.

id · string

The item ID that changed.

event · object · required

The event detail object containing type and data fields.

Node / TypeScript

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' },
})

Python

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'},
})

Rust

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

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(),
})

Python

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(),
})

Rust

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

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 }
})

Python

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)

Rust

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

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: {},
})

Python

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': {}})

Rust

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

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 },
})

Python

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'},
})

Rust

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"),
)?;