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

> Create custom streaming data sources with the IStream interface

The `IStream` interface allows you to create custom streaming data sources for real-time collaborative data, pub/sub systems, or stateful streams.

## IStream Interface

Implement the `IStream<TData>` interface to create custom stream backends.

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

## Stream Operations

### get()

Retrieve an item from the stream.

```typescript theme={null}
get(input: StreamGetInput): Promise<TData | null>
```

<ParamField path="input" type="StreamGetInput" required>
  <Expandable title="StreamGetInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>

    <ParamField path="group_id" type="string" required>
      Group/namespace ID
    </ParamField>

    <ParamField path="item_id" type="string" required>
      Unique item ID
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="data" type="TData | null">
  The item data, or `null` if not found
</ResponseField>

### set()

Set an item in the stream.

```typescript theme={null}
set(input: StreamSetInput): Promise<StreamSetResult<TData> | null>
```

<ParamField path="input" type="StreamSetInput" required>
  <Expandable title="StreamSetInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>

    <ParamField path="group_id" type="string" required>
      Group/namespace ID
    </ParamField>

    <ParamField path="item_id" type="string" required>
      Unique item ID
    </ParamField>

    <ParamField path="data" type="any" required>
      Data to set
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="result" type="StreamSetResult<TData> | null">
  Result containing old and new values

  <Expandable title="StreamSetResult properties">
    <ResponseField name="old_value" type="TData">
      Previous value (if any)
    </ResponseField>

    <ResponseField name="new_value" type="TData" required>
      New value that was set
    </ResponseField>
  </Expandable>
</ResponseField>

### delete()

Delete an item from the stream.

```typescript theme={null}
delete(input: StreamDeleteInput): Promise<DeleteResult>
```

<ParamField path="input" type="StreamDeleteInput" required>
  <Expandable title="StreamDeleteInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>

    <ParamField path="group_id" type="string" required>
      Group/namespace ID
    </ParamField>

    <ParamField path="item_id" type="string" required>
      Unique item ID
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="result" type="DeleteResult">
  <Expandable title="DeleteResult properties">
    <ResponseField name="old_value" type="any">
      The deleted value (if any)
    </ResponseField>
  </Expandable>
</ResponseField>

### list()

List all items in a group.

```typescript theme={null}
list(input: StreamListInput): Promise<TData[]>
```

<ParamField path="input" type="StreamListInput" required>
  <Expandable title="StreamListInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>

    <ParamField path="group_id" type="string" required>
      Group/namespace ID
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="items" type="TData[]">
  Array of all items in the group
</ResponseField>

### listGroups()

List all groups in the stream.

```typescript theme={null}
listGroups(input: StreamListGroupsInput): Promise<string[]>
```

<ParamField path="input" type="StreamListGroupsInput" required>
  <Expandable title="StreamListGroupsInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="groups" type="string[]">
  Array of group IDs
</ResponseField>

### update()

Perform atomic updates on an item.

```typescript theme={null}
update(input: StreamUpdateInput): Promise<StreamUpdateResult<TData> | null>
```

<ParamField path="input" type="StreamUpdateInput" required>
  <Expandable title="StreamUpdateInput properties">
    <ParamField path="stream_name" type="string" required>
      Name of the stream
    </ParamField>

    <ParamField path="group_id" type="string" required>
      Group/namespace ID
    </ParamField>

    <ParamField path="item_id" type="string" required>
      Unique item ID
    </ParamField>

    <ParamField path="ops" type="UpdateOp[]" required>
      Array of update operations

      <Expandable title="Update operations">
        **Set a value:**

        ```typescript theme={null}
        { type: 'set', path: 'field.nested', value: any }
        ```

        **Increment a number:**

        ```typescript theme={null}
        { type: 'increment', path: 'counter', by: 1 }
        ```

        **Decrement a number:**

        ```typescript theme={null}
        { type: 'decrement', path: 'counter', by: 1 }
        ```

        **Remove a field:**

        ```typescript theme={null}
        { type: 'remove', path: 'field' }
        ```

        **Merge an object:**

        ```typescript theme={null}
        { type: 'merge', path: 'object', value: { key: 'value' } }
        ```
      </Expandable>
    </ParamField>
  </Expandable>
