Streaming

Overview

The Stream API provides a real-time, key-value data store with automatic synchronization across workers. Unlike traditional databases, Stream data is organized into named streams with groups and items, making it ideal for collaborative applications, live dashboards, and multiplayer experiences.

Stream Structure

Hierarchy:

Stream: Named collection (e.g., “todos”, “users”, “messages”)
Group: Logical partition within stream (e.g., “inbox”, “team-1”, “room-42”)
Item: Individual record with unique ID and data

Streams are automatically created on first use. No schema definition required.

Stream Interface

The IStream<TData> interface defines operations:

// Source: packages/node/iii/src/stream.ts:111-118
interface IStream<TData> {
  get(input: StreamGetInput): Promise<TData | null>
  set(input: StreamSetInput): Promise<StreamSetResult<TData> | null>
  delete(input: StreamDeleteInput): Promise<DeleteResult>
  list(input: StreamListInput): Promise<TData[]>
  listGroups(input: StreamListGroupsInput): Promise<string[]>
  update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData> | null>
}

Creating Streams

Custom Stream Implementation

Implement your own backend:

import type { IStream } from 'iii-sdk'

class RedisStream<TData> implements IStream<TData> {
  constructor(private redis: RedisClient, private streamName: string) {}
  
  async get(input: StreamGetInput): Promise<TData | null> {
    const key = `${input.stream_name}:${input.group_id}:${input.item_id}`
    const data = await this.redis.get(key)
    return data ? JSON.parse(data) : null
  }
  
  async set(input: StreamSetInput): Promise<StreamSetResult<TData>> {
    const key = `${input.stream_name}:${input.group_id}:${input.item_id}`
    const old_value = await this.get(input)
    await this.redis.set(key, JSON.stringify(input.data))
    return { old_value, new_value: input.data }
  }
  
  async delete(input: StreamDeleteInput): Promise<DeleteResult> {
    const old_value = await this.get(input)
    const key = `${input.stream_name}:${input.group_id}:${input.item_id}`
    await this.redis.del(key)
    return { old_value }
  }
  
  async list(input: StreamListInput): Promise<TData[]> {
    const pattern = `${input.stream_name}:${input.group_id}:*`
    const keys = await this.redis.keys(pattern)
    const values = await Promise.all(
      keys.map(key => this.redis.get(key))
    )
    return values.map(v => JSON.parse(v))
  }
  
  async listGroups(input: StreamListGroupsInput): Promise<string[]> {
    const pattern = `${input.stream_name}:*`
    const keys = await this.redis.keys(pattern)
    const groups = new Set(
      keys.map(key => key.split(':')[1])
    )
    return Array.from(groups)
  }
  
  async update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData>> {
    const current = await this.get(input)
    if (!current) return null
    
    let updated = { ...current }
    for (const op of input.ops) {
      updated = applyUpdateOp(updated, op)
    }
    
    return await this.set({ ...input, data: updated })
  }
}

const stream = new RedisStream<Todo>(redisClient, 'todos')
iii.createStream('todos', stream)

Registering Stream Functions

The SDK registers stream functions automatically:

// Source: packages/node/iii/src/iii.ts:382-391
createStream = <TData>(streamName: string, stream: IStream<TData>): void => {
  this.registerFunction({ id: `stream::get(${streamName})` }, stream.get.bind(stream))
  this.registerFunction({ id: `stream::set(${streamName})` }, stream.set.bind(stream))
  this.registerFunction({ id: `stream::delete(${streamName})` }, stream.delete.bind(stream))
  this.registerFunction({ id: `stream::list(${streamName})` }, stream.list.bind(stream))
  this.registerFunction(
    { id: `stream::list_groups(${streamName})` },
    stream.listGroups.bind(stream)
  )
}

Functions are named: stream::get(streamName), stream::set(streamName), etc.

Stream Operations

Get Item

Retrieve a single item by ID:

// Input types
type StreamGetInput = {
  stream_name: string
  group_id: string
  item_id: string
}

// Usage
const todo = await streams.get<Todo>('todos', 'inbox', 'todo-123')

