Phase 1: data pipeline (Postgres+Redis with in-memory fallback, mock telemetry publisher, WebSocket fan-out, hub /debug page)

- packages/db: postgres.js + ioredis wrapper, StateStore interface with
  InMemory + Postgres implementations, schema migration runner
- apps/api: refactored to use @kerbal-rt/db; live WebSocket hub subscribes
  to store changes and broadcasts to all clients
- apps/tools/mock-telemetry: Node script that generates realistic KSP
  state and POSTs to /api/v1/ingest at 1Hz (uses same keplerian math
  as the live-map renderer)
- apps/hub: new /debug page that connects to /api/v1/live and shows
  the live state
- ksp/README.md: documents Phase 1c (real kRPC bridge) and the two
  implementation options
- Tests: 17 total (5 kepler math, 5 db store, 5 API health/state,
  2 API WebSocket fan-out)

End-to-end verified: mock publisher → API → hub /debug page,
11 snapshots/5s over WebSocket, vessels advancing in mean anomaly.

packages/db/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "@kerbal-rt/db",
+  "version": "0.1.0",
+  "private": true,
+  "description": "Persistence layer: Postgres + Timescale (telemetry) and Redis (pubsub + cache)",
+  "type": "module",
+  "main": "./src/index.ts",
+  "types": "./src/index.ts",
+  "exports": {
+    ".": "./src/index.ts",
+    "./schema": "./src/schema.ts",
+    "./migrate": "./src/migrate.ts"
+  },
+  "scripts": {
+    "typecheck": "tsc --noEmit",
+    "lint": "echo 'no linter yet'",
+    "test": "vitest run",
+    "test:watch": "vitest",
+    "build": "tsc"
+  },
+  "dependencies": {
+    "@kerbal-rt/shared-types": "workspace:*",
+    "ioredis": "^5.4.1",
+    "postgres": "^3.4.4"
+  },
+  "devDependencies": {
+    "typescript": "^5.6.2",
+    "vitest": "^2.1.1"
+  }
+}

packages/db/src/config.ts

@@ -0,0 +1,45 @@
+/**
+ * Configuration helpers for the db package.
+ *
+ * Reads from environment variables. Throws clearly if a required var
+ * is missing.
+ */
+
+export interface DbConfig {
+  /** Postgres connection URL, e.g. postgres://user:pass@host:5432/db */
+  databaseUrl: string;
+  /** Redis connection URL, e.g. redis://host:6379 */
+  redisUrl: string;
+  /** When true, callers should use the in-memory fallback instead. */
+  useInMemory: boolean;
+}
+
+const REQUIRED_FOR_DB = ['DATABASE_URL', 'REDIS_URL'] as const;
+
+export function loadConfig(env: NodeJS.ProcessEnv = process.env): DbConfig {
+  const databaseUrl = env.DATABASE_URL;
+  const redisUrl = env.REDIS_URL;
+  // Allow opt-out for dev (e.g. when Docker isn't available)
+  const forceInMemory = env.USE_IN_MEMORY === '1' || env.USE_IN_MEMORY === 'true';
+
+  if (forceInMemory) {
+    return { databaseUrl: '', redisUrl: '', useInMemory: true };
+  }
+
+  const missing = REQUIRED_FOR_DB.filter((k) => !env[k]);
+  if (missing.length > 0) {
+    // Don't crash on import — let the caller decide. The API's
+    // bootstrap() logs a warning and falls back to in-memory mode.
+    return {
+      databaseUrl: databaseUrl ?? '',
+      redisUrl: redisUrl ?? '',
+      useInMemory: true,
+    };
+  }
+
+  return {
+    databaseUrl: databaseUrl!,
+    redisUrl: redisUrl!,
+    useInMemory: false,
+  };
+}

packages/db/src/index.ts

