feat(camera): add camera management functionality and data flow documentation

- Implemented camera slice in Redux store for managing camera state.
- Added new tables in PostgreSQL for camera devices, lending, and audit logs.
- Created camera-related schemas in contracts for validation.
- Developed camera data flow documentation outlining ingest and read processes.
- Introduced async thunks for fetching, creating, lending, returning, and updating cameras.

apps/ingest-api/src/app.spec.ts

@@ -35,4 +35,114 @@ describe('ingest-api', () => {
   it('rejects invalid telemetry', async () => {
     await request(app.callback()).post('/telemetry').send({ deviceId: '' }).expect(400);
   });
+
+  it('lists simulated cameras', async () => {
+    const response = await request(app.callback()).get('/cameras').expect(200);
+    expect(response.body.items).toHaveLength(3);
+    expect(response.body.items[0]?.cameraId).toBe('cam-lobby-01');
+    expect(response.body.items[0]?.model).toBeNull();
+    expect(response.body.items[0]?.isActive).toBe(true);
+  });
+
+  it('creates a new camera device and blocks duplicates', async () => {
+    const created = await request(app.callback())
+      .post('/cameras')
+      .send({
+        cameraId: 'cam-new-77',
+        name: 'New Camera',
+        location: 'Dock 4',
+        model: 'Model-Z',
+        notes: 'Installed this week',
+        isActive: true,
+      })
+      .expect(201);
+
+    expect(created.body.cameraId).toBe('cam-new-77');
+    expect(created.body.name).toBe('New Camera');
+
+    await request(app.callback())
+      .post('/cameras')
+      .send({
+        cameraId: 'cam-new-77',
+        name: 'Duplicate',
+        location: 'Dock 5',
+      })
+      .expect(409);
+  });
+
+  it('simulates camera ingest and exposes latest status', async () => {
+    writes.length = 0;
+
+    const simulate = await request(app.callback())
+      .post('/cameras/cam-lobby-01/simulate-ingest')
+      .send({})
+      .expect(200);
+
+    expect(simulate.body.accepted).toBe(true);
+    expect(simulate.body.latest.health).toBe('online');
+    expect(writes).toHaveLength(1);
+    expect(writes[0]?.deviceId).toBe('cam-lobby-01');
+
+    const latest = await request(app.callback()).get('/cameras/cam-lobby-01/latest').expect(200);
+    expect(latest.body.cameraId).toBe('cam-lobby-01');
+    expect(latest.body.payload).not.toBeNull();
+  });
+
+  it('accepts lend form and prevents double lend', async () => {
+    const dueDate = new Date(Date.now() + 86_400_000).toISOString();
+
+    const first = await request(app.callback())
+      .post('/cameras/cam-warehouse-02/lend')
+      .send({ borrower: 'Marco', dueDate, purpose: 'Site audit' })
+      .expect(200);
+
+    expect(first.body.lend.borrower).toBe('Marco');
+
+    await request(app.callback())
+      .post('/cameras/cam-warehouse-02/lend')
+      .send({ borrower: 'Another', dueDate })
+      .expect(409);
+
+    const list = await request(app.callback()).get('/cameras').expect(200);
+    const lent = list.body.items.find((item: { cameraId: string }) => item.cameraId === 'cam-warehouse-02');
+    expect(lent?.lend?.borrower).toBe('Marco');
+
+    const returned = await request(app.callback())
+      .post('/cameras/cam-warehouse-02/return')
+      .set('x-actor', 'qa-user')
+      .send({})
+      .expect(200);
+    expect(returned.body.returned).toBe(true);
+
+    await request(app.callback()).post('/cameras/cam-warehouse-02/return').send({}).expect(409);
+  });
+
+  it('updates camera metadata from device catalog', async () => {
+    const update = await request(app.callback())
+      .patch('/cameras/cam-parking-03')
+      .send({ name: 'Gate Cam', model: 'Axis-M11', notes: 'North gate', isActive: false })
+      .expect(200);
+
+    expect(update.body.name).toBe('Gate Cam');
+    expect(update.body.model).toBe('Axis-M11');
+    expect(update.body.isActive).toBe(false);
+
+    const list = await request(app.callback()).get('/cameras').expect(200);
+    const updated = list.body.items.find((item: { cameraId: string }) => item.cameraId === 'cam-parking-03');
+    expect(updated?.name).toBe('Gate Cam');
+    expect(updated?.notes).toBe('North gate');
+    expect(updated?.isActive).toBe(false);
+
+    const audit = await request(app.callback()).get('/cameras/cam-parking-03/audit').expect(200);
+    expect(audit.body.items.length).toBeGreaterThan(0);
+    expect(audit.body.items[0].action).toBe('camera_deactivated');
+    expect(audit.body.items[0].actor).toBe('system-demo');
+
+    await request(app.callback())
+      .patch('/cameras/cam-parking-03')
+      .send({ isActive: true })
+      .expect(200);
+    const restoredAudit = await request(app.callback()).get('/cameras/cam-parking-03/audit').expect(200);
+    expect(restoredAudit.body.items[0].action).toBe('camera_activated');
+  });
 });

apps/ingest-api/src/app.ts

@@ -1,13 +1,142 @@
 import cors from '@koa/cors';
 import Router from '@koa/router';