if (todo) {
  console.log('Description:', todo.description)
} else {
  console.log('Todo not found')
}

Type definition:

// Source: packages/node/iii/src/stream.ts:27-31
type StreamGetInput = {
  stream_name: string
  group_id: string
  item_id: string
}

Set Item

Create or update an item:

type StreamSetInput = {
  stream_name: string
  group_id: string
  item_id: string
  data: any
}

type StreamSetResult<TData> = {
  old_value?: TData    // Previous value if item existed
  new_value: TData     // Current value after set
}

const result = await streams.set<Todo>('todos', 'inbox', 'todo-123', {
  id: 'todo-123',
  description: 'Buy groceries',
  groupId: 'inbox',
  createdAt: new Date().toISOString(),
  completedAt: null
})

console.log('Previous value:', result.old_value)
console.log('New value:', result.new_value)

Type definitions:

// Source: packages/node/iii/src/stream.ts:33-39, 56-59
type StreamSetInput = {
  stream_name: string
  group_id: string
  item_id: string
  data: any
}

type StreamSetResult<TData> = {
  old_value?: TData
  new_value: TData
}

Delete Item

Remove an item:

type StreamDeleteInput = {
  stream_name: string
  group_id: string
  item_id: string
}

type DeleteResult = {
  old_value?: any  // Value before deletion
}

const result = await streams.delete('todos', 'inbox', 'todo-123')

if (result.old_value) {
  console.log('Deleted todo:', result.old_value.description)
} else {
  console.log('Todo did not exist')
}

Type definitions:

// Source: packages/node/iii/src/stream.ts:41-45, 97-100
type StreamDeleteInput = {
  stream_name: string
  group_id: string
  item_id: string
}

type DeleteResult = {
  old_value?: any
}

List Items in Group

Retrieve all items in a group:

type StreamListInput = {
  stream_name: string
  group_id: string
}

const todos = await streams.list<Todo>('todos', 'inbox')

console.log(`Found ${todos.length} todos in inbox`)
todos.forEach(todo => {
  console.log(`- ${todo.description}`)
})

Type definition:

// Source: packages/node/iii/src/stream.ts:47-50
type StreamListInput = {
  stream_name: string
  group_id: string
}

List Groups

Retrieve all group IDs in a stream:

type StreamListGroupsInput = {
  stream_name: string
}

const groups = await streams.listGroups('todos')

console.log('Todo groups:', groups)  // ["inbox", "completed", "archived"]

Type definition:

// Source: packages/node/iii/src/stream.ts:52-54
type StreamListGroupsInput = {
  stream_name: string
}

Partial Updates

Update specific fields without fetching the entire item:

Update Operations

type UpdateOp =
  | UpdateSet        // Set field value
  | UpdateIncrement  // Increment number
  | UpdateDecrement  // Decrement number
  | UpdateRemove     // Remove field
  | UpdateMerge      // Merge object

Type definitions:

// Source: packages/node/iii/src/stream.ts:66-102
type UpdateSet = {
  type: 'set'
  path: string
  value: any
}

type UpdateIncrement = {
  type: 'increment'
  path: string
  by: number
}

type UpdateDecrement = {
  type: 'decrement'
  path: string
  by: number
}

type UpdateRemove = {
  type: 'remove'
  path: string
}

type UpdateMerge = {
  type: 'merge'
  path: string
  value: any
}

Set Field

await streams.update('todos', 'inbox', 'todo-123', [
  { type: 'set', path: 'description', value: 'Updated description' },
  { type: 'set', path: 'priority', value: 'high' }
])

Increment/Decrement

await streams.update('counters', 'global', 'page-views', [
  { type: 'increment', path: 'count', by: 1 }
])

await streams.update('inventory', 'warehouse-1', 'widget-123', [
  { type: 'decrement', path: 'quantity', by: 5 }
])

Remove Field

await streams.update('todos', 'inbox', 'todo-123', [
  { type: 'remove', path: 'dueDate' }
])

Merge Object

await streams.update('users', 'active', 'user-456', [
  {
    type: 'merge',
    path: 'preferences',
    value: {
      theme: 'dark',
      notifications: true
    }
  }
])