@@ -0,0 +1,94 @@
+/**
+ * Public API of @kerbal-rt/db.
+ *
+ * Callers (the API, the mock-telemetry publisher, the kRPC bridge) get
+ * a single `createStore()` function that returns a fully-wired
+ * StateStore + the underlying clients. This way they don't need to
+ * know whether they're in dev (in-memory) or prod (Postgres+Redis).
+ */
+export { loadConfig, type DbConfig } from './config.js';
+export { createPgClient, type PgClient } from './postgres.js';
+export { createRedis, type RedisPair } from './redis.js';
+export { runMigrations } from './migrate.js';
+export { InMemoryStateStore, PgStateStore, type StateStore } from './store.js';
+
+import { loadConfig, type DbConfig } from './config.js';
+import { createPgClient, type PgClient } from './postgres.js';
+import { createRedis, type RedisPair } from './redis.js';
+import { runMigrations } from './migrate.js';
+import { InMemoryStateStore, PgStateStore, type StateStore } from './store.js';
+
+export interface WiredStore {
+  store: StateStore;
+  config: DbConfig;
+  /** Underlying clients (null in in-memory mode). */
+  pg: PgClient | null;
+  redis: RedisPair | null;
+  /** Redis pub/sub channel name for snapshot notifications. */
+  channel: string;
+  /** Tear down all clients. */
+  close(): Promise<void>;
+}
+
+const SNAPSHOT_CHANNEL = 'kerbal-rt:snapshots';
+
+export async function createStore(env: NodeJS.ProcessEnv = process.env): Promise<WiredStore> {
+  const config = loadConfig(env);
+
+  if (config.useInMemory) {
+    // Dev mode — no DB required.
+    return {
+      store: new InMemoryStateStore(),
+      config,
+      pg: null,
+      redis: null,
+      channel: SNAPSHOT_CHANNEL,
+      async close() {
+        /* nothing to close */
+      },
+    };
+  }
+
+  // Production mode — Postgres + Redis.
+  const pg = createPgClient(config.databaseUrl);
+  const redis = createRedis(config.redisUrl);
+
+  // Run schema migrations on boot (idempotent).
+  const migrationResult = await runMigrations(pg.sql);
+  for (const m of migrationResult.applied) {
+    // eslint-disable-next-line no-console
+    console.log(`[db] migration applied: ${m}`);
+  }
+
+  // Snapshot fan-out over Redis. The "publish" side is one-liner;
+  // the "subscribe" side needs a dedicated connection (handled by
+  // the RedisPair wrapper).
+  const store = new PgStateStore({
+    sql: pg.sql,
+    async publish(payload) {
+      await redis.client.publish(SNAPSHOT_CHANNEL, payload);
+    },
+    async redisSubscribe(handler) {
+      await redis.subscriber.subscribe(SNAPSHOT_CHANNEL);
+      const onMessage = (channel: string, message: string) => {
+        if (channel === SNAPSHOT_CHANNEL) handler(message);
+      };
+      redis.subscriber.on('message', onMessage);
+      return async () => {
+        redis.subscriber.off('message', onMessage);
+        await redis.subscriber.unsubscribe(SNAPSHOT_CHANNEL).catch(() => {});
+      };
+    },
+  });
+
+  return {
+    store,
+    config,
+    pg,
+    redis,
+    channel: SNAPSHOT_CHANNEL,
+    async close() {
+      await Promise.all([pg.close().catch(() => {}), redis.close().catch(() => {})]);
+    },
+  };
+}

packages/db/src/migrate.ts

