> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
> Use this file to discover all available pages before exploring further.

# Streaming

> Stream API for real-time data operations with get, set, delete, list, and update operations

## 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

```mermaid theme={null}
graph TD
    S[Stream: "todos"] --> G1[Group: "inbox"]
    S --> G2[Group: "completed"]
    G1 --> I1[Item: "todo-1"]
    G1 --> I2[Item: "todo-2"]
    G2 --> I3[Item: "todo-3"]
    I1 --> D1[Data: {description, createdAt, ...}]
    I2 --> D2[Data: {...}]
    I3 --> D3[Data: {...}]
```

**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

<Info>
  Streams are automatically created on first use. No schema definition required.
</Info>

## Stream Interface

The `IStream<TData>` interface defines operations:

```typescript theme={null}
// 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:

```typescript theme={null}
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:

```typescript theme={null}
// 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:

```typescript theme={null}
// 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:**

```typescript theme={null}
// 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:

```typescript theme={null}
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:**

```typescript theme={null}
// 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:

```typescript theme={null}
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:**

```typescript theme={null}
// 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:

```typescript theme={null}
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:**

```typescript theme={null}
// 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:

```typescript theme={null}
type StreamListGroupsInput = {
  stream_name: string
}

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

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

**Type definition:**

```typescript theme={null}
// 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

```typescript theme={null}
type UpdateOp =
  | UpdateSet        // Set field value
  | UpdateIncrement  // Increment number
  | UpdateDecrement  // Decrement number
  | UpdateRemove     // Remove field
  | UpdateMerge      // Merge object
```

**Type definitions:**

```typescript theme={null}
// 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

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

### Increment/Decrement

```typescript theme={null}
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

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

### Merge Object

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

### Multiple Operations

```typescript theme={null}
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:**

```typescript theme={null}
// 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

```typescript theme={null}
// 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:

```typescript theme={null}
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:**

```typescript theme={null}
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' })
```

<Tip>
  Use **Stream** when you need grouping and listing. Use **State** for simple key-value storage.
</Tip>

## Backend Implementation Notes

When implementing `IStream`:

<Steps>
  <Step title="Handle Nulls">
    Return `null` from `get()` when item doesn't exist. Return `null` from `update()` if item doesn't exist.
  </Step>

  <Step title="Atomic Updates">
    Implement `update()` atomically to prevent race conditions. Use database transactions or compare-and-swap.
  </Step>

  <Step title="Group Listing">
    `listGroups()` should return unique group IDs. Consider caching for large streams.
  </Step>

  <Step title="Serialization">
    Stream data is passed as JSON. Ensure your types serialize correctly.
  </Step>

  <Step title="Error Handling">
    Throw descriptive errors. The Engine will convert them to `InvocationResult` errors.
  </Step>
</Steps>

## Multi-Language Support

<Tabs>
  <Tab title="TypeScript">
    ```typescript theme={null}
    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')
    ```
  </Tab>

  <Tab title="Python">
    ```python theme={null}
    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')
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    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?;
    ```
  </Tab>
</Tabs>

## Next Steps

<CardGroup cols={2}>
  <Card title="Channels" icon="arrows-left-right" href="/concepts/channels">
    Stream large binary data with channels
  </Card>

  <Card title="Functions" icon="function" href="/concepts/functions">
    Call stream operations from functions
  </Card>
</CardGroup>