Add Session Explorer API

apps/api/package.json

@@ -78,6 +78,7 @@
     "better-sqlite3": "^12.8.0",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.15.1",
+    "cluster-key-slot": "^1.1.2",
     "cron": "^4.4.0",
     "fastify": "^5.12.5",
     "hnswlib-node": "^3.0.0",

apps/api/src/app.module.ts

@@ -27,6 +27,7 @@ import { WorkspaceAuthModule } from './auth/workspace-auth.module';
 import { McpModule } from './mcp/mcp.module';
 import { MetricForecastingModule } from './metric-forecasting/metric-forecasting.module';
 import { ScalingReadinessModule } from './scaling-readiness/scaling-readiness.module';
+import { SessionExplorerModule } from './session-explorer/session-explorer.module';
 import { InferenceLatencyModule } from './inference-latency/inference-latency.module';
 import { CliModule } from './cli/cli.module';
 import { PosthogProxyModule } from './posthog-proxy/posthog-proxy.module';
@@ -223,6 +224,7 @@ const baseImports = [
   MigrationModule,
   MetricForecastingModule,
   ScalingReadinessModule,
+  SessionExplorerModule,
   InferenceLatencyModule,
   CliModule,
   PosthogProxyModule,

apps/api/src/common/interfaces/storage-port.interface.ts

@@ -39,6 +39,7 @@ import type {
   ReadinessDimension,
   ReadinessDimensionKey,
   ScalingReadinessSettings,
+  SessionNamespace,
 } from '@betterdb/shared';
 export type { MetricForecastSettings, MetricKind } from '@betterdb/shared';
 export type { StoredCveDataset, CveScanResult } from '@betterdb/shared';
@@ -606,6 +607,11 @@ export interface StoragePort {
   getScalingReadinessSettings(connectionId: string): Promise<ScalingReadinessSettings | null>;
   saveScalingReadinessSettings(settings: ScalingReadinessSettings): Promise<ScalingReadinessSettings>;
 
+  listSessionNamespaces(connectionId: string): Promise<SessionNamespace[]>;
+  getSessionNamespace(connectionId: string, id: string): Promise<SessionNamespace | null>;
+  saveSessionNamespace(ns: SessionNamespace): Promise<SessionNamespace>;
+  deleteSessionNamespace(connectionId: string, id: string): Promise<boolean>;
+
   // Command Stats Sample Methods - connectionId required for writes
   saveCommandStatsSamples(
     samples: Omit<StoredCommandStatsSample, 'id' | 'connectionId'>[],

apps/api/src/session-explorer/__tests__/fake-valkey.ts

@@ -0,0 +1,254 @@
+import type Valkey from 'iovalkey';
+import type { DatabasePort } from '../../common/interfaces/database-port.interface';
+import type { ClusterNode } from '../../common/types/metrics.types';
+
+export interface FakeEntry {
+  type: string;
+  pttl?: number;
+  memory?: number | null;
+  jsonMemory?: number | null;
+  idle?: number;
+  value?: string | Buffer | Array<string | Buffer> | Record<string, string | Buffer>;
+}
+
+const WRITE_COMMANDS = [
+  'set',
+  'del',
+  'expire',
+  'persist',
+  'unlink',
+  'hset',
+  'lpush',
+  'keys',
+  'flushall',
+  'flushdb',
+];
+
+const WRITE_CALLS = ['JSON.SET', 'JSON.DEL', 'SET', 'DEL', 'EXPIRE', 'PERSIST', 'UNLINK', 'KEYS'];
+
+const READ_COMMANDS = [
+  'type',
+  'pttl',
+  'memory',
+  'object',
+  'strlen',
+  'llen',
+  'zcard',
+  'getBuffer',
+  'getrangeBuffer',
+  'lrangeBuffer',
+  'hgetallBuffer',
+  'hscanBuffer',
+  'zrangeBuffer',
+  'xrangeBuffer',
+  'callBuffer',
+  'call',
+];
+
+const KEY_INDEPENDENT = new Set(['memory', 'object']);
+
+export class FakeValkey {
+  calls: Array<{ name: string; args: unknown[] }> = [];
+  pipelineExecs = 0;
+  scanPages?: string[][];
+  connectError?: Error;
+  disconnected = false;
+  options: Record<string, unknown>;
+  failKeys = new Map<string, string>();
+  movedKeys = new Set<string>();
+
+  constructor(
+    public store: Map<string, FakeEntry> = new Map(),
+    options: Record<string, unknown> = { host: 'seed', port: 7000 },
+  ) {
+    this.options = options;
+    for (const name of WRITE_COMMANDS) {
+      (this as unknown as Record<string, unknown>)[name] = () => {
+        throw new Error(`write command ${name} issued`);
+      };
+    }
+    for (const name of READ_COMMANDS) {
+      (this as unknown as Record<string, unknown>)[name] = async (...args: unknown[]) => {
+        this.record(name, args);
+        return this.run(name, args);
+      };
+    }
+  }
+
+  private record(name: string, args: unknown[]) {
+    this.calls.push({ name, args });
+  }
+
+  callsOf(name: string) {
+    return this.calls.filter((c) => c.name === name);
+  }
+
+  get directReads() {
+    return this.calls.filter((c) => READ_COMMANDS.includes(c.name));
+  }
+
+  on() {
+    return this;
+  }
+
+  async connect() {
+    this.record('connect', []);
+    if (this.connectError) throw this.connectError;
+  }
+
+  disconnect() {
+    this.disconnected = true;
+  }
+
+  async quit() {
+    this.disconnected = true;
+    return 'OK';
+  }
+
+  async scan(cursor: string, ...args: unknown[]): Promise<[string, string[]]> {
+    this.record('scan', [cursor, ...args]);
+    if (this.scanPages) {
+      const index = Number(cursor);
+      const page = this.scanPages[index] ?? [];
+      const next = index + 1 < this.scanPages.length ? String(index + 1) : '0';
+      return [next, page];
+    }
+    return ['0', [...this.store.keys()]];
+  }
+
+  private keyOf(name: string, args: unknown[]): string {
+    if (KEY_INDEPENDENT.has(name)) return args[1] as string;
+    if (name === 'call' || name === 'callBuffer') {
+      return (args[0] === 'JSON.DEBUG' ? args[2] : args[1]) as string;
+    }
+    return args[0] as string;
+  }
+
+  private run(name: string, args: unknown[]): unknown {
+    const key = this.keyOf(name, args);
+    if (this.movedKeys.has(key)) throw new Error('MOVED 1 10.0.0.9:7000');
+    if (this.failKeys.has(key) && !['type', 'pttl', 'memory', 'object'].includes(name)) {
+      throw new Error(this.failKeys.get(key));
+    }
+    const e = this.store.get(key);
+    const raw = e?.value;
+    switch (name) {
+      case 'type':
+        return e ? e.type : 'none';
+      case 'pttl':
+        return e ? (e.pttl ?? -1) : -2;
+      case 'memory':
+        if (e && e.memory === null) throw new Error('NOPERM memory');
+        return e ? (e.memory ?? 100) : null;
+      case 'object':
+        if (!e) throw new Error('ERR no such key');
+        return e.idle ?? 0;
+      case 'strlen':
+        return e ? Buffer.from(raw as string | Buffer).length : 0;
+      case 'llen':
+        return e ? (raw as unknown[]).length : 0;
+      case 'zcard':
+        return e ? (raw as unknown[]).length / 2 : 0;
+      case 'getBuffer':
+        return e ? Buffer.from(raw as string | Buffer) : null;
+      case 'getrangeBuffer': {
+        const [, start, end] = args as [string, number, number];
+        return e ? Buffer.from(raw as string | Buffer).subarray(start, end + 1) : Buffer.alloc(0);
+      }
+      case 'lrangeBuffer': {
+        const [, start, stop] = args as [string, number, number];
+        return ((raw as Array<string | Buffer>) ?? []).slice(start, stop + 1).map((v) => Buffer.from(v));
+      }
+      case 'hgetallBuffer':
+        return Object.fromEntries(
+          Object.entries((raw as Record<string, string>) ?? {}).map(([k, v]) => [k, Buffer.from(v)]),
+        );
+      case 'hscanBuffer':
+        return [
+          Buffer.from('0'),
+          Object.entries((raw as Record<string, string>) ?? {}).flatMap(([k, v]) => [
+            Buffer.from(k),
+            Buffer.from(v),
+          ]),
+        ];
+      case 'zrangeBuffer': {
+        const [, start, stop] = args as [string, number, string];
+        return ((raw as string[]) ?? [])
+          .slice(start * 2, (Number(stop) + 1) * 2)
+          .map((v) => Buffer.from(v));
+      }
+      case 'xrangeBuffer':
+        return [];
+      case 'call':
+      case 'callBuffer': {
+        const command = String(args[0]).toUpperCase();
+        if (WRITE_CALLS.includes(command)) throw new Error(`write command ${command} issued`);
+        if (command === 'JSON.GET') return e ? Buffer.from(raw as string) : null;
+        if (command === 'JSON.DEBUG' && args[1] === 'MEMORY') {
+          if (!e || e.jsonMemory === null || e.jsonMemory === undefined) {
+            throw new Error('ERR unknown command');
+          }
+          return e.jsonMemory;
+        }
+        throw new Error(`unexpected call ${command}`);
+      }
+      default:
+        throw new Error(`unexpected command ${name}`);
+    }
+  }
+
+  pipeline() {
+    const ops: Array<{ name: string; args: unknown[] }> = [];
+    const chain: Record<string, unknown> = {};
+    for (const name of READ_COMMANDS) {
+      chain[name] = (...args: unknown[]) => {
+        this.record(`pipeline.${name}`, args);
+        ops.push({ name, args });
+        return chain;
+      };
+    }
+    for (const name of WRITE_COMMANDS) {
+      chain[name] = () => {
+        throw new Error(`write command ${name} issued in pipeline`);
+      };
+    }
+    chain.exec = async () => {
+      this.pipelineExecs++;
+      return ops.map(({ name, args }) => {
+        try {
+          return [null, this.run(name, args)];
+        } catch (err) {
+          return [err, null];
+        }
+      });
+    };
+    return chain;
+  }
+}
+
+export function masterNode(
+  id: string,
+  address: string,
+  extraFlags: string[] = [],
+  slots: number[][] = [[0, 16383]],
+): ClusterNode {
+  return {
+    id,
+    address: `${address}@17000`,
+    flags: ['master', ...extraFlags],
+    master: '-',
+    pingSent: 0,
+    pongReceived: 0,
+    configEpoch: 0,
+    linkState: 'connected',
+    slots,
+  };
+}
+
+export function fakeDb(parent: FakeValkey, cluster?: { nodes: ClusterNode[] }): DatabasePort {
+  return {
+    getClient: () => parent as unknown as Valkey,
+    getCapabilities: () => ({ clusterEnabled: cluster !== undefined }),
+    getClusterNodes: async () => cluster?.nodes ?? [],
+  } as unknown as DatabasePort;
+}

apps/api/src/session-explorer/__tests__/namespace-storage-contract.ts

@@ -0,0 +1,81 @@
+import type { CustomSchema, SessionNamespace } from '@betterdb/shared';
+import type { StoragePort } from '../../common/interfaces/storage-port.interface';
+
+const schema: CustomSchema = {
+  valueType: 'list',
+  messages: { from: 'items' },
+  fields: { role: 'type', content: 'data.content' },
+};
+
+export const ns = (o: Partial<SessionNamespace> = {}): SessionNamespace => ({
+  id: 'ns-1',
+  connectionId: 'conn-a',
+  name: 'Chat history',
+  pattern: 'message_store:*',
+  preset: 'langchain-community',
+  customSchema: null,
+  oversizeThresholdBytes: 1048576,
+  createdAt: 1_000,
+  updatedAt: 1_000,
+  ...o,
+});
+
+export function runNamespaceStorageContract(getStorage: () => StoragePort): void {
+  let storage: StoragePort;
+  beforeEach(() => {
+    storage = getStorage();
+  });
+
+  it('round-trips a namespace including the custom schema', async () => {
+    const input = ns({ preset: 'custom', customSchema: schema, oversizeThresholdBytes: 5_000_000_000 });
+    const saved = await storage.saveSessionNamespace(input);
+    expect(saved.id).toBe('ns-1');
+    expect(await storage.getSessionNamespace('conn-a', 'ns-1')).toEqual(input);
+  });
+
+  it('lists only the namespaces of the given connection in creation order', async () => {
+    await storage.saveSessionNamespace(ns({ id: 'b', createdAt: 2_000 }));
+    await storage.saveSessionNamespace(ns({ id: 'a', createdAt: 1_000 }));
+    await storage.saveSessionNamespace(ns({ id: 'c', connectionId: 'conn-b' }));
+    expect((await storage.listSessionNamespaces('conn-a')).map((n) => n.id)).toEqual(['a', 'b']);
+    expect((await storage.listSessionNamespaces('conn-b')).map((n) => n.id)).toEqual(['c']);
+  });
+
+  it('does not expose a namespace through another connection', async () => {
+    await storage.saveSessionNamespace(ns());
+    expect(await storage.getSessionNamespace('conn-b', 'ns-1')).toBeNull();
+    expect(await storage.deleteSessionNamespace('conn-b', 'ns-1')).toBe(false);
+    expect(await storage.getSessionNamespace('conn-a', 'ns-1')).not.toBeNull();
+  });
+
+  it('upserts by id and keeps createdAt', async () => {
+    await storage.saveSessionNamespace(ns());
+    await storage.saveSessionNamespace(
+      ns({ name: 'Renamed', pattern: 'app:*', createdAt: 9_999, updatedAt: 2_000 }),
+    );
+    const all = await storage.listSessionNamespaces('conn-a');
+    expect(all).toHaveLength(1);
+    expect(all[0]).toMatchObject({
+      name: 'Renamed',
+      pattern: 'app:*',
+      createdAt: 1_000,
+      updatedAt: 2_000,
+    });
+  });
+
+  it('deletes an existing namespace and returns false for an unknown id', async () => {
+    await storage.saveSessionNamespace(ns());
+    expect(await storage.deleteSessionNamespace('conn-a', 'ns-1')).toBe(true);
+    expect(await storage.getSessionNamespace('conn-a', 'ns-1')).toBeNull();
+    expect(await storage.deleteSessionNamespace('conn-a', 'missing')).toBe(false);
+  });
+
+  it('rejects saving an id owned by another connection and leaves it unchanged', async () => {
+    await storage.saveSessionNamespace(ns());
+    await expect(
+      storage.saveSessionNamespace(ns({ connectionId: 'conn-b', name: 'Hijacked', updatedAt: 5_000 })),
+    ).rejects.toThrow();
+    expect(await storage.getSessionNamespace('conn-a', 'ns-1')).toEqual(ns());
+    expect(await storage.listSessionNamespaces('conn-b')).toEqual([]);
+  });
+}