# Camera Data Flow

This document describes the current camera ingest and read flow.

## Overview

- Active cameras auto-generate ingest events every 1 second.
- Events are enqueued to Redis Stream `telemetry:stream`.
- A Redis Stream worker consumes events, retries failed writes up to 3 times, and dead-letters failures.
- Worker writes:
  - Mongo raw telemetry events
  - Redis presence key with TTL
  - Postgres latest current state on every event
  - Postgres analytics snapshot at most once every 5 seconds per device
- UI reads camera status by polling every 1 second.

## System Flow Diagram

```mermaid
flowchart LR
  subgraph CameraLane[Camera Lane]
    C1[Active Cameras]
    T1[Auto Simulator in ingest-api\nEvery 1 second]
    C1 --> T1
  end

  subgraph IngestAPI[ingest-api Koa]
    E1[Build telemetry event]
    Q1[XADD telemetry:stream]
    API[POST /telemetry]
    E1 --> Q1
    API --> Q1
  end

  T1 --> E1

  subgraph Redis[Redis]
    S1[(Stream: telemetry:stream)]
    G1[Consumer Group Worker\nRetries up to 3]
    DLQ[(Stream: telemetry:stream:dlq)]
    P1[(Presence key device:id TTL)]
  end

  Q1 --> S1
  S1 --> G1
  G1 -->|on repeated failure| DLQ

  subgraph Stores[Storage Writes by Worker]
    M1[(Mongo telemetry raw events)]
    PG1[(Postgres device_current_state\nLatest status)]
    PG2[(Postgres device_status_snapshots\nAt most once per 5 sec per device)]
  end

  G1 --> M1
  G1 --> P1
  G1 --> PG1
  G1 --> PG2

  subgraph ReadPath[Read Path]
    UI[Web /devices]
    CAMAPI[GET /cameras\nGET /cameras/:id/latest]
    UI -->|poll every 1 second| CAMAPI
    CAMAPI --> PG1
    CAMAPI --> P1
  end
```

## Event Timeline Diagram

```mermaid
sequenceDiagram
  participant Cam as Camera Simulator
  participant Ingest as ingest api
  participant Stream as Redis Stream
  participant Worker as Stream Worker
  participant Mongo as MongoDB
  participant Pg as Postgres
  participant Web as Web UI

  loop every second for each active camera
    Cam->>Ingest: create telemetry event
    Ingest->>Stream: XADD telemetry stream
    Ingest-->>Cam: accepted when enqueue succeeds
  end

  loop worker processing
    Worker->>Stream: XREADGROUP
    Worker->>Mongo: insert raw telemetry
    Worker->>Pg: upsert latest device current state
    Worker->>Pg: insert snapshot if older than 5 seconds
    Worker->>Stream: XACK
    Note over Worker,Stream: retry up to three times then write to DLQ
  end

  loop web refresh every second
    Web->>Ingest: GET cameras
    Ingest->>Pg: read latest state
    Ingest-->>Web: return camera rows with latest status
  end
```