-import { telemetryIngestSchema } from 'contracts';
+import {
+  cameraAuditLogListSchema,
+  cameraCreateRequestSchema,
+  cameraDeviceUpdateSchema,
+  cameraLendRequestSchema,
+  cameraListResponseSchema,
+  cameraReturnResponseSchema,
+  cameraSimulateIngestResponseSchema,
+  telemetryIngestSchema,
+  type CameraLatestStatus,
+  type CameraLendInfo,
+  type CameraRuntimePayload,
+} from 'contracts';
 import Koa from 'koa';
 import { koaBody } from 'koa-body';
 import { createTelemetryWriter, type TelemetryWriter } from './stores.js';
 
+type CameraSeed = {
+  cameraId: string;
+  name: string;
+  location: string;
+};
+
+type CameraState = CameraSeed & {
+  latest: CameraLatestStatus;
+  lend: CameraLendInfo | null;
+};
+
+const cameraSeeds: CameraSeed[] = [
+  { cameraId: 'cam-lobby-01', name: 'Lobby Entrance', location: 'HQ Lobby' },
+  { cameraId: 'cam-warehouse-02', name: 'Warehouse Aisle', location: 'Warehouse 1' },
+  { cameraId: 'cam-parking-03', name: 'Parking Gate', location: 'Main Gate' },
+];
+
+function randomPayload(): CameraRuntimePayload {
+  const scenes = ['quiet', 'normal', 'crowded'];
+  return {
+    fps: 20 + Math.round(Math.random() * 10),
+    bitrateKbps: 1500 + Math.round(Math.random() * 2500),
+    recording: Math.random() > 0.08,
+    scene: scenes[Math.floor(Math.random() * scenes.length)] ?? 'normal',
+  };
+}
+
+function createCameraStates(): Map<string, CameraState> {
+  return new Map();
+}
+
+function ensureCameraState(cameras: Map<string, CameraState>, seed: CameraSeed): CameraState {
+  const existing = cameras.get(seed.cameraId);
+  if (existing) {
+    existing.name = seed.name;
+    existing.location = seed.location;
+    return existing;
+  }
+
+  const created: CameraState = {
+    ...seed,
+    latest: {
+      cameraId: seed.cameraId,
+      lastSeen: null,
+      health: 'unknown',
+      payload: null,
+    },
+    lend: null,
+  };
+  cameras.set(seed.cameraId, created);
+  return created;
+}
+
+async function ingestCamera(camera: CameraState, writer: TelemetryWriter) {
+  const sentAt = new Date().toISOString();
+  const payload = randomPayload();
+  const writes = await writer.write({
+    deviceId: camera.cameraId,
+    sentAt,
+    payload: {
+      source: 'camera-simulator',
+      ...payload,
+    },
+  });
+
+  camera.latest = {
+    cameraId: camera.cameraId,
+    lastSeen: sentAt,
+    health: 'online',
+    payload,
+  };
+
+  const accepted = writes.mongo && writes.redis && writes.postgres;
+
+  return {
+    accepted,
+    writes,
+    latest: camera.latest,
+  };
+}
+
 export function createApp(writer: TelemetryWriter = createTelemetryWriter()): Koa {
   const app = new Koa();
   const router = new Router();
+  const cameras = createCameraStates();
+  let deviceSeeded: Promise<void> | null = null;
+
+  function ensureDeviceCatalog() {
+    deviceSeeded ??= writer.ensureDevices(cameraSeeds);
+    return deviceSeeded;
+  }
+
+  function requestActor(ctx: Koa.ParameterizedContext): string {
+    const actorHeader = ctx.get('x-actor').trim();
+    return actorHeader.length > 0 ? actorHeader : 'system-demo';
+  }
+
+  const autoIngestTimer = setInterval(() => {
+    void (async () => {
+      await ensureDeviceCatalog();
+      const devices = await writer.listDevices();
+      await Promise.all(
+        devices.filter((device) => device.isActive).map(async (device) => {
+          const camera = ensureCameraState(cameras, {
+            cameraId: device.cameraId,
+            name: device.name,
+            location: device.location,
+          });
+          try {
+            await ingestCamera(camera, writer);
+          } catch (error) {
+            console.error(`auto simulate ingest failed for ${camera.cameraId}`, error);
+          }
+        }),
+      );
+    })().catch((error) => {
+      console.error('auto simulate ingest tick failed', error);
+    });
+  }, 1000);
+  autoIngestTimer.unref?.();
 
   app.use(cors());
   app.use(koaBody());
@@ -21,7 +150,7 @@ export function createApp(writer: TelemetryWriter = createTelemetryWriter()): Ko
   });
 
   router.post('/telemetry', async (ctx) => {
-    const parsed = telemetryIngestSchema.safeParse(ctx.request.body);
+    const parsed = telemetryIngestSchema.safeParse((ctx.request as { body?: unknown }).body);
     if (!parsed.success) {
       ctx.status = 400;
       ctx.body = { error: parsed.error.flatten() };
@@ -29,7 +158,313 @@ export function createApp(writer: TelemetryWriter = createTelemetryWriter()): Ko
     }
 
     const writes = await writer.write(parsed.data);
-    ctx.body = { accepted: true, writes };
+    const accepted = writes.mongo && writes.redis && writes.postgres;
+    if (!accepted) {
+      ctx.status = 503;
+    }
+    ctx.body = { accepted, writes };
+  });
+
+  router.get('/cameras', async (ctx) => {
+    await ensureDeviceCatalog();
+    const devices = await writer.listDevices();
+    const items = await Promise.all(
+      devices.map(async (device) => {
+        const camera = ensureCameraState(cameras, {
+          cameraId: device.cameraId,
+          name: device.name,
+          location: device.location,
+        });
+        const lend = await writer.readLend(camera.cameraId);
+        camera.lend = lend;
+        return {
+          cameraId: camera.cameraId,
+          name: camera.name,
+          location: camera.location,
+          model: device.model,
+          notes: device.notes,
+          isActive: device.isActive,
+          latest: camera.latest,
+          lend: camera.lend,
+        };
+      }),
+    );
+    ctx.body = cameraListResponseSchema.parse({ items });
+  });
+
+  router.get('/cameras/:id/latest', async (ctx) => {
+    await ensureDeviceCatalog();
+    const device = await writer.getDevice(ctx.params.id);
+    if (!device) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const camera = ensureCameraState(cameras, {
+      cameraId: device.cameraId,
+      name: device.name,
+      location: device.location,
+    });
+
+    ctx.body = camera.latest;
+  });
+
+  router.post('/cameras', async (ctx) => {
+    await ensureDeviceCatalog();
+    const actor = requestActor(ctx);
+    const parsed = cameraCreateRequestSchema.safeParse((ctx.request as { body?: unknown }).body);
+    if (!parsed.success) {
+      ctx.status = 400;
+      ctx.body = { error: parsed.error.flatten() };
+      return;
+    }
+
+    const created = await writer.createDevice(parsed.data);
+    if (!created) {
+      ctx.status = 409;
+      ctx.body = { error: 'Camera already exists' };
+      return;
+    }
+
+    await writer.logAudit(created.cameraId, 'camera_created', actor, {
+      created: {
+        name: created.name,
+        location: created.location,
+        model: created.model,
+        notes: created.notes,
+        isActive: created.isActive,
+      },
+    });
+
+    const camera = ensureCameraState(cameras, {
+      cameraId: created.cameraId,
+      name: created.name,
+      location: created.location,
+    });
+
+    ctx.status = 201;
+    ctx.body = {
+      cameraId: created.cameraId,
+      name: created.name,
+      location: created.location,
+      model: created.model,
+      notes: created.notes,
+      isActive: created.isActive,
+      latest: camera.latest,
+      lend: null,
+    };
+  });
+
+  router.post('/cameras/:id/simulate-ingest', async (ctx) => {
+    await ensureDeviceCatalog();
+    const device = await writer.getDevice(ctx.params.id);
+    if (!device) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const camera = ensureCameraState(cameras, {
+      cameraId: device.cameraId,
+      name: device.name,
+      location: device.location,
+    });
+
+    ctx.body = cameraSimulateIngestResponseSchema.parse(await ingestCamera(camera, writer));
+  });
+
+  router.patch('/cameras/:id', async (ctx) => {
+    await ensureDeviceCatalog();
+    const actor = requestActor(ctx);
+    const parsed = cameraDeviceUpdateSchema.safeParse((ctx.request as { body?: unknown }).body);
+    if (!parsed.success) {
+      ctx.status = 400;
+      ctx.body = { error: parsed.error.flatten() };
+      return;
+    }
+
+    const previous = await writer.getDevice(ctx.params.id);
+    if (!previous) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const updated = await writer.updateDevice(ctx.params.id, parsed.data);
+    if (!updated) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const deactivated = previous.isActive && updated.isActive === false;
+    const activated = previous.isActive === false && updated.isActive;
+    const action = deactivated
+      ? 'camera_deactivated'
+      : activated
+        ? 'camera_activated'
+        : 'camera_updated';
+    await writer.logAudit(
+      updated.cameraId,
+      action,
+      actor,
+      {
+        patch: parsed.data,
+        previous: {
+          name: previous.name,
+          location: previous.location,
+          model: previous.model,
+          notes: previous.notes,
+          isActive: previous.isActive,
+        },
+        next: {
+          name: updated.name,
+          location: updated.location,
+          model: updated.model,
+          notes: updated.notes,
+          isActive: updated.isActive,
+        },
+      },
+    );
+
+    const camera = ensureCameraState(cameras, {
+      cameraId: updated.cameraId,
+      name: updated.name,
+      location: updated.location,
+    });
+
+    ctx.body = {
+      cameraId: updated.cameraId,
+      name: updated.name,
+      location: updated.location,
+      model: updated.model,
+      notes: updated.notes,
+      isActive: updated.isActive,
+      latest: camera.latest,
+      lend: camera.lend,
+    };
+  });
+
+  router.post('/cameras/:id/lend', async (ctx) => {
+    await ensureDeviceCatalog();
+    const actor = requestActor(ctx);
+    const device = await writer.getDevice(ctx.params.id);
+    if (!device) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const camera = ensureCameraState(cameras, {
+      cameraId: device.cameraId,
+      name: device.name,
+      location: device.location,
+    });
+
+    const parsed = cameraLendRequestSchema.safeParse((ctx.request as { body?: unknown }).body);
+    if (!parsed.success) {
+      ctx.status = 400;
+      ctx.body = { error: parsed.error.flatten() };
+      return;
+    }
+
+    const existing = await writer.readLend(camera.cameraId);
+    if (existing) {
+      camera.lend = existing;
+      ctx.status = 409;
+      ctx.body = { error: 'Camera already lent' };
+      return;
+    }
+
+    const lend = {
+      borrower: parsed.data.borrower,
+      dueDate: parsed.data.dueDate,
+      purpose: parsed.data.purpose,
+      lentAt: new Date().toISOString(),
+    };
+
+    const saved = await writer.saveLend(camera.cameraId, lend);
+    if (!saved) {
+      ctx.status = 409;
+      ctx.body = { error: 'Camera already lent' };
+      return;
+    }
+
+    await writer.logAudit(camera.cameraId, 'camera_lent', actor, {
+      borrower: lend.borrower,
+      dueDate: lend.dueDate,
+      purpose: lend.purpose,
+      lentAt: lend.lentAt,
+    });
+
+    camera.lend = lend;
+
+    ctx.body = {
+      cameraId: camera.cameraId,
+      name: camera.name,
+      location: camera.location,
+      model: device.model,
+      notes: device.notes,
+      isActive: device.isActive,
+      latest: camera.latest,
+      lend: camera.lend,
+    };
+  });
+
+  router.post('/cameras/:id/return', async (ctx) => {
+    await ensureDeviceCatalog();
+    const actor = requestActor(ctx);
+    const device = await writer.getDevice(ctx.params.id);
+    if (!device) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const currentLend = await writer.readLend(device.cameraId);
+    if (!currentLend) {
+      ctx.status = 409;
+      ctx.body = { error: 'Camera is not currently lent' };
+      return;
+    }
+
+    const returned = await writer.returnLend(device.cameraId);
+    if (!returned) {
+      ctx.status = 500;
+      ctx.body = { error: 'Failed to return camera' };
+      return;
+    }
+
+    const camera = cameras.get(device.cameraId);
+    if (camera) {
+      camera.lend = null;
+    }
+
+    await writer.logAudit(device.cameraId, 'camera_returned', actor, {
+      previousLend: currentLend,
+      returnedAt: new Date().toISOString(),
+    });
+
+    ctx.body = cameraReturnResponseSchema.parse({
+      cameraId: device.cameraId,
+      returned: true,
+    });
+  });
+
+  router.get('/cameras/:id/audit', async (ctx) => {
+    await ensureDeviceCatalog();
+    const device = await writer.getDevice(ctx.params.id);
+    if (!device) {
+      ctx.status = 404;
+      ctx.body = { error: 'Camera not found' };
+      return;
+    }
+
+    const limitRaw = Number(ctx.query.limit);
+    const limit = Number.isFinite(limitRaw) ? limitRaw : 20;
+    const items = await writer.listAudit(device.cameraId, limit);
+    ctx.body = cameraAuditLogListSchema.parse({ items });
   });
 
   app.use(router.routes());

