From d3b6c5032f2720a722c9271d5d91db475b16a3ec Mon Sep 17 00:00:00 2001 From: Marco Chan Date: Fri, 4 Sep 2026 12:05:52 +0800 Subject: [PATCH] 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 | 110 +++++ apps/ingest-api/src/app.ts | 441 +++++++++++++++++- apps/ingest-api/src/stores.ts | 697 +++++++++++++++++++++++++++-- apps/web/src/App.tsx | 13 +- apps/web/src/pages/DevicesPage.tsx | 642 ++++++++++++++++++++++---- apps/web/src/store/camera-slice.ts | 466 +++++++++++++++++++ apps/web/src/store/index.ts | 2 + docs/camera-data-flow.md | 100 +++++ infra/postgres/init.sql | 38 ++ packages/contracts/src/camera.ts | 128 ++++++ packages/contracts/src/index.ts | 1 + 11 files changed, 2524 insertions(+), 114 deletions(-) create mode 100644 apps/web/src/store/camera-slice.ts create mode 100644 docs/camera-data-flow.md create mode 100644 packages/contracts/src/camera.ts diff --git a/apps/ingest-api/src/app.spec.ts b/apps/ingest-api/src/app.spec.ts index 3d1d814..90b62f1 100644 --- a/apps/ingest-api/src/app.spec.ts +++ b/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'); + }); }); diff --git a/apps/ingest-api/src/app.ts b/apps/ingest-api/src/app.ts index 96a3e8a..b71d15c 100644 --- a/apps/ingest-api/src/app.ts +++ b/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 { + return new Map(); +} + +function ensureCameraState(cameras: Map, 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 | 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()); diff --git a/apps/ingest-api/src/stores.ts b/apps/ingest-api/src/stores.ts index 41dec26..b8f20b6 100644 --- a/apps/ingest-api/src/stores.ts +++ b/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; + listDevices(): Promise; + getDevice(cameraId: string): Promise; + createDevice(input: CameraCreateRequest): Promise; + updateDevice(cameraId: string, patch: CameraDeviceUpdate): Promise; write(event: TelemetryIngest): Promise; + readLend(cameraId: string): Promise; + saveLend(cameraId: string, lend: CameraLendRecord): Promise; + returnLend(cameraId: string): Promise; + logAudit( + cameraId: string, + action: CameraAuditAction, + actor: string, + metadata?: Record, + ): Promise; + listAudit(cameraId: string, limit?: number): Promise; close(): Promise; }; @@ -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 | null = null; + let postgresSchemaReady: Promise | null = null; + let workerReady: Promise | null = null; + let workerTask: Promise | 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 { mongoReady ??= mongo.connect(); return mongoReady; } - return { - async write(event: TelemetryIngest): Promise { - const report: WriteReport = { mongo: false, redis: false, postgres: false }; - const lastSeen = new Date(event.sentAt); + function ensurePostgresSchema(): Promise { + 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 { + 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 { + 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 { + const fieldMap = new Map(); + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 = {}, + ): Promise { + 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 { + 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; + 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(); + const devices = new Map(); + 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; }, diff --git a/apps/web/src/App.tsx b/apps/web/src/App.tsx index a712c5c..8260aa6 100644 --- a/apps/web/src/App.tsx +++ b/apps/web/src/App.tsx @@ -1,4 +1,4 @@ -import { NavLink, Navigate, Route, Routes } from 'react-router-dom' +import { NavLink, Navigate, Route, Routes, useLocation } from 'react-router-dom' import { DevicesPage } from './pages/DevicesPage' import { GreetingPage } from './pages/GreetingPage' @@ -8,10 +8,17 @@ const linkClass = ({ isActive }: { isActive: boolean }) => }` export default function App() { + const location = useLocation() + const isDevicesRoute = location.pathname.startsWith('/devices') + const containerClass = isDevicesRoute ? 'mx-auto w-full max-w-[96rem] px-4 py-8' : 'mx-auto max-w-3xl px-4 py-8' + const headerContainerClass = isDevicesRoute + ? 'mx-auto flex w-full max-w-[96rem] items-center justify-between px-4 py-3' + : 'mx-auto flex max-w-3xl items-center justify-between px-4 py-3' + return (
-
+
marco-trest-test
-
+
} /> } /> diff --git a/apps/web/src/pages/DevicesPage.tsx b/apps/web/src/pages/DevicesPage.tsx index 3acb45d..e954643 100644 --- a/apps/web/src/pages/DevicesPage.tsx +++ b/apps/web/src/pages/DevicesPage.tsx @@ -1,104 +1,594 @@ import { useEffect } from 'react' -import type { DeviceStatus } from 'contracts' -import { PRESENCE_TTL_SECONDS } from 'contracts' -import { fetchDeviceStatus, pingDevice } from '../store/device-slice' +import type { CameraListItem } from 'contracts' +import { + closeCreateModal, + closeAuditModal, + closeEditModal, + closeLendModal, + createCamera, + fetchCameraAudit, + fetchCameras, + openCreateModal, + openEditModal, + openLendModal, + returnCamera, + saveCameraUpdates, + setCreateCameraId, + setCreateIsActive, + setCreateLocation, + setCreateModel, + setCreateName, + setCreateNotes, + setEditIsActive, + setEditLocation, + setEditModel, + setEditName, + setEditNotes, + setLendBorrower, + setLendDueDate, + setLendPurpose, + simulateCameraIngest, + submitCameraLend, + toggleCameraActivation, +} from '../store/camera-slice' import { useAppDispatch, useAppSelector } from '../store/hooks' -function StatusCard({ title, status }: { title: string; status: DeviceStatus | undefined }) { - if (!status) { - return ( -
-

{title}

-

No data yet.

-
- ) +function statusPillClass(health: string) { + if (health === 'online') { + return 'bg-emerald-50 text-emerald-700 ring-1 ring-emerald-200' } - - return ( -
-

{title}

-
-
-
online
-
- {status.online ? 'yes' : 'no'} -
-
-
-
source
-
{status.source}
-
-
-
age
-
{status.ageMs === null ? '—' : `${status.ageMs} ms`}
-
-
-
lastSeen
-
{status.lastSeen ?? '—'}
-
-
-
payload
-
- {JSON.stringify(status.payload)} -
-
-
-
- ) + if (health === 'offline') { + return 'bg-rose-50 text-rose-700 ring-1 ring-rose-200' + } + return 'bg-slate-100 text-slate-600 ring-1 ring-slate-200' } export function DevicesPage() { const dispatch = useAppDispatch() - const { deviceId, pair, pinging, error } = useAppSelector((state) => state.device) + const { + cameras, + loading, + error, + message, + isCreating, + simulatingCameraId, + createModalOpen, + lendModalCameraId, + editModalCameraId, + auditModalCameraId, + auditLogs, + isAuditLoading, + isLending, + isUpdating, + isReturning, + isTogglingActivation, + createForm, + lendForm, + editForm, + } = useAppSelector((state) => state.cameraPage) + + const lendModalCamera = cameras.find((camera) => camera.cameraId === lendModalCameraId) ?? null + const editModalCamera = cameras.find((camera) => camera.cameraId === editModalCameraId) ?? null + const auditModalCamera = cameras.find((camera) => camera.cameraId === auditModalCameraId) ?? null useEffect(() => { - void dispatch(fetchDeviceStatus(deviceId)) + void dispatch(fetchCameras()) + const timer = window.setInterval(() => { - void dispatch(fetchDeviceStatus(deviceId)) + void dispatch(fetchCameras()) }, 1000) + return () => window.clearInterval(timer) - }, [dispatch, deviceId]) + }, [dispatch]) + + function submitLend(camera: CameraListItem) { + void dispatch(submitCameraLend(camera.cameraId)) + } + + function submitUpdate(camera: CameraListItem) { + void dispatch(saveCameraUpdates(camera.cameraId)) + } return (
-

Device status

+

Camera Operations

- Product UI reads Express → Postgres. Redis is only a TTL presence copy. After{' '} - {PRESENCE_TTL_SECONDS}s without a ping, Redis goes empty; Postgres keeps the last row. + Manage camera status in one table, auto-refreshed every second.

-

- Ping device talks to ingest-api — demo only. Production UI never calls the device API. +

+ Auto ingest is running every second.

- - {error ? ( -

{error}

+
+

Total cameras: {cameras.length}

+ +
+ + {error ?

{error}

: null} + {message ? ( +

{message}

) : null} + +
+
+ + + + + + + + + + + + + + + {loading && cameras.length === 0 ? ( + + + + ) : null} + {!loading && cameras.length === 0 ? ( + + + + ) : null} + {cameras.map((camera) => ( + + + + + + + + + + + ))} + +
CameraLocationHealthLast SeenLatest PayloadLendingStateActions
+ Loading cameras... +
+ No cameras found. +
+

{camera.name}

+

{camera.cameraId}

+

{camera.model ?? 'Model n/a'}

+ 		{camera.location} + + {camera.latest.health} + + {camera.latest.lastSeen ?? '—'} + 
+                        {JSON.stringify(camera.latest.payload ?? {}, null, 0)}
+
+ 		+ {camera.lend ? ( +
+

{camera.lend.borrower}

+

Due {new Date(camera.lend.dueDate).toLocaleString()}

+
+ ) : ( + Available + )} + 		+ {camera.isActive ? ( + + Activated + + ) : ( + + Deactivated + + )} + +
+ + + + + + +
+
+
+
-
-

Data flow

- 
{`Device ping → Koa ingest :3001
-  → Mongo (raw event)
-  → Redis SET device:id EX ${PRESENCE_TTL_SECONDS}s
-  → Postgres UPSERT device_current_state
+      {lendModalCamera ? (
+        

+          

+            

+              

+                
Lend camera

+                

+                  {lendModalCamera.name} ({lendModalCamera.cameraId})
+                

+              

+              
+            

 
-React poll 1s → Express crud :3000
-  → SELECT postgres snapshot   ← near-real-time product path
-  → GET redis key              ← optional presence`}
-
+
{ + event.preventDefault() + submitLend(lendModalCamera) + }} + className="mt-4 space-y-3" + > + -
- - -
+ + + + +
+ + +
+
+
+ + ) : null} + + {createModalOpen ? ( +
+
+
+
+

Create camera

+

Register a new camera device in the catalog.

+
+ +
+ +
{ + event.preventDefault() + void dispatch(createCamera()) + }} + className="mt-4 space-y-3" + > + + + + + + + + + + + + +
+ + +
+
+
+
+ ) : null} + + {editModalCamera ? ( +
+
+
+
+

Update camera

+

+ {editModalCamera.name} ({editModalCamera.cameraId}) +

+
+ +
+ +
{ + event.preventDefault() + submitUpdate(editModalCamera) + }} + className="mt-4 space-y-3" + > + + + + + + + + + + +
+ + +
+
+
+
+ ) : null} + + {auditModalCamera ? ( +
+
+
+
+

Audit log

+

+ {auditModalCamera.name} ({auditModalCamera.cameraId}) +

+
+ +
+ +
+ + + + + + + + + + + {isAuditLoading ? ( + + + + ) : null} + {!isAuditLoading && auditLogs.length === 0 ? ( + + + + ) : null} + {auditLogs.map((log) => ( + + + + + + + ))} + +
TimeActionActorMetadata
+ Loading audit logs... +
+ No audit logs found. +
{new Date(log.createdAt).toLocaleString()}{log.action}{log.actor} + 
+                          {JSON.stringify(log.metadata)}
+
+
+
+
+
+ ) : null} ) } diff --git a/apps/web/src/store/camera-slice.ts b/apps/web/src/store/camera-slice.ts new file mode 100644 index 0000000..34ba413 --- /dev/null +++ b/apps/web/src/store/camera-slice.ts @@ -0,0 +1,466 @@ +import { createAsyncThunk, createSlice, type PayloadAction } from '@reduxjs/toolkit' +import { + cameraAuditLogListSchema, + cameraCreateRequestSchema, + cameraDeviceUpdateSchema, + cameraListResponseSchema, + cameraReturnResponseSchema, + cameraSimulateIngestResponseSchema, + type CameraAuditLog, + type CameraListItem, +} from 'contracts' +import { ingestApi } from '../api/client' +import { toErrorMessage } from './greeting-slice' +import type { RootState } from './index' + +export type CameraFormState = { + borrower: string + purpose: string + dueDate: string +} + +export type CameraEditFormState = { + name: string + location: string + model: string + notes: string + isActive: boolean +} + +export type CameraPageState = { + cameras: CameraListItem[] + loading: boolean + error: string | null + message: string | null + isCreating: boolean + simulatingCameraId: string | null + createModalOpen: boolean + lendModalCameraId: string | null + editModalCameraId: string | null + auditModalCameraId: string | null + auditLogs: CameraAuditLog[] + isAuditLoading: boolean + isLending: boolean + isUpdating: boolean + isReturning: string | null + isTogglingActivation: string | null + lendForm: CameraFormState + createForm: CameraEditFormState & { cameraId: string } + editForm: CameraEditFormState +} + +const initialState: CameraPageState = { + cameras: [], + loading: false, + error: null, + message: null, + isCreating: false, + simulatingCameraId: null, + createModalOpen: false, + lendModalCameraId: null, + editModalCameraId: null, + auditModalCameraId: null, + auditLogs: [], + isAuditLoading: false, + isLending: false, + isUpdating: false, + isReturning: null, + isTogglingActivation: null, + lendForm: { + borrower: '', + purpose: '', + dueDate: '', + }, + createForm: { + cameraId: '', + name: '', + location: '', + model: '', + notes: '', + isActive: true, + }, + editForm: { + name: '', + location: '', + model: '', + notes: '', + isActive: true, + }, +} + +export const fetchCameras = createAsyncThunk( + 'cameraPage/fetchCameras', + async (_, { rejectWithValue }) => { + try { + const response = await ingestApi.get('/cameras') + const parsed = cameraListResponseSchema.parse(response.data) + return parsed.items + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } + }, +) + +export const createCamera = createAsyncThunk( + 'cameraPage/createCamera', + async (_, { dispatch, getState, rejectWithValue }) => { + const { createForm } = getState().cameraPage + try { + const body = cameraCreateRequestSchema.parse({ + cameraId: createForm.cameraId, + name: createForm.name, + location: createForm.location, + model: createForm.model ? createForm.model : null, + notes: createForm.notes ? createForm.notes : null, + isActive: createForm.isActive, + }) + + await ingestApi.post('/cameras', body) + await dispatch(fetchCameras()).unwrap() + return body.cameraId + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } + }, +) + +export const simulateCameraIngest = createAsyncThunk( + 'cameraPage/simulateIngest', + async (cameraId, { dispatch, rejectWithValue }) => { + try { + const response = await ingestApi.post(`/cameras/${cameraId}/simulate-ingest`, {}) + cameraSimulateIngestResponseSchema.parse(response.data) + await dispatch(fetchCameras()).unwrap() + return cameraId + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } + }, +) + +export const submitCameraLend = createAsyncThunk( + 'cameraPage/submitLend', + async (cameraId, { dispatch, getState, rejectWithValue }) => { + const { lendForm } = getState().cameraPage + if (!lendForm.borrower || !lendForm.dueDate) { + return rejectWithValue('Borrower and due date are required') + } + + try { + await ingestApi.post(`/cameras/${cameraId}/lend`, { + borrower: lendForm.borrower, + dueDate: new Date(lendForm.dueDate).toISOString(), + purpose: lendForm.purpose || undefined, + }) + await dispatch(fetchCameras()).unwrap() + return cameraId + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } + }, +) + +export const returnCamera = createAsyncThunk( + 'cameraPage/returnCamera', + async (cameraId, { dispatch, rejectWithValue }) => { + try { + const response = await ingestApi.post(`/cameras/${cameraId}/return`, {}) + cameraReturnResponseSchema.parse(response.data) + await dispatch(fetchCameras()).unwrap() + return cameraId + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } + }, +) + +export const saveCameraUpdates = createAsyncThunk< + string, + string, + { state: RootState; rejectValue: string } +>('cameraPage/saveUpdates', async (cameraId, { dispatch, getState, rejectWithValue }) => { + const { editForm } = getState().cameraPage + + try { + const patch = cameraDeviceUpdateSchema.parse({ + name: editForm.name, + location: editForm.location, + model: editForm.model ? editForm.model : null, + notes: editForm.notes ? editForm.notes : null, + isActive: editForm.isActive, + }) + await ingestApi.patch(`/cameras/${cameraId}`, patch) + await dispatch(fetchCameras()).unwrap() + return cameraId + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } +}) + +export const toggleCameraActivation = createAsyncThunk< + { cameraId: string; nextActive: boolean }, + { cameraId: string; isActive: boolean }, + { rejectValue: string } +>('cameraPage/toggleActivation', async ({ cameraId, isActive }, { dispatch, rejectWithValue }) => { + try { + const nextActive = !isActive + await ingestApi.patch(`/cameras/${cameraId}`, { isActive: nextActive }) + await dispatch(fetchCameras()).unwrap() + return { cameraId, nextActive } + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } +}) + +export const fetchCameraAudit = createAsyncThunk< + { cameraId: string; items: CameraAuditLog[] }, + string, + { rejectValue: string } +>('cameraPage/fetchAudit', async (cameraId, { rejectWithValue }) => { + try { + const response = await ingestApi.get(`/cameras/${cameraId}/audit`, { params: { limit: 50 } }) + const parsed = cameraAuditLogListSchema.parse(response.data) + return { cameraId, items: parsed.items } + } catch (error) { + return rejectWithValue(toErrorMessage(error)) + } +}) + +const cameraPageSlice = createSlice({ + name: 'cameraPage', + initialState, + reducers: { + clearBanner(state) { + state.error = null + state.message = null + }, + openLendModal(state, action: PayloadAction) { + state.lendModalCameraId = action.payload + state.lendForm = { borrower: '', purpose: '', dueDate: '' } + state.error = null + state.message = null + }, + openCreateModal(state) { + state.createModalOpen = true + state.createForm = { + cameraId: '', + name: '', + location: '', + model: '', + notes: '', + isActive: true, + } + state.error = null + state.message = null + }, + closeCreateModal(state) { + state.createModalOpen = false + }, + setCreateCameraId(state, action: PayloadAction) { + state.createForm.cameraId = action.payload + }, + setCreateName(state, action: PayloadAction) { + state.createForm.name = action.payload + }, + setCreateLocation(state, action: PayloadAction) { + state.createForm.location = action.payload + }, + setCreateModel(state, action: PayloadAction) { + state.createForm.model = action.payload + }, + setCreateNotes(state, action: PayloadAction) { + state.createForm.notes = action.payload + }, + setCreateIsActive(state, action: PayloadAction) { + state.createForm.isActive = action.payload + }, + closeLendModal(state) { + state.lendModalCameraId = null + }, + setLendBorrower(state, action: PayloadAction) { + state.lendForm.borrower = action.payload + }, + setLendPurpose(state, action: PayloadAction) { + state.lendForm.purpose = action.payload + }, + setLendDueDate(state, action: PayloadAction) { + state.lendForm.dueDate = action.payload + }, + openEditModal(state, action: PayloadAction) { + const camera = action.payload + state.editModalCameraId = camera.cameraId + state.editForm = { + name: camera.name, + location: camera.location, + model: camera.model ?? '', + notes: camera.notes ?? '', + isActive: camera.isActive, + } + state.error = null + state.message = null + }, + closeEditModal(state) { + state.editModalCameraId = null + }, + setEditName(state, action: PayloadAction) { + state.editForm.name = action.payload + }, + setEditLocation(state, action: PayloadAction) { + state.editForm.location = action.payload + }, + setEditModel(state, action: PayloadAction) { + state.editForm.model = action.payload + }, + setEditNotes(state, action: PayloadAction) { + state.editForm.notes = action.payload + }, + setEditIsActive(state, action: PayloadAction) { + state.editForm.isActive = action.payload + }, + closeAuditModal(state) { + state.auditModalCameraId = null + state.auditLogs = [] + }, + }, + extraReducers: (builder) => { + builder + .addCase(fetchCameras.pending, (state) => { + state.loading = true + state.error = null + }) + .addCase(fetchCameras.fulfilled, (state, action) => { + state.loading = false + state.cameras = action.payload + }) + .addCase(fetchCameras.rejected, (state, action) => { + state.loading = false + state.error = action.payload ?? action.error.message ?? 'Failed to fetch camera list' + }) + .addCase(createCamera.pending, (state) => { + state.isCreating = true + state.error = null + state.message = null + }) + .addCase(createCamera.fulfilled, (state, action) => { + state.isCreating = false + state.createModalOpen = false + state.message = `Camera ${action.payload} created successfully` + }) + .addCase(createCamera.rejected, (state, action) => { + state.isCreating = false + state.error = action.payload ?? action.error.message ?? 'Failed to create camera' + }) + .addCase(simulateCameraIngest.pending, (state, action) => { + state.simulatingCameraId = action.meta.arg + state.error = null + state.message = null + }) + .addCase(simulateCameraIngest.fulfilled, (state, action) => { + state.simulatingCameraId = null + state.message = `Simulated ingest for ${action.payload}` + }) + .addCase(simulateCameraIngest.rejected, (state, action) => { + state.simulatingCameraId = null + state.error = action.payload ?? action.error.message ?? 'Failed to simulate camera ingest' + }) + .addCase(submitCameraLend.pending, (state) => { + state.isLending = true + state.error = null + state.message = null + }) + .addCase(submitCameraLend.fulfilled, (state, action) => { + state.isLending = false + state.lendModalCameraId = null + state.lendForm = { borrower: '', purpose: '', dueDate: '' } + state.message = `Camera ${action.payload} lent successfully` + }) + .addCase(submitCameraLend.rejected, (state, action) => { + state.isLending = false + state.error = action.payload ?? action.error.message ?? 'Failed to lend camera' + }) + .addCase(returnCamera.pending, (state, action) => { + state.isReturning = action.meta.arg + state.error = null + state.message = null + }) + .addCase(returnCamera.fulfilled, (state, action) => { + state.isReturning = null + state.message = `Camera ${action.payload} returned successfully` + }) + .addCase(returnCamera.rejected, (state, action) => { + state.isReturning = null + state.error = action.payload ?? action.error.message ?? 'Failed to return camera' + }) + .addCase(saveCameraUpdates.pending, (state) => { + state.isUpdating = true + state.error = null + state.message = null + }) + .addCase(saveCameraUpdates.fulfilled, (state, action) => { + state.isUpdating = false + state.editModalCameraId = null + state.message = `Camera ${action.payload} updated successfully` + }) + .addCase(saveCameraUpdates.rejected, (state, action) => { + state.isUpdating = false + state.error = action.payload ?? action.error.message ?? 'Failed to update camera' + }) + .addCase(toggleCameraActivation.pending, (state, action) => { + state.isTogglingActivation = action.meta.arg.cameraId + state.error = null + state.message = null + }) + .addCase(toggleCameraActivation.fulfilled, (state, action) => { + state.isTogglingActivation = null + state.message = action.payload.nextActive + ? `Camera ${action.payload.cameraId} activated` + : `Camera ${action.payload.cameraId} deactivated` + }) + .addCase(toggleCameraActivation.rejected, (state, action) => { + state.isTogglingActivation = null + state.error = action.payload ?? action.error.message ?? 'Failed to update camera activation status' + }) + .addCase(fetchCameraAudit.pending, (state, action) => { + state.auditModalCameraId = action.meta.arg + state.auditLogs = [] + state.isAuditLoading = true + state.error = null + }) + .addCase(fetchCameraAudit.fulfilled, (state, action) => { + state.isAuditLoading = false + state.auditModalCameraId = action.payload.cameraId + state.auditLogs = action.payload.items + }) + .addCase(fetchCameraAudit.rejected, (state, action) => { + state.isAuditLoading = false + state.error = action.payload ?? action.error.message ?? 'Failed to load audit logs' + }) + }, +}) + +export const { + clearBanner, + openCreateModal, + closeCreateModal, + setCreateCameraId, + setCreateName, + setCreateLocation, + setCreateModel, + setCreateNotes, + setCreateIsActive, + openLendModal, + closeLendModal, + setLendBorrower, + setLendPurpose, + setLendDueDate, + openEditModal, + closeEditModal, + setEditName, + setEditLocation, + setEditModel, + setEditNotes, + setEditIsActive, + closeAuditModal, +} = cameraPageSlice.actions + +export default cameraPageSlice.reducer diff --git a/apps/web/src/store/index.ts b/apps/web/src/store/index.ts index 1acec9a..51fdba3 100644 --- a/apps/web/src/store/index.ts +++ b/apps/web/src/store/index.ts @@ -1,4 +1,5 @@ import { configureStore } from '@reduxjs/toolkit' +import cameraPageReducer from './camera-slice' import deviceReducer from './device-slice' import greetingReducer from './greeting-slice' @@ -6,6 +7,7 @@ export const store = configureStore({ reducer: { greeting: greetingReducer, device: deviceReducer, + cameraPage: cameraPageReducer, }, }) diff --git a/docs/camera-data-flow.md b/docs/camera-data-flow.md new file mode 100644 index 0000000..3400829 --- /dev/null +++ b/docs/camera-data-flow.md @@ -0,0 +1,100 @@ +# 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 +``` diff --git a/infra/postgres/init.sql b/infra/postgres/init.sql index ce08bab..c1169b2 100644 --- a/infra/postgres/init.sql +++ b/infra/postgres/init.sql @@ -4,3 +4,41 @@ CREATE TABLE IF NOT EXISTS device_current_state ( last_seen TIMESTAMPTZ NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); + +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() +); + +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() +); + +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() +); + +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() +); diff --git a/packages/contracts/src/camera.ts b/packages/contracts/src/camera.ts new file mode 100644 index 0000000..4648485 --- /dev/null +++ b/packages/contracts/src/camera.ts @@ -0,0 +1,128 @@ +import { z } from 'zod'; + +export const cameraHealthSchema = z.enum(['online', 'offline', 'unknown']); + +export const cameraDeviceSchema = z.object({ + cameraId: z.string().min(1), + name: z.string().min(1), + location: z.string().min(1), + model: z.string().min(1).nullable(), + notes: z.string().min(1).nullable(), + isActive: z.boolean(), +}); + +export const cameraDeviceUpdateSchema = z + .object({ + name: z.string().min(1).optional(), + location: z.string().min(1).optional(), + model: z.string().min(1).nullable().optional(), + notes: z.string().min(1).nullable().optional(), + isActive: z.boolean().optional(), + }) + .refine((value) => Object.keys(value).length > 0, { + message: 'At least one field is required', + }); + +export const cameraAuditActionSchema = z.enum([ + 'camera_created', + 'camera_lent', + 'camera_returned', + 'camera_updated', + 'camera_activated', + 'camera_deactivated', + 'camera_soft_deleted', +]); + +export const cameraAuditLogSchema = z.object({ + id: z.number().int().positive(), + cameraId: z.string().min(1), + action: cameraAuditActionSchema, + actor: z.string().min(1), + metadata: z.record(z.unknown()), + createdAt: z.string().datetime(), +}); + +export const cameraAuditLogListSchema = z.object({ + items: z.array(cameraAuditLogSchema), +}); + +export const cameraRuntimePayloadSchema = z.object({ + fps: z.number().int().nonnegative(), + bitrateKbps: z.number().int().positive(), + recording: z.boolean(), + scene: z.string().min(1), +}); + +export const cameraLatestStatusSchema = z.object({ + cameraId: z.string().min(1), + lastSeen: z.string().datetime().nullable(), + health: cameraHealthSchema, + payload: cameraRuntimePayloadSchema.nullable(), +}); + +export const cameraLendInfoSchema = z.object({ + borrower: z.string().min(1), + dueDate: z.string().datetime(), + purpose: z.string().min(1).optional(), + lentAt: z.string().datetime(), +}); + +export const cameraListItemSchema = z.object({ + cameraId: z.string().min(1), + name: z.string().min(1), + location: z.string().min(1), + model: z.string().min(1).nullable(), + notes: z.string().min(1).nullable(), + isActive: z.boolean(), + latest: cameraLatestStatusSchema, + lend: cameraLendInfoSchema.nullable(), +}); + +export const cameraListResponseSchema = z.object({ + items: z.array(cameraListItemSchema), +}); + +export const cameraLendRequestSchema = z.object({ + borrower: z.string().min(1), + dueDate: z.string().datetime(), + purpose: z.string().min(1).optional(), +}); + +export const cameraCreateRequestSchema = z.object({ + cameraId: z.string().min(1), + name: z.string().min(1), + location: z.string().min(1), + model: z.string().min(1).nullable().optional(), + notes: z.string().min(1).nullable().optional(), + isActive: z.boolean().optional(), +}); + +export const cameraReturnResponseSchema = z.object({ + cameraId: z.string().min(1), + returned: z.boolean(), +}); + +export const cameraSimulateIngestResponseSchema = z.object({ + accepted: z.boolean(), + writes: z.object({ + mongo: z.boolean(), + redis: z.boolean(), + postgres: z.boolean(), + }), + latest: cameraLatestStatusSchema, +}); + +export type CameraHealth = z.infer; +export type CameraDevice = z.infer; +export type CameraDeviceUpdate = z.infer; +export type CameraAuditAction = z.infer; +export type CameraAuditLog = z.infer; +export type CameraRuntimePayload = z.infer; +export type CameraLatestStatus = z.infer; +export type CameraLendInfo = z.infer; +export type CameraListItem = z.infer; +export type CameraListResponse = z.infer; +export type CameraLendRequest = z.infer; +export type CameraCreateRequest = z.infer; +export type CameraReturnResponse = z.infer; +export type CameraSimulateIngestResponse = z.infer; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 7703d2b..2d194e3 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -1,4 +1,5 @@ export * from './device-status.js'; +export * from './camera.js'; export * from './greeting.js'; export * from './presence.js'; export * from './telemetry.js';