@@ -0,0 +1,48 @@
+/**
+ * Apply the schema migrations to a Postgres database.
+ *
+ * Reads the SQL files from infra/init-sql/ and runs them in lexical
+ * order. Idempotent — safe to run on every boot.
+ *
+ * In production you'd use a proper migration tool (drizzle-kit,
+ * prisma, knex, sqlx, etc.). For Phase 1 we just want a simple,
+ * zero-deps way to get the schema into a fresh database.
+ */
+import { readFile, readdir } from 'node:fs/promises';
+import { join, dirname, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import type { Sql } from 'postgres';
+
+const INIT_SQL_DEFAULT = '../../../infra/init-sql';
+
+export async function runMigrations(
+  sql: Sql,
+  initDir: string = INIT_SQL_DEFAULT,
+): Promise<{ applied: string[] }> {
+  const dir = resolve(dirname(fileURLToPath(import.meta.url)), initDir);
+  const entries = (await readdir(dir)).filter((f) => f.endsWith('.sql')).sort();
+
+  const applied: string[] = [];
+  for (const file of entries) {
+    const path = join(dir, file);
+    const content = await readFile(path, 'utf8');
+    try {
+      await sql.unsafe(content);
+      applied.push(file);
+    } catch (err) {
+      // Some errors are non-fatal (e.g. "extension already exists" race),
+      // but anything else should fail loudly.
+      if (isBenignError(err)) {
+        applied.push(`${file} (skipped: ${(err as Error).message.split('\n')[0]})`);
+        continue;
+      }
+      throw new Error(`Migration ${file} failed: ${(err as Error).message}`);
+    }
+  }
+  return { applied };
+}
+
+function isBenignError(err: unknown): boolean {
+  const msg = String((err as Error).message ?? err);
+  return /already exists/i.test(msg) || /duplicate key/i.test(msg);
+}

packages/db/src/postgres.ts

@@ -0,0 +1,31 @@
+/**
+ * Postgres + Timescale connection wrapper.
+ *
+ * Uses `postgres.js` for queries and connection pooling. The schema
+ * lives in ../../infra/init-sql/01-schema.sql and is applied by
+ * `migrate.ts` (and automatically on first boot by the API's
+ * bootstrap helper).
+ */
+import postgres from 'postgres';
+import type { Sql } from 'postgres';
+
+export interface PgClient {
+  sql: Sql;
+  close(): Promise<void>;
+}
+
+export function createPgClient(url: string): PgClient {
+  const sql = postgres(url, {
+    max: 10,
+    idle_timeout: 30,
+    connect_timeout: 10,
+    // Reduce log noise; we use console for the rest of the app
+    onnotice: () => {},
+  });
+  return {
+    sql,
+    async close() {
+      await sql.end({ timeout: 5 });
+    },
+  };
+}

packages/db/src/redis.ts

@@ -0,0 +1,33 @@
+/**
+ * Redis connection wrapper. Provides one client for normal commands
+ * and a separate one for pub/sub (Redis requires dedicated connections
+ * for subscribe mode).
+ */
+import Redis from 'ioredis';
+
+export interface RedisPair {
+  /** For GET/SET/PUBLISH etc. */
+  client: Redis;
+  /** For SUBSCRIBE only — must not be used for other commands. */
+  subscriber: Redis;
+  close(): Promise<void>;
+}
+
+export function createRedis(url: string): RedisPair {
+  const client = new Redis(url, {
+    lazyConnect: false,
+    maxRetriesPerRequest: 3,
+    enableReadyCheck: true,
+  });
+  const subscriber = new Redis(url, {
+    lazyConnect: false,
+    maxRetriesPerRequest: null, // subscribers reconnect forever
+  });
+  return {
+    client,
+    subscriber,
+    async close() {
+      await Promise.all([client.quit().catch(() => {}), subscriber.quit().catch(() => {})]);
+    },
+  };
+}

packages/db/src/store.ts

@@ -0,0 +1,234 @@
+/**
+ * StateStore — the interface every API caller uses to read/write
+ * the current universe state. Two implementations:
+ *
+ *  - InMemoryStateStore  : in-process; for dev when Postgres isn't up
+ *  - PgStateStore        : Postgres + Timescale; for production
+ *
+ * Both implement the same interface so the API code is identical.
+ */
+import type { CelestialBody, UniverseSnapshot, Vessel } from '@kerbal-rt/shared-types';
+import type { Sql } from 'postgres';
+
+// ─── Interface ─────────────────────────────────────────────────────────────
+
+export interface StateStore {
+  /** Read the most recent snapshot. Returns null if none yet. */
+  latestSnapshot(): Promise<UniverseSnapshot | null>;
+
+  /** List celestial bodies (catalog is small, kept in memory). */
+  listBodies(): Promise<CelestialBody[]>;
+
+  /** List currently-known vessels. */
+  listVessels(): Promise<Vessel[]>;
+
+  /** Look up a single vessel by ID. */
+  getVessel(id: string): Promise<Vessel | null>;
+
+  /**
+   * Persist a new snapshot. Returns the row count for telemetry
+   * (rows in telemetry_snapshots, vessels upserted, etc.).
+   * Publishes a notification on the "snapshots" channel so any
+   * WebSocket subscribers (in this or another API instance) see it.
+   */
+  applySnapshot(snap: UniverseSnapshot): Promise<{ rowsInserted: number }>;
+
+  /** Subscribe to snapshot updates. Yields the snapshot. */
+  subscribe(handler: (snap: UniverseSnapshot) => void): Promise<() => Promise<void>>;
+}
+
+// ─── In-memory implementation ──────────────────────────────────────────────
+
+export class InMemoryStateStore implements StateStore {
+  private snapshot: UniverseSnapshot | null = null;
+  private vesselsById = new Map<string, Vessel>();
+  private bodiesById = new Map<string, CelestialBody>();
+  private subscribers = new Set<(snap: UniverseSnapshot) => void>();
+
+  async latestSnapshot() {
+    return this.snapshot;
+  }
+  async listBodies() {
+    return Array.from(this.bodiesById.values());
+  }
+  async listVessels() {
+    return Array.from(this.vesselsById.values());
+  }
+  async getVessel(id: string) {
+    return this.vesselsById.get(id) ?? null;
+  }
+  async applySnapshot(snap: UniverseSnapshot) {
+    this.snapshot = snap;
+    this.vesselsById = new Map(snap.vessels.map((v) => [v.id, v]));
+    this.bodiesById = new Map(snap.bodies.map((b) => [b.id, b]));
+    // Notify all subscribers asynchronously
+    queueMicrotask(() => {
+      for (const sub of this.subscribers) sub(snap);
+    });
+    return { rowsInserted: 1 };
+  }
+  async subscribe(handler: (snap: UniverseSnapshot) => void) {
+    this.subscribers.add(handler);
+    return async () => {
+      this.subscribers.delete(handler);
+    };
+  }
+}
+
+// ─── Postgres implementation ───────────────────────────────────────────────
+
+interface PgStateStoreOpts {
+  sql: Sql;
+  /** Function to publish a JSON-serialized snapshot to Redis. */
+  publish: (payload: string) => Promise<void>;
+  /** Function to subscribe to snapshots over Redis. */
+  redisSubscribe: (handler: (payload: string) => void) => Promise<() => Promise<void>>;
+}
+
+export class PgStateStore implements StateStore {
+  private latest: UniverseSnapshot | null = null;
+  private latestCacheAt = 0;
+  private static readonly CACHE_TTL_MS = 1000; // 1s hot cache
+
+  constructor(private opts: PgStateStoreOpts) {}
+
+  async latestSnapshot(): Promise<UniverseSnapshot | null> {
+    // Hot cache: avoids hitting Postgres on every read.
+    if (this.latest && Date.now() - this.latestCacheAt < PgStateStore.CACHE_TTL_MS) {
+      return this.latest;
+    }
+    const rows = await this.opts.sql<{ payload: UniverseSnapshot }[]>`
+      SELECT payload FROM telemetry_snapshots
+      ORDER BY ts DESC LIMIT 1
+    `;
+    if (rows.length === 0) return null;
+    this.latest = rows[0].payload;
+    this.latestCacheAt = Date.now();
+    return this.latest;
+  }
+
+  async listBodies(): Promise<CelestialBody[]> {
+    // Body catalog is small (~17 rows) and changes rarely.
+    // Cache it for 30s.
+    if (this._bodyCache && Date.now() - this._bodyCacheAt < PgStateStore.BODY_CACHE_TTL_MS) {
+      return this._bodyCache;
+    }
+    const rows = await this.opts.sql<
+      {
+        id: string;
+        name: string;
+        kind: string;
+        parent_id: string | null;
+        radius: number;
+        sphere_of_influence: number;
+        gravitational_parameter: number;
+        rotation_period: number;
+        axial_tilt: number;
+        elements: CelestialBody['orbit'];
+      }[]
+    >`
+      SELECT id, name, kind, parent_id, radius, sphere_of_influence,
+             gravitational_parameter, rotation_period, axial_tilt, elements
+      FROM bodies
+      ORDER BY id
+    `;
+    const bodies: CelestialBody[] = rows.map((r) => ({
+      id: r.id,
+      name: r.name,
+      kind: r.kind as CelestialBody['kind'],
+      parentId: r.parent_id,
+      radius: r.radius,
+      sphereOfInfluence: r.sphere_of_influence,
+      gravitationalParameter: r.gravitational_parameter,
+      rotationPeriod: r.rotation_period,
+      axialTilt: r.axial_tilt,
+      orbit: r.elements,
+    }));
+    this._bodyCache = bodies;
+    this._bodyCacheAt = Date.now();
+    return bodies;
+  }
+
+  async listVessels(): Promise<Vessel[]> {
+    const snap = await this.latestSnapshot();
+    return snap?.vessels ?? [];
+  }
+
+  async getVessel(id: string): Promise<Vessel | null> {
+    const vessels = await this.listVessels();
+    return vessels.find((v) => v.id === id) ?? null;
+  }
+
+  async applySnapshot(snap: UniverseSnapshot): Promise<{ rowsInserted: number }> {
+    // 1) Persist the raw snapshot to Timescale (time-series).
+    //    We also keep a "current state" copy in `vessels` for fast
+    //    single-row lookups, and a `bodies` upsert so the catalog
+    //    stays in sync with whatever the bridge sends.
+    await this.opts.sql`
+      INSERT INTO telemetry_snapshots (ts, ut, payload)
+      VALUES (now(), ${snap.ut}, ${this.opts.sql.json(JSON.parse(JSON.stringify(snap)))})
+    `;
+
+    // Upsert vessels (so vessel id is stable across snapshots).
+    for (const v of snap.vessels) {
+      await this.opts.sql`
+        INSERT INTO vessels (id, name, type, owner, status, created_at, retired_at)
+        VALUES (${v.id}, ${v.name}, ${v.type}, ${v.owner}, ${v.status},
+                ${v.createdAt}::timestamptz, ${v.retiredAt}::timestamptz)
+        ON CONFLICT (id) DO UPDATE SET
+          name = EXCLUDED.name,
+          type = EXCLUDED.type,
+          owner = EXCLUDED.owner,
+          status = EXCLUDED.status,
+          retired_at = EXCLUDED.retired_at
+      `;
+    }
+
+    // Upsert bodies (catalog).
+    for (const b of snap.bodies) {
+      await this.opts.sql`
+        INSERT INTO bodies (id, name, kind, parent_id, radius, sphere_of_influence,
+                            gravitational_parameter, rotation_period, axial_tilt, elements)
+        VALUES (${b.id}, ${b.name}, ${b.kind}, ${b.parentId}, ${b.radius},
+                ${b.sphereOfInfluence}, ${b.gravitationalParameter},
+                ${b.rotationPeriod}, ${b.axialTilt}, ${this.opts.sql.json(JSON.parse(JSON.stringify(b.orbit)))})
+        ON CONFLICT (id) DO UPDATE SET
+          name = EXCLUDED.name,
+          kind = EXCLUDED.kind,
+          parent_id = EXCLUDED.parent_id,
+          radius = EXCLUDED.radius,
+          sphere_of_influence = EXCLUDED.sphere_of_influence,
+          gravitational_parameter = EXCLUDED.gravitational_parameter,
+          rotation_period = EXCLUDED.rotation_period,
+          axial_tilt = EXCLUDED.axial_tilt,
+          elements = EXCLUDED.elements
+      `;
+    }
+
+    // 2) Bust the hot cache.
+    this.latest = snap;
+    this.latestCacheAt = Date.now();
+    this._bodyCache = null;
+
+    // 3) Publish to Redis so other API instances + WS clients see it.
+    await this.opts.publish(JSON.stringify(snap));
+
+    return { rowsInserted: 1 };
+  }
+
+  async subscribe(handler: (snap: UniverseSnapshot) => void) {
+    return this.opts.redisSubscribe((payload) => {
+      try {
+        const snap = JSON.parse(payload) as UniverseSnapshot;
+        handler(snap);
+      } catch {
+        // ignore malformed messages
+      }
+    });
+  }
+
+  // ── private ──
+  private _bodyCache: CelestialBody[] | null = null;
+  private _bodyCacheAt = 0;
+  private static readonly BODY_CACHE_TTL_MS = 30_000;
+}

packages/db/tests/store.test.ts

@@ -0,0 +1,106 @@
+import { describe, it, expect } from 'vitest';
+import { InMemoryStateStore } from '../src/store.js';
+import type { UniverseSnapshot } from '@kerbal-rt/shared-types';
+
+function makeSnapshot(ut: number, vesselId = 'v-1'): UniverseSnapshot {
+  return {
+    ut,
+    capturedAt: new Date().toISOString(),
+    activeVesselId: vesselId,
+    bodies: [
+      {
+        id: 'kerbin',
+        name: 'Kerbin',
+        kind: 'planet',
+        parentId: 'kerbol',
+        radius: 600_000,
+        sphereOfInfluence: 84_159_286,
+        gravitationalParameter: 3.5316e12,
+        rotationPeriod: 21_600,
+        axialTilt: 0,
+        orbit: {
+          semiMajorAxis: 13_599_840_256,
+          eccentricity: 0,
+          inclination: 0,
+          longitudeOfAscendingNode: 0,
+          argumentOfPeriapsis: 0,
+          meanAnomalyAtEpoch: 0,
+          epoch: 0,
+        },
+      },
+    ],
+    vessels: [
+      {
+        id: vesselId,
+        name: 'Test',
+        type: 'Probe',
+        owner: 'KASA',
+        situation: 'ORBITING',
+        status: 'ACTIVE',
+        orbit: {
+          semiMajorAxis: 7e6,
+          eccentricity: 0,
+          inclination: 0,
+          longitudeOfAscendingNode: 0,
+          argumentOfPeriapsis: 0,
+          meanAnomalyAtEpoch: 0,
+          epoch: 0,
+        },
+        referenceBodyId: 'kerbin',
+        createdAt: '2026-01-01T00:00:00Z',
+        retiredAt: null,
+      },
+    ],
+    groundStations: [],
+  };
+}
+
+describe('InMemoryStateStore', () => {
+  it('starts empty', async () => {
+    const store = new InMemoryStateStore();
+    expect(await store.latestSnapshot()).toBeNull();
+    expect(await store.listVessels()).toEqual([]);
+  });
+
+  it('stores and returns snapshots', async () => {
+    const store = new InMemoryStateStore();
+    await store.applySnapshot(makeSnapshot(100));
+    const snap = await store.latestSnapshot();
+    expect(snap).not.toBeNull();
+    expect(snap!.ut).toBe(100);
+  });
+
+  it('looks up vessels by id', async () => {
+    const store = new InMemoryStateStore();
+    await store.applySnapshot(makeSnapshot(1, 'kasa-1'));
+    const v = await store.getVessel('kasa-1');
+    expect(v?.name).toBe('Test');
+    expect(await store.getVessel('missing')).toBeNull();
+  });
+
+  it('overwrites on subsequent snapshots', async () => {
+    const store = new InMemoryStateStore();
+    await store.applySnapshot(makeSnapshot(1));
+    await store.applySnapshot(makeSnapshot(2));
+    expect((await store.latestSnapshot())!.ut).toBe(2);
+  });
+
+  it('notifies subscribers when a snapshot is applied', async () => {
+    const store = new InMemoryStateStore();
+    const seen: number[] = [];
+    const unsub = await store.subscribe((snap) => seen.push(snap.ut));
+
+    await store.applySnapshot(makeSnapshot(1));
+    // allow microtask to flush
+    await new Promise((r) => setTimeout(r, 10));
+    await store.applySnapshot(makeSnapshot(2));
+    await new Promise((r) => setTimeout(r, 10));
+
+    expect(seen).toEqual([1, 2]);
+
+    await unsub();
+    await store.applySnapshot(makeSnapshot(3));
+    await new Promise((r) => setTimeout(r, 10));
+    expect(seen).toEqual([1, 2]); // unsub'd
+  });
+});

packages/db/tsconfig.json

@@ -0,0 +1,10 @@
+{
+  "extends": "../../tsconfig.base.json",
+  "compilerOptions": {
+    "outDir": "./dist",
+    "rootDir": "./src",
+    "module": "ESNext",
+    "moduleResolution": "Bundler"
+  },
+  "include": ["src/**/*"]
+}

packages/db/vitest.config.ts

@@ -0,0 +1,8 @@
+import { defineConfig } from 'vitest/config';
+
+export default defineConfig({
+  test: {
+    include: ['tests/**/*.test.ts'],
+    environment: 'node',
+  },
+});