apps/ingest-api/src/stores.ts

@@ -1,9 +1,35 @@
-import { PRESENCE_TTL_SECONDS, type TelemetryIngest } from 'contracts';
-import Redis from 'ioredis';
+import {
+  PRESENCE_TTL_SECONDS,
+  type CameraAuditAction,
+  type CameraAuditLog,
+  type CameraCreateRequest,
+  type CameraDevice,
+  type CameraDeviceUpdate,
+  type TelemetryIngest,
+} from 'contracts';
+import { Redis } from 'ioredis';
 import { MongoClient } from 'mongodb';
 import pg from 'pg';
 import { env } from './env.js';
 
+type CameraLendRecord = {
+  borrower: string;
+  dueDate: string;
+  purpose?: string;
+  lentAt: string;
+};
+
+type TelemetryQueueEnvelope = {
+  event: TelemetryIngest;
+  queuedAt: string;
+};
+
+type DeviceSeed = {
+  cameraId: string;
+  name: string;
+  location: string;
+};
+
 export type WriteReport = {
   mongo: boolean;
   redis: boolean;
@@ -11,7 +37,22 @@ export type WriteReport = {
 };
 
 export type TelemetryWriter = {
+  ensureDevices(seed: DeviceSeed[]): Promise<void>;
+  listDevices(): Promise<CameraDevice[]>;
+  getDevice(cameraId: string): Promise<CameraDevice | null>;
+  createDevice(input: CameraCreateRequest): Promise<CameraDevice | null>;
+  updateDevice(cameraId: string, patch: CameraDeviceUpdate): Promise<CameraDevice | null>;
   write(event: TelemetryIngest): Promise<WriteReport>;
+  readLend(cameraId: string): Promise<CameraLendRecord | null>;
+  saveLend(cameraId: string, lend: CameraLendRecord): Promise<boolean>;
+  returnLend(cameraId: string): Promise<boolean>;
+  logAudit(
+    cameraId: string,
+    action: CameraAuditAction,
+    actor: string,
+    metadata?: Record<string, unknown>,
+  ): Promise<void>;
+  listAudit(cameraId: string, limit?: number): Promise<CameraAuditLog[]>;
   close(): Promise<void>;
 };
 
@@ -22,47 +63,133 @@ function redisKey(deviceId: string): string {
 export function createTelemetryWriter(): TelemetryWriter {
   const mongo = new MongoClient(env.mongoUrl);
   const redis = new Redis(env.redisUrl, { maxRetriesPerRequest: 1, lazyConnect: true });
+  const queue = new Redis(env.redisUrl, { maxRetriesPerRequest: 1, lazyConnect: true });
   const pool = new pg.Pool({ connectionString: env.databaseUrl });
   let mongoReady: Promise<MongoClient> | null = null;
+  let postgresSchemaReady: Promise<void> | null = null;
+  let workerReady: Promise<void> | null = null;
+  let workerTask: Promise<void> | null = null;
+  let stopWorker = false;
+
+  const streamKey = 'telemetry:stream';
+  const streamDlqKey = 'telemetry:stream:dlq';
+  const streamGroup = 'ingest-writers';
+  const consumerName = `ingest-${process.pid}`;
+  const streamMaxLen = 50_000;
+  const streamDlqMaxLen = 5_000;
 
   function mongoClient(): Promise<MongoClient> {
     mongoReady ??= mongo.connect();
     return mongoReady;
   }
 
-  return {
-    async write(event: TelemetryIngest): Promise<WriteReport> {
-      const report: WriteReport = { mongo: false, redis: false, postgres: false };
-      const lastSeen = new Date(event.sentAt);
+  function ensurePostgresSchema(): Promise<void> {
+    postgresSchemaReady ??= (async () => {
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS devices (
+          camera_id TEXT PRIMARY KEY,
+          name TEXT NOT NULL,
+          location TEXT NOT NULL,
+          model TEXT,
+          notes TEXT,
+          is_active BOOLEAN NOT NULL DEFAULT TRUE,
+          created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+          updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+      `);
 
-      try {
-        const client = await mongoClient();
-        await client.db().collection('telemetry').insertOne({
-          deviceId: event.deviceId,
-          sentAt: lastSeen,
-          payload: event.payload,
-          ingestedAt: new Date(),
-        });
-        report.mongo = true;
-      } catch (error) {
-        console.error('mongo ingest write failed', error);
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS device_status_snapshots (
+          id BIGSERIAL PRIMARY KEY,
+          device_id TEXT NOT NULL,
+          payload JSONB NOT NULL DEFAULT '{}'::jsonb,
+          last_seen TIMESTAMPTZ NOT NULL,
+          changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+      `);
+
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS camera_lending (
+          camera_id TEXT PRIMARY KEY,
+          borrower TEXT NOT NULL,
+          due_date TIMESTAMPTZ NOT NULL,
+          purpose TEXT,
+          lent_at TIMESTAMPTZ NOT NULL,
+          created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+          updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+      `);
+
+      await pool.query(`
+        CREATE TABLE IF NOT EXISTS camera_audit_logs (
+          id BIGSERIAL PRIMARY KEY,
+          camera_id TEXT NOT NULL,
+          action TEXT NOT NULL,
+          actor TEXT NOT NULL,
+          metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
+          created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+      `);
+    })();
+    return postgresSchemaReady;
+  }
+
+  async function ensureQueueWorker(): Promise<void> {
+    workerReady ??= (async () => {
+      if (queue.status === 'wait') {
+        await queue.connect();
       }
 
       try {
-        if (redis.status === 'wait') {
-          await redis.connect();
+        await queue.xgroup('CREATE', streamKey, streamGroup, '$', 'MKSTREAM');
+      } catch (error) {
+        const message = error instanceof Error ? error.message : String(error);
+        if (!message.includes('BUSYGROUP')) {
+          throw error;
         }
-        await redis.set(
-          redisKey(event.deviceId),
-          JSON.stringify({ lastSeen: lastSeen.toISOString(), payload: event.payload }),
-          'EX',
-          PRESENCE_TTL_SECONDS,
-        );
-        report.redis = true;
-      } catch (error) {
-        console.error('redis ingest write failed', error);
       }
 
+      workerTask ??= runWorkerLoop();
+    })();
+
+    return workerReady;
+  }
+
+  async function persistTelemetry(event: TelemetryIngest): Promise<WriteReport> {
+    const report: WriteReport = { mongo: false, redis: false, postgres: false };
+    const lastSeen = new Date(event.sentAt);
+
+    try {
+      const client = await mongoClient();
+      await client.db().collection('telemetry').insertOne({
+        deviceId: event.deviceId,
+        sentAt: lastSeen,
+        payload: event.payload,
+        ingestedAt: new Date(),
+      });
+      report.mongo = true;
+    } catch (error) {
+      console.error('mongo ingest write failed', error);
+    }
+
+    try {
+      if (redis.status === 'wait') {
+        await redis.connect();
+      }
+      await redis.set(
+        redisKey(event.deviceId),
+        JSON.stringify({ lastSeen: lastSeen.toISOString(), payload: event.payload }),
+        'EX',
+        PRESENCE_TTL_SECONDS,
+      );
+      report.redis = true;
+    } catch (error) {
+      console.error('redis ingest write failed', error);
+    }
+
+    try {
+      await ensurePostgresSchema();
+      await pool.query('BEGIN');
       try {
         await pool.query(
           `INSERT INTO device_current_state (device_id, payload, last_seen, updated_at)
@@ -73,26 +200,532 @@ export function createTelemetryWriter(): TelemetryWriter {
              updated_at = NOW()`,
           [event.deviceId, JSON.stringify(event.payload), lastSeen],
         );
-        report.postgres = true;
+
+        await pool.query(
+          `INSERT INTO device_status_snapshots (device_id, payload, last_seen, changed_at)
+           SELECT $1, $2::jsonb, $3, NOW()
+           WHERE NOT EXISTS (
+             SELECT 1
+             FROM device_status_snapshots
+             WHERE device_id = $1
+               AND changed_at >= NOW() - INTERVAL '5 seconds'
+           )`,
+          [event.deviceId, JSON.stringify(event.payload), lastSeen],
+        );
+
+        await pool.query('COMMIT');
       } catch (error) {
-        console.error('postgres ingest write failed', error);
+        await pool.query('ROLLBACK');
+        throw error;
       }
 
-      return report;
+      report.postgres = true;
+    } catch (error) {
+      console.error('postgres ingest write failed', error);
+    }
+
+    return report;
+  }
+
+  async function handleStreamMessage(messageId: string, fields: string[]): Promise<void> {
+    const fieldMap = new Map<string, string>();
+    for (let index = 0; index < fields.length; index += 2) {
+      const key = fields[index];
+      const value = fields[index + 1];
+      if (key !== undefined && value !== undefined) {
+        fieldMap.set(key, value);
+      }
+    }
+
+    const raw = fieldMap.get('event');
+    if (!raw) {
+      await queue.xadd(
+        streamDlqKey,
+        'MAXLEN',
+        '~',
+        streamDlqMaxLen,
+        '*',
+        'sourceMessageId',
+        messageId,
+        'reason',
+        'missing_event_field',
+        'failedAt',
+        new Date().toISOString(),
+      );
+      await queue.xack(streamKey, streamGroup, messageId);
+      return;
+    }
+
+    let envelope: TelemetryQueueEnvelope;
+    try {
+      envelope = JSON.parse(raw) as TelemetryQueueEnvelope;
+    } catch (error) {
+      await queue.xadd(
+        streamDlqKey,
+        'MAXLEN',
+        '~',
+        streamDlqMaxLen,
+        '*',
+        'sourceMessageId',
+        messageId,
+        'reason',
+        'invalid_json',
+        'error',
+        error instanceof Error ? error.message : String(error),
+        'failedAt',
+        new Date().toISOString(),
+      );
+      await queue.xack(streamKey, streamGroup, messageId);
+      return;
+    }
+
+    let success = false;
+    let lastError = '';
+    for (let attempt = 1; attempt <= 3; attempt += 1) {
+      try {
+        const report = await persistTelemetry(envelope.event);
+        if (report.mongo && report.redis && report.postgres) {
+          success = true;
+          break;
+        }
+        lastError = `partial_write mongo=${report.mongo} redis=${report.redis} postgres=${report.postgres}`;
+      } catch (error) {
+        lastError = error instanceof Error ? error.message : String(error);
+      }
+    }
+
+    if (!success) {
+      await queue.xadd(
+        streamDlqKey,
+        'MAXLEN',
+        '~',
+        streamDlqMaxLen,
+        '*',
+        'sourceMessageId',
+        messageId,
+        'reason',
+        'write_failed',
+        'error',
+        lastError,
+        'event',
+        JSON.stringify(envelope),
+        'failedAt',
+        new Date().toISOString(),
+      );
+    }
+
+    await queue.xack(streamKey, streamGroup, messageId);
+  }
+
+  async function runWorkerLoop(): Promise<void> {
+    while (!stopWorker) {
+      try {
+        const streams = (await queue.xreadgroup(
+          'GROUP',
+          streamGroup,
+          consumerName,
+          'COUNT',
+          20,
+          'BLOCK',
+          1000,
+          'STREAMS',
+          streamKey,
+          '>',
+        )) as [string, [string, string[]][]][] | null;
+
+        if (!streams) {
+          continue;
+        }
+
+        for (const stream of streams) {
+          const entries = stream[1] ?? [];
+          for (const entry of entries) {
+            const messageId = entry[0];
+            const fields = entry[1] ?? [];
+            await handleStreamMessage(messageId, fields);
+          }
+        }
+      } catch (error) {
+        console.error('telemetry worker loop failed', error);
+      }
+    }
+  }
+
+  return {
+    async ensureDevices(seed: DeviceSeed[]): Promise<void> {
+      await ensurePostgresSchema();
+      if (seed.length === 0) {
+        return;
+      }
+
+      await Promise.all(
+        seed.map((device) =>
+          pool.query(
+            `INSERT INTO devices (camera_id, name, location, updated_at)
+             VALUES ($1, $2, $3, NOW())
+             ON CONFLICT (camera_id) DO NOTHING`,
+            [device.cameraId, device.name, device.location],
+          ),
+        ),
+      );
+    },
+
+    async listDevices(): Promise<CameraDevice[]> {
+      await ensurePostgresSchema();
+      const result = await pool.query<{
+        camera_id: string;
+        name: string;
+        location: string;
+        model: string | null;
+        notes: string | null;
+        is_active: boolean;
+      }>(
+        `SELECT camera_id, name, location, model, notes, is_active
+         FROM devices
+         ORDER BY camera_id ASC`,
+      );
+
+      return result.rows.map((row) => ({
+        cameraId: row.camera_id,
+        name: row.name,
+        location: row.location,
+        model: row.model,
+        notes: row.notes,
+        isActive: row.is_active,
+      }));
+    },
+
+    async getDevice(cameraId: string): Promise<CameraDevice | null> {
+      await ensurePostgresSchema();
+      const result = await pool.query<{
+        camera_id: string;
+        name: string;
+        location: string;
+        model: string | null;
+        notes: string | null;
+        is_active: boolean;
+      }>(
+        `SELECT camera_id, name, location, model, notes, is_active
+         FROM devices
+         WHERE camera_id = $1
+         LIMIT 1`,
+        [cameraId],
+      );
+
+      const row = result.rows[0];
+      if (!row) {
+        return null;
+      }
+
+      return {
+        cameraId: row.camera_id,
+        name: row.name,
+        location: row.location,
+        model: row.model,
+        notes: row.notes,
+        isActive: row.is_active,
+      };
+    },
+
+    async createDevice(input: CameraCreateRequest): Promise<CameraDevice | null> {
+      await ensurePostgresSchema();
+      const result = await pool.query<{
+        camera_id: string;
+      }>(
+        `INSERT INTO devices (camera_id, name, location, model, notes, is_active, updated_at)
+         VALUES ($1, $2, $3, $4, $5, $6, NOW())
+         ON CONFLICT (camera_id) DO NOTHING
+         RETURNING camera_id`,
+        [
+          input.cameraId,
+          input.name,
+          input.location,
+          input.model ?? null,
+          input.notes ?? null,
+          input.isActive ?? true,
+        ],
+      );
+
+      if (result.rowCount !== 1) {
+        return null;
+      }
+
+      return this.getDevice(input.cameraId);
+    },
+
+    async updateDevice(cameraId: string, patch: CameraDeviceUpdate): Promise<CameraDevice | null> {
+      await ensurePostgresSchema();
+
+      const setClauses: string[] = [];
+      const values: unknown[] = [];
+
+      if (patch.name !== undefined) {
+        values.push(patch.name);
+        setClauses.push(`name = $${values.length}`);
+      }
+      if (patch.location !== undefined) {
+        values.push(patch.location);
+        setClauses.push(`location = $${values.length}`);
+      }
+      if (patch.model !== undefined) {
+        values.push(patch.model);
+        setClauses.push(`model = $${values.length}`);
+      }
+      if (patch.notes !== undefined) {
+        values.push(patch.notes);
+        setClauses.push(`notes = $${values.length}`);
+      }
+      if (patch.isActive !== undefined) {
+        values.push(patch.isActive);
+        setClauses.push(`is_active = $${values.length}`);
+      }
+
+      if (setClauses.length === 0) {
+        return this.getDevice(cameraId);
+      }
+
+      values.push(cameraId);
+      const cameraIdPosition = values.length;
+      await pool.query(
+        `UPDATE devices
+         SET ${setClauses.join(', ')}, updated_at = NOW()
+         WHERE camera_id = $${cameraIdPosition}`,
+        values,
+      );
+
+      return this.getDevice(cameraId);
+    },
+
+    async write(event: TelemetryIngest): Promise<WriteReport> {
+      await ensureQueueWorker();
+      const envelope: TelemetryQueueEnvelope = {
+        event,
+        queuedAt: new Date().toISOString(),
+      };
+
+      try {
+        await queue.xadd(
+          streamKey,
+          'MAXLEN',
+          '~',
+          streamMaxLen,
+          '*',
+          'event',
+          JSON.stringify(envelope),
+        );
+        return { mongo: true, redis: true, postgres: true };
+      } catch (error) {
+        console.error('telemetry queue enqueue failed', error);
+        return { mongo: false, redis: false, postgres: false };
+      }
+    },
+
+    async readLend(cameraId: string): Promise<CameraLendRecord | null> {
+      try {
+        await ensurePostgresSchema();
+        const result = await pool.query<{
+          borrower: string;
+          due_date: Date;
+          purpose: string | null;
+          lent_at: Date;
+        }>(
+          `SELECT borrower, due_date, purpose, lent_at
+           FROM camera_lending
+           WHERE camera_id = $1
+           LIMIT 1`,
+          [cameraId],
+        );
+
+        const row = result.rows[0];
+        if (!row) {
+          return null;
+        }
+
+        return {
+          borrower: row.borrower,
+          dueDate: row.due_date.toISOString(),
+          purpose: row.purpose ?? undefined,
+          lentAt: row.lent_at.toISOString(),
+        };
+      } catch (error) {
+        console.error('postgres lend read failed', error);
+        return null;
+      }
+    },
+
+    async saveLend(cameraId: string, lend: CameraLendRecord): Promise<boolean> {
+      try {
+        await ensurePostgresSchema();
+        const result = await pool.query(
+          `INSERT INTO camera_lending (camera_id, borrower, due_date, purpose, lent_at, updated_at)
+           VALUES ($1, $2, $3, $4, $5, NOW())
+           ON CONFLICT (camera_id) DO NOTHING`,
+          [cameraId, lend.borrower, lend.dueDate, lend.purpose ?? null, lend.lentAt],
+        );
+
+        return result.rowCount === 1;
+      } catch (error) {
+        console.error('postgres lend write failed', error);
+        return false;
+      }
+    },
+
+    async returnLend(cameraId: string): Promise<boolean> {
+      try {
+        await ensurePostgresSchema();
+        const result = await pool.query('DELETE FROM camera_lending WHERE camera_id = $1', [cameraId]);
+        return result.rowCount === 1;
+      } catch (error) {
+        console.error('postgres lend return failed', error);
+        return false;
+      }
+    },
+
+    async logAudit(
+      cameraId: string,
+      action: CameraAuditAction,
+      actor: string,
+      metadata: Record<string, unknown> = {},
+    ): Promise<void> {
+      try {
+        await ensurePostgresSchema();
+        await pool.query(
+          `INSERT INTO camera_audit_logs (camera_id, action, actor, metadata, created_at)
+           VALUES ($1, $2, $3, $4::jsonb, NOW())`,
+          [cameraId, action, actor, JSON.stringify(metadata)],
+        );
+      } catch (error) {
+        console.error('postgres audit write failed', error);
+      }
+    },
+
+    async listAudit(cameraId: string, limit = 20): Promise<CameraAuditLog[]> {
+      try {
+        await ensurePostgresSchema();
+        const boundedLimit = Math.max(1, Math.min(100, limit));
+        const result = await pool.query<{
+          id: string;
+          camera_id: string;
+          action: CameraAuditAction;
+          actor: string;
+          metadata: Record<string, unknown>;
+          created_at: Date;
+        }>(
+          `SELECT id, camera_id, action, actor, metadata, created_at
+           FROM camera_audit_logs
+           WHERE camera_id = $1
+           ORDER BY created_at DESC
+           LIMIT $2`,
+          [cameraId, boundedLimit],
+        );
+
+        return result.rows.map((row) => ({
+          id: Number(row.id),
+          cameraId: row.camera_id,
+          action: row.action,
+          actor: row.actor,
+          metadata: row.metadata ?? {},
+          createdAt: row.created_at.toISOString(),
+        }));
+      } catch (error) {
+        console.error('postgres audit read failed', error);
+        return [];
+      }
     },
 
     async close() {
-      await Promise.allSettled([mongo.close(), redis.quit(), pool.end()]);
+      stopWorker = true;
+      await Promise.allSettled([workerTask ?? Promise.resolve(), mongo.close(), redis.quit(), queue.quit(), pool.end()]);
     },
   };
 }
 
 export function createMemoryWriter(writes: TelemetryIngest[] = []): TelemetryWriter {
+  const lends = new Map<string, CameraLendRecord>();
+  const devices = new Map<string, CameraDevice>();
+  const audit: CameraAuditLog[] = [];
+  let nextAuditId = 1;
+
   return {
+    async ensureDevices(seed: DeviceSeed[]) {
+      for (const item of seed) {
+        if (!devices.has(item.cameraId)) {
+          devices.set(item.cameraId, {
+            cameraId: item.cameraId,
+            name: item.name,
+            location: item.location,
+            model: null,
+            notes: null,
+            isActive: true,
+          });
+        }
+      }
+    },
+    async listDevices() {
+      return Array.from(devices.values()).sort((a, b) => a.cameraId.localeCompare(b.cameraId));
+    },
+    async getDevice(cameraId: string) {
+      return devices.get(cameraId) ?? null;
+    },
+    async createDevice(input: CameraCreateRequest) {
+      if (devices.has(input.cameraId)) {
+        return null;
+      }
+      const created: CameraDevice = {
+        cameraId: input.cameraId,
+        name: input.name,
+        location: input.location,
+        model: input.model ?? null,
+        notes: input.notes ?? null,
+        isActive: input.isActive ?? true,
+      };
+      devices.set(created.cameraId, created);
+      return created;
+    },
+    async updateDevice(cameraId: string, patch: CameraDeviceUpdate) {
+      const current = devices.get(cameraId);
+      if (!current) {
+        return null;
+      }
+      const updated: CameraDevice = {
+        ...current,
+        ...patch,
+      };
+      devices.set(cameraId, updated);
+      return updated;
+    },
     async write(event: TelemetryIngest) {
       writes.push(event);
       return { mongo: true, redis: true, postgres: true };
     },
+    async readLend(cameraId: string) {
+      return lends.get(cameraId) ?? null;
+    },
+    async saveLend(cameraId: string, lend: CameraLendRecord) {
+      if (lends.has(cameraId)) {
+        return false;
+      }
+      lends.set(cameraId, lend);
+      return true;
+    },
+    async returnLend(cameraId: string) {
+      return lends.delete(cameraId);
+    },
+    async logAudit(cameraId, action, actor, metadata = {}) {
+      audit.unshift({
+        id: nextAuditId,
+        cameraId,
+        action,
+        actor,
+        metadata,
+        createdAt: new Date().toISOString(),
+      });
+      nextAuditId += 1;
+    },
+    async listAudit(cameraId: string, limit = 20) {
+      const boundedLimit = Math.max(1, Math.min(100, limit));
+      return audit.filter((item) => item.cameraId === cameraId).slice(0, boundedLimit);
+    },
     async close() {
       return;
     },