import { eq } from 'drizzle-orm'; import { drizzle } from 'drizzle-orm/node-postgres'; import Redis from 'ioredis'; import pg from 'pg'; import type { DeviceStatusPair } from 'contracts'; import { env } from './env.js'; import { deviceCurrentState } from './db/schema.js'; import { emptyStatus, redisKey, toDeviceStatus } from './status.js'; export type DeviceStores = { readStatus(deviceId: string): Promise; close(): Promise; }; type RedisSnapshot = { lastSeen: string; payload: Record; }; export function createDeviceStores(): DeviceStores { const pool = new pg.Pool({ connectionString: env.databaseUrl }); const db = drizzle(pool); const redis = new Redis(env.redisUrl, { maxRetriesPerRequest: 1, lazyConnect: true }); return { async readStatus(deviceId: string): Promise { const pair = emptyStatus(deviceId); try { const rows = await db .select() .from(deviceCurrentState) .where(eq(deviceCurrentState.deviceId, deviceId)) .limit(1); const row = rows[0]; if (row) { pair.postgres = toDeviceStatus({ deviceId, lastSeen: row.lastSeen, payload: row.payload, source: 'postgres', }); } } catch (error) { console.error('postgres status read failed', error); } try { if (redis.status === 'wait') { await redis.connect(); } const raw = await redis.get(redisKey(deviceId)); if (raw) { const snapshot = JSON.parse(raw) as RedisSnapshot; pair.redis = toDeviceStatus({ deviceId, lastSeen: new Date(snapshot.lastSeen), payload: snapshot.payload ?? {}, source: 'redis', }); } } catch (error) { console.error('redis status read failed', error); } return pair; }, async close() { await Promise.allSettled([pool.end(), redis.quit()]); }, }; } export function createMemoryStores( seed: Partial> = {}, ): DeviceStores { return { async readStatus(deviceId: string) { return seed[deviceId] ?? emptyStatus(deviceId); }, async close() { return; }, }; }