</ParamField>

<ResponseField name="result" type="StreamUpdateResult<TData> | null">
  <Expandable title="StreamUpdateResult properties">
    <ResponseField name="old_value" type="TData">
      Value before update
    </ResponseField>

    <ResponseField name="new_value" type="TData" required>
      Value after update
    </ResponseField>
  </Expandable>
</ResponseField>

## createStream()

Register a stream implementation with the III SDK.

```typescript theme={null}
iii.createStream<TData>(streamName, stream)
```

<ParamField path="streamName" type="string" required>
  Unique stream name
</ParamField>

<ParamField path="stream" type="IStream<TData>" required>
  Stream implementation
</ParamField>

This automatically registers the following functions:

* `stream::get(streamName)`
* `stream::set(streamName)`
* `stream::delete(streamName)`
* `stream::list(streamName)`
* `stream::list_groups(streamName)`

## Example: In-Memory Stream

```typescript theme={null}
import { init, type IStream, type StreamGetInput, type StreamSetInput, type StreamDeleteInput, type StreamListInput, type StreamListGroupsInput } from 'iii-sdk'

const iii = init('ws://localhost:49199')

interface TodoItem {
  id: string
  title: string
  completed: boolean
  created_at: string
}

class InMemoryStream implements IStream<TodoItem> {
  private data = new Map<string, Map<string, TodoItem>>()
  
  async get(input: StreamGetInput): Promise<TodoItem | null> {
    const group = this.data.get(input.group_id)
    return group?.get(input.item_id) ?? null
  }
  
  async set(input: StreamSetInput) {
    let group = this.data.get(input.group_id)
    if (!group) {
      group = new Map()
      this.data.set(input.group_id, group)
    }
    
    const old_value = group.get(input.item_id)
    group.set(input.item_id, input.data)
    
    return {
      old_value,
      new_value: input.data
    }
  }
  
  async delete(input: StreamDeleteInput) {
    const group = this.data.get(input.group_id)
    const old_value = group?.get(input.item_id)
    group?.delete(input.item_id)
    
    return { old_value }
  }
  
  async list(input: StreamListInput): Promise<TodoItem[]> {
    const group = this.data.get(input.group_id)
    return group ? Array.from(group.values()) : []
  }
  
  async listGroups(input: StreamListGroupsInput): Promise<string[]> {
    return Array.from(this.data.keys())
  }
  
  async update(input: any) {
    // Implement atomic updates if needed
    return null
  }
}

// Register the stream
iii.createStream('todos', new InMemoryStream())

// Now you can access it via functions
const setResult = await iii.call('stream::set(todos)', {
  stream_name: 'todos',
  group_id: 'user-123',
  item_id: 'todo-1',
  data: {
    id: 'todo-1',
    title: 'Buy milk',
    completed: false,
    created_at: new Date().toISOString()
  }
})

const todo = await iii.call('stream::get(todos)', {
  stream_name: 'todos',
  group_id: 'user-123',
  item_id: 'todo-1'
})

const allTodos = await iii.call('stream::list(todos)', {
  stream_name: 'todos',
  group_id: 'user-123'
})
```

## Example: Redis Stream

```typescript theme={null}
import { init, type IStream } from 'iii-sdk'
import { createClient } from 'redis'

const iii = init('ws://localhost:49199')
const redis = createClient()
await redis.connect()

class RedisStream<TData> implements IStream<TData> {
  constructor(private redis: ReturnType<typeof createClient>) {}
  
  private getKey(group_id: string, item_id: string): string {
    return `stream:${group_id}:${item_id}`
  }
  
  async get(input: any): Promise<TData | null> {
    const key = this.getKey(input.group_id, input.item_id)
    const data = await this.redis.get(key)
    return data ? JSON.parse(data) : null
  }
  
  async set(input: any) {
    const key = this.getKey(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: any) {
    const key = this.getKey(input.group_id, input.item_id)
    const old_value = await this.get(input)
    
    await this.redis.del(key)
    
    return { old_value }
  }
  
  async list(input: any): Promise<TData[]> {
    const pattern = `stream:${input.group_id}:*`
    const keys = await this.redis.keys(pattern)
    
    const items: TData[] = []
    for (const key of keys) {
      const data = await this.redis.get(key)
      if (data) {
        items.push(JSON.parse(data))
      }
    }
    
    return items
  }
  
  async listGroups(input: any): Promise<string[]> {
    const pattern = 'stream:*'
    const keys = await this.redis.keys(pattern)
    
    const groups = new Set<string>()
    for (const key of keys) {
      const parts = key.split(':')
      if (parts.length >= 2) {
        groups.add(parts[1])
      }
    }
    
    return Array.from(groups)
  }
  
  async update(input: any) {
    // Implement using Redis transactions
    return null
  }
}

iii.createStream('sessions', new RedisStream(redis))
```

