# 1.4 Socket.IO Real-time Layer

## Architectural Rationale

The system separates request ingestion (HTTP) from result delivery (WebSocket) to avoid keeping long-running HTTP connections alive during Ollama inference. Socket.IO is attached to the same Fastify listener (port 3000) via `SocketIOModule.attach(APP)`, sharing the TLS certificate and CORS configuration while maintaining distinct event namespaces. A Redis adapter is configured for horizontal scaling, enabling cross-server room broadcast without session affinity requirements.

```mermaid
flowchart TB
    subgraph Clients["Clients"]
        Dashboard[Dashboard UI]
        RESTClient[REST Client]
        MCPClient[MCP Client]
    end

    subgraph Server["@CKIR.IO/VISIONS Server"]
        Gateway[Fastify Gateway]
        IOMod["Socket.IO Module"]
        IOServer["Socket.IO Server"]
        Adapter[Redis Adapter]
    end

    subgraph Events["Handlers"]
        Conn[Connection Handler]
        Room[Room Handler]
        Vision[Vision Event Handler]
    end

    subgraph Services["Services"]
        SocketSvc[SocketService]
    end

    subgraph Infra["Infrastructure"]
        Redis[KeyDB/Redis Pub/Sub]
    end

    Dashboard -->|HTTP Upgrade| Gateway
    RESTClient -->|HTTP Upgrade| Gateway
    MCPClient -->|HTTP Upgrade| Gateway

    Gateway --> IOMod --> IOServer
    IOServer --> Adapter --> Redis
    IOServer --> Conn --> Room --> Vision --> SocketSvc --> IOServer

    IOServer -->|Emit Events| Dashboard
    IOServer -->|Emit Events| RESTClient
    IOServer -->|Emit Events| MCPClient
```

## Server Configuration

```typescript
// Conceptual configuration via @ehildt/nestjs-socket.io
SocketIOModule.forRoot({
  cors: {
    origin: process.env.CORS_ORIGIN,
    credentials: true,
  },
  transports: ['websocket', 'polling', 'webtransport'],
  adapter: createAdapter(pubClient, subClient), // Redis adapter
});
```

| Parameter | Default | Semantics |
|-----------|---------|-----------|
| `transports` | `websocket,polling,webtransport` | Fallback chain; polling ensures connectivity through restrictive proxies |
| `cors.origin` | `*` | Allowlist; overridden in production by `CORS_ORIGIN` |
| `cors.credentials` | `true` | Required for Cookie-based auth (future-proofing) |
| `pingInterval` | `25000` | Server heartbeat interval (ms) |
| `pingTimeout` | `5000` | Client considered disconnected after this timeout |

## Connection Flow

```mermaid
sequenceDiagram
    participant C as Client
    participant HTTP as Fastify
    participant IO as Socket.IO
    participant Redis as Redis Adapter
    participant EH as Event Handler

    C->>HTTP: GET / (Upgrade Request)
    HTTP->>IO: WebSocket upgrade
    IO->>C: connection established
    IO->>EH: 'connection' event
    EH->>C: ack + socket.id

    C->>IO: emit('join', 'room-123')
    IO->>EH: 'join' event
    EH->>Redis: Subscribe to room channel
    EH->>C: Joined room-123
```

## Event System

### Built-in Events

| Event | Direction | Trigger |
|-------|-----------|---------|
| `connect` | Server → Client | Underlying transport established |
| `disconnect` | Server → Client | Transport closed or heartbeat timeout |
| `connect_error` | Server → Client | Handshake or authentication failure |

### Application Events

| Event | Direction | Payload | Semantics |
|-------|-----------|---------|-----------|
| `join` | Client → Server | `{ roomId: string }` | Subscribe to a room; used for result isolation |
| `leave` | Client → Server | `{ roomId: string }` | Unsubscribe from a room |
| `cancel` | Client → Server | `{ requestId: string }` | Signal cancellation intent |
| `cancel_result` | Server → Client | `{ requestId, success: boolean }` | Acknowledge cancellation (success = true if job found and marked, false if already completed/not found)

### Configurable Result Event

The result event name is configurable via `SOCKET_IO_EVENT` (default: `vision`). This allows multiple service instances or tenants to share the same Socket.IO server without interleaved payloads.

```typescript
interface VisionResponse {
  meta: ImageMeta[];
  task: 'describe' | 'compare' | 'ocr';
  message: { role: 'assistant'; content: string };
  done?: boolean;        // Streaming sentinel
  canceled?: boolean;    // Cancellation marker
  status?: string;       // 'completed' | 'failed' | 'canceled'
}

interface ImageMeta {
  name: string;
  type: string;
  hash: string;
  requestId: string;
  variant?: string;
}
```

## Room Management

### Room Lifecycle

```mermaid
stateDiagram-v2
    [*] --> Global: Connection
    Global --> Room: join(roomId)
    Room --> Room: Receive events
    Room --> Global: leave(roomId)
    Global --> [*]: Disconnection
```

### Room Naming Convention

Rooms are derived from `requestId` to guarantee isolation:

```typescript
const roomId = filters.roomId ?? `vision-${filters.requestId}`;
socket.join(roomId);
```

This design prevents cross-talk between concurrent requests, even when originating from the same client.

## Scaling with Redis Adapter

