TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import Redis from "ioredis";2import { config, log } from "./config";34let redis: Redis | null = null;56export function getRedis(): Redis {7 if (!redis) {8 redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 3, lazyConnect: false, enableOfflineQueue: true });9 redis.on("error", (e) => log.warn({ err: e.message }, "redis error"));10 }11 return redis;12}1314export const STREAM_EVENTS = "ws:events";15export const CHANNEL_LIVE = "ws:live";16export const STREAM_CHANGES = "ws:changes";1718/**19 * Publish a compact event payload to the stream (durable, replayable by stream id) and the pub/sub20 * channel (realtime). The stream id is embedded as `sid` so clients can resume with `since`.21 */22export async function publishEvent(payload: Record<string, unknown>): Promise<string | null> {23 const r = getRedis();24 try {25 const sid = await r.xadd(STREAM_EVENTS, "MAXLEN", "~", "20000", "*", "event", JSON.stringify(payload));26 const json = JSON.stringify({ ...payload, sid });27 // keep the stream entry consistent with what subscribers saw28 await r.publish(CHANNEL_LIVE, json);29 return sid ?? null;30 } catch (e) {31 log.warn({ err: (e as Error).message }, "publish failed");32 return null;33 }34}3536/** Engine heartbeat for the ops dashboard (`/api/v1/admin/ops`, `/api/ready`). */37export async function publishEngineStatus(status: Record<string, unknown>): Promise<void> {38 try {39 await getRedis().set("ws:engine:status", JSON.stringify({ ...status, at: new Date().toISOString(), pid: process.pid }), "EX", 30);40 } catch {41 // best effort42 }43}4445export async function publishChange(payload: Record<string, unknown>): Promise<void> {46 try {47 await getRedis().xadd(STREAM_CHANGES, "MAXLEN", "~", "5000", "*", "change", JSON.stringify(payload));48 } catch {49 // best effort50 }51}5253export async function closeRedis(): Promise<void> {54 if (redis) await redis.quit().catch(() => undefined);55 redis = null;56}57