## Example: Database Stream

```typescript theme={null}
import { init, type IStream } from 'iii-sdk'
import { PrismaClient } from '@prisma/client'

const iii = init('ws://localhost:49199')
const prisma = new PrismaClient()

interface Document {
  id: string
  title: string
  content: string
  updated_at: string
}

class DatabaseStream implements IStream<Document> {
  async get(input: any): Promise<Document | null> {
    const doc = await prisma.document.findFirst({
      where: {
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return doc ? {
      id: doc.id,
      title: doc.title,
      content: doc.content,
      updated_at: doc.updated_at.toISOString()
    } : null
  }
  
  async set(input: any) {
    const old_value = await this.get(input)
    
    const doc = await prisma.document.upsert({
      where: {
        group_id_id: {
          group_id: input.group_id,
          id: input.item_id
        }
      },
      update: input.data,
      create: {
        ...input.data,
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return {
      old_value,
      new_value: {
        id: doc.id,
        title: doc.title,
        content: doc.content,
        updated_at: doc.updated_at.toISOString()
      }
    }
  }
  
  async delete(input: any) {
    const old_value = await this.get(input)
    
    await prisma.document.deleteMany({
      where: {
        group_id: input.group_id,
        id: input.item_id
      }
    })
    
    return { old_value }
  }
  
  async list(input: any): Promise<Document[]> {
    const docs = await prisma.document.findMany({
      where: { group_id: input.group_id }
    })
    
    return docs.map(doc => ({
      id: doc.id,
      title: doc.title,
      content: doc.content,
      updated_at: doc.updated_at.toISOString()
    }))
  }
  
  async listGroups(): Promise<string[]> {
    const groups = await prisma.document.findMany({
      select: { group_id: true },
      distinct: ['group_id']
    })
    
    return groups.map(g => g.group_id)
  }
  
  async update(input: any) {
    // Implement using Prisma transactions
    return null
  }
}

iii.createStream('documents', new DatabaseStream())
```

## Use Cases

<AccordionGroup>
  <Accordion title="Real-time Collaboration">
    Implement operational transformation or CRDTs for collaborative editing:

    ```typescript theme={null}
    class CollaborativeDocStream implements IStream<Document> {
      // Implement get, set, delete, list, listGroups
      
      async update(input: any) {
        // Apply operational transforms
        const doc = await this.get(input)
        const updated = applyOperations(doc, input.ops)
        return await this.set({ ...input, data: updated })
      }
    }
    ```
  </Accordion>

  <Accordion title="Event Sourcing">
    Store events and compute state from event history:

    ```typescript theme={null}
    class EventSourcedStream implements IStream<State> {
      async set(input: any) {
        // Append event
        await appendEvent(input.data)
        
        // Recompute state
        const events = await getEvents(input.group_id)
        const new_value = computeState(events)
        
        return { old_value: null, new_value }
      }
    }
    ```
  </Accordion>

  <Accordion title="Caching Layer">
    Add caching on top of another data source:

    ```typescript theme={null}
    class CachedStream implements IStream<Data> {
      constructor(
        private backend: IStream<Data>,
        private cache: Map<string, Data>
      ) {}
      
      async get(input: any) {
        const key = `${input.group_id}:${input.item_id}`
        
        if (this.cache.has(key)) {
          return this.cache.get(key)!
        }
        
        const data = await this.backend.get(input)
        if (data) this.cache.set(key, data)
        
        return data
      }
      
      // Implement other methods with cache invalidation
    }
    ```
  </Accordion>
</AccordionGroup>