Multiple Operations

const result = await streams.update<Todo>('todos', 'inbox', 'todo-123', [
  { type: 'set', path: 'completedAt', value: new Date().toISOString() },
  { type: 'set', path: 'status', value: 'completed' },
  { type: 'increment', path: 'completionCount', by: 1 }
])

console.log('Old value:', result.old_value)
console.log('New value:', result.new_value)

Update input type:

// Source: packages/node/iii/src/stream.ts:104-109
type StreamUpdateInput = {
  stream_name: string
  group_id: string
  item_id: string
  ops: UpdateOp[]
}

Real-World Example: Todo Application

// Source: packages/node/iii-example/src/index.ts:7-130 (excerpts)
import { useApi } from './hooks'
import { streams } from './stream'
import type { Todo } from './types'

type Todo = {
  id: string
  description: string
  groupId: string
  createdAt: string
  dueDate?: string
  completedAt?: string | null
}

// Create todo
useApi(
  {
    api_path: '/todo',
    http_method: 'POST',
    description: 'Create a new todo'
  },
  async (req, ctx) => {
    ctx.logger.info('Creating new todo', { body: req.body })
    
    const { description, dueDate } = req.body
    const todoId = `todo-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`
    
    if (!description) {
      return { status_code: 400, body: { error: 'Description is required' } }
    }
    
    const newTodo: Todo = {
      id: todoId,
      description,
      groupId: 'inbox',
      createdAt: new Date().toISOString(),
      dueDate: dueDate,
      completedAt: null
    }
    const todo = await streams.set<Todo>('todo', 'inbox', todoId, newTodo)
    
    return { status_code: 201, body: todo }
  }
)

// Update todo
useApi(
  {
    api_path: 'todo/:id',
    http_method: 'PUT',
    description: 'Update a todo'
  },
  async (req, ctx) => {
    const todoId = req.path_params.id
    const existingTodo = todoId ? await streams.get<Todo | null>('todo', 'inbox', todoId) : null
    
    ctx.logger.info('Updating todo', { body: req.body, todoId })
    
    if (!existingTodo) {
      ctx.logger.error('Todo not found')
      return { status_code: 404, body: { error: 'Todo not found' } }
    }
    
    const todo = await streams.set<Todo>('todo', 'inbox', todoId, { ...existingTodo, ...req.body })
    
    ctx.logger.info('Todo updated successfully', { todoId })
    
    return { status_code: 200, body: todo }
  }
)

// Delete todo
useApi(
  {
    api_path: 'todo',
    http_method: 'DELETE',
    description: 'Delete a todo'
  },
  async (req, ctx) => {
    const { todoId } = req.body
    
    ctx.logger.info('Deleting todo', { body: req.body })
    
    if (!todoId) {
      ctx.logger.error('todoId is required')
      return { status_code: 400, body: { error: 'todoId is required' } }
    }
    
    await streams.delete('todo', 'inbox', todoId)
    
    ctx.logger.info('Todo deleted successfully', { todoId })
    
    return { status_code: 200, body: { success: true } }
  }
)

Stream Helper Wrapper

Create a type-safe helper for specific streams:

class TodoStream {
  constructor(private streams: StreamAPI) {}
  
  async create(todo: Omit<Todo, 'id' | 'createdAt'>): Promise<Todo> {
    const id = `todo-${Date.now()}-${Math.random().toString(36).substring(2, 9)}`
    const fullTodo: Todo = {
      ...todo,
      id,
      createdAt: new Date().toISOString(),
      completedAt: null
    }
    
    const result = await this.streams.set<Todo>('todos', todo.groupId, id, fullTodo)
    return result.new_value
  }
  
  async get(groupId: string, id: string): Promise<Todo | null> {
    return this.streams.get<Todo>('todos', groupId, id)
  }
  
  async list(groupId: string): Promise<Todo[]> {
    return this.streams.list<Todo>('todos', groupId)
  }
  
  async complete(groupId: string, id: string): Promise<Todo> {
    const result = await this.streams.update<Todo>('todos', groupId, id, [
      { type: 'set', path: 'completedAt', value: new Date().toISOString() }
    ])
    return result!.new_value
  }
  