```mermaid
flowchart TB
    subgraph S1["Server Instance 1"]
        IO1[Socket.IO Server]
    end
    subgraph S2["Server Instance 2"]
        IO2[Socket.IO Server]
    end
    subgraph R["KeyDB / Redis"]
        Pub[Publisher]
        Sub[Subscriber]
    end
    subgraph C1["Client A"]
        Client1
    end
    subgraph C2["Client B"]
        Client2
    end

    Client1 --> IO1
    Client2 --> IO2
    IO1 -->|Publish| Pub
    IO2 -->|Subscribe| Sub
    Pub -->|Broadcast| Sub
```

### Benefits

| Capability | Mechanism |
|------------|-----------|
| Cross-server communication | Redis pub/sub channel per room |
| Room synchronization | Room membership stored in shared Redis state |
| Session affinity independence | Clients can reconnect to any server instance |
| Horizontal elasticity | New server instances auto-subscribe on join events |

## Result Delivery: Vision Event Flow

```mermaid
sequenceDiagram
    participant W as Vision Worker
    participant Svc as SocketIOService
    participant IO as Socket.IO
    participant Redis as Redis Adapter
    participant C as Client

    W->>Svc: emitTo(event, roomId, payload)
    Svc->>IO: io.to(roomId).emit(event, payload)
    IO->>Redis: Publish to room channel
    Redis->>IO: Broadcast to all servers
    IO->>C: vision event
    C->>C: Handle / Render
```

### Streaming Mode

In streaming mode, the worker emits multiple progressive `vision` events:

```
Event 1: { content: "The image", done: false }
Event 2: { content: "The image shows", done: false }
Event 3: { content: "The image shows a cat...", done: true }
```

The `done: true` sentinel signals the final chunk. Clients should concatenate intermediate chunks while rendering.

### Non-Streaming Mode

A single event with the complete `content` string and no `done` field:

```
Event: { content: "The image shows a cat sitting...", done: undefined }
```

## Error Handling

### Connection Errors

```typescript
socket.on('connect_error', (error) => {
  this.logger.error('Connection error:', error);
  // Retry logic via exponential backoff
});
```

### Cancellation Flow

Cancel requests are handled via the `cancel` Socket.IO event or `POST /api/v1/vision/cancel`. The `JobTrackingService` atomically marks the request as canceled. Active workers poll this flag during the Ollama streaming loop:

```typescript
if (this.jobTracking.isCanceled(requestId)) {
  await this.emitToSocket(roomId, event, { requestId, status: 'canceled', canceled: true });
  throw new UnrecoverableError('Job canceled during streaming');
}
```

## Security

### CORS Configuration

CORS is shared between Fastify and Socket.IO via a unified configuration object:

```typescript
APP.enableCors(appConfigService.config.cors);
// CORS_ORIGIN=http://localhost:5173
```

### Room Isolation

- Wildcard room subscriptions are prohibited
- Room IDs are derived from UUID-based `requestId` values, making enumeration impractical
- No room listing API is exposed

## Environment Variables

| Variable | Default | Description |
|----------|---------|-------------|
| `SOCKET_IO_EVENT` | `vision` | Primary result event name |
| `SOCKET_IO_TRANSPORTS` | `websocket,polling,webtransport` | Allowed transports |
| `SOCKET_IO_PING_INTERVAL` | `25000` | Heartbeat interval (ms) |
| `SOCKET_IO_PING_TIMEOUT` | `5000` | Heartbeat timeout (ms) |
| `SOCKET_IO_CORS_ORIGIN` | `*` | Allowed origins |
| `SOCKET_IO_CORS_CREDENTIALS` | `true` | Allow credentials |
| `CORS_ORIGIN` | `*` | Shared CORS origin (prefer `SOCKET_IO_CORS_ORIGIN`) |

## Client Configuration Example

```typescript
import { io } from 'socket.io-client';

const socket = io('http://localhost:3000', {
  transports: ['websocket', 'polling'],
  reconnection: true,
  reconnectionAttempts: 5,
  reconnectionDelay: 1000,
});

socket.emit('join', 'room-123');
socket.on('vision', (data: VisionResponse) => {
  if (data.done) console.log('Complete:', data.message.content);
  else console.log('Chunk:', data.message.content);
});
```

## Monitoring Metrics

| Metric | Measurement Point |
|--------|-------------------|
| Connected clients | `io.sockets.sockets.size` (with Redis: `io.of('/').adapter.sidsCount`) |
| Active rooms | `io.of('/').adapter.roomsCount` |
| Events per second | Custom counter in `SocketService` |
| Reconnection rate | Client-side `reconnect` event frequency |

## Troubleshooting

| Symptom | Cause | Resolution |
|---------|-------|------------|
| Connection refused | Server down or port misconfigured | Verify `PORT` and firewall rules |
| CORS errors | Origin mismatch | Check `CORS_ORIGIN` against client URL |
| Events not received | Room mismatch or wrong event name | Verify `roomId` matches between request and `join` event |
| High latency | Redis connection across WAN | Co-locate KeyDB with API servers |

## Related Documentation

- [1. Server Overview](1-server.md) — Integration with Fastify
- [1.3 BullMQ Async Processing](1.3-bullmq.md) — Worker emission to Socket.IO
- [2.3 Dashboard State Management & Real-time Events](2.3-state-and-realtime.md) — Client-side consumption