  async delete(groupId: string, id: string): Promise<void> {
    await this.streams.delete('todos', groupId, id)
  }
}

const todos = new TodoStream(streams)

// Type-safe API
const todo = await todos.create({
  description: 'Buy groceries',
  groupId: 'inbox',
  dueDate: '2024-03-15'
})

const completed = await todos.complete('inbox', todo.id)

Stream vs State

The III SDK provides both Stream and State APIs:

Aspect	Stream	State
Structure	stream → group → item	scope → key
Hierarchy	3 levels	2 levels
Use Case	Collaborative data, grouped records	Simple key-value storage
Grouping	Built-in with `group_id`	Manual with key prefixes
List API	`list(group)`, `listGroups()`	Depends on implementation

State example:

import { state } from 'iii-sdk'

// Set state
await state.set({ scope: 'todos', key: 'todo-123', data: todoData })

// Get state
const todo = await state.get<Todo>({ scope: 'todos', key: 'todo-123' })

// Delete state
await state.delete({ scope: 'todos', key: 'todo-123' })

Use Stream when you need grouping and listing. Use State for simple key-value storage.

Backend Implementation Notes

When implementing IStream:

Handle Nulls

Return null from get() when item doesn’t exist. Return null from update() if item doesn’t exist.

Atomic Updates

Implement update() atomically to prevent race conditions. Use database transactions or compare-and-swap.

Group Listing

listGroups() should return unique group IDs. Consider caching for large streams.

Serialization

Stream data is passed as JSON. Ensure your types serialize correctly.

Error Handling

Throw descriptive errors. The Engine will convert them to InvocationResult errors.

Multi-Language Support

TypeScript
Python
Rust

import type { IStream } from 'iii-sdk'

const stream: IStream<MyData> = new CustomStream()
iii.createStream('mydata', stream)

const item = await streams.get<MyData>('mydata', 'group1', 'item1')

from iii import IStream

class CustomStream(IStream[MyData]):
    async def get(self, input: StreamGetInput) -> MyData | None:
        # Implementation
        pass

stream = CustomStream()
iii.create_stream('mydata', stream)

item = await streams.get('mydata', 'group1', 'item1')

use iii::{Streams, StreamGetInput};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct MyData { /* fields */ }

let streams = Streams::new(&iii);

let item: Option<MyData> = streams.get(
    "mydata",
    "group1",
    "item1"
).await?;

Get Started

Core Concepts

Guides

Overview

Stream Structure

Stream Interface

Creating Streams

Custom Stream Implementation

Registering Stream Functions

Stream Operations

Get Item

Set Item

Delete Item

List Items in Group

List Groups

Partial Updates

Update Operations

Set Field

Increment/Decrement

Remove Field

Merge Object

Multiple Operations

Real-World Example: Todo Application

Stream Helper Wrapper

Stream vs State

Backend Implementation Notes

Multi-Language Support

Next Steps

Channels

Functions

Build docs developers (and LLMs) love

Get Started

Core Concepts

Guides

Documentation Index

​Overview

​Stream Structure

​Stream Interface

​Creating Streams

​Custom Stream Implementation

​Registering Stream Functions

​Stream Operations

​Get Item

​Set Item

​Delete Item

​List Items in Group

​List Groups

​Partial Updates

​Update Operations

​Set Field

​Increment/Decrement

​Remove Field

​Merge Object

​Multiple Operations

​Real-World Example: Todo Application

​Stream Helper Wrapper

​Stream vs State

​Backend Implementation Notes

​Multi-Language Support

​Next Steps

Channels

Functions

Build docs developers (and LLMs) love

Overview

Stream Structure

Stream Interface

Creating Streams

Custom Stream Implementation

Registering Stream Functions

Stream Operations

Get Item

Set Item

Delete Item

List Items in Group

List Groups

Partial Updates

Update Operations

Set Field

Increment/Decrement

Remove Field

Merge Object

Multiple Operations

Real-World Example: Todo Application

Stream Helper Wrapper

Stream vs State

Backend Implementation Notes

Multi-Language Support

Next Steps