diff --git a/apps/api/package.json b/apps/api/package.json index fbfc01b7f..548482574 100644 --- a/apps/api/package.json +++ b/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", diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 0583b7412..e65018afe 100644 --- a/apps/api/src/app.module.ts +++ b/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, diff --git a/apps/api/src/common/interfaces/storage-port.interface.ts b/apps/api/src/common/interfaces/storage-port.interface.ts index 5535fac1d..dfc7ed1c4 100644 --- a/apps/api/src/common/interfaces/storage-port.interface.ts +++ b/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; saveScalingReadinessSettings(settings: ScalingReadinessSettings): Promise; + listSessionNamespaces(connectionId: string): Promise; + getSessionNamespace(connectionId: string, id: string): Promise; + saveSessionNamespace(ns: SessionNamespace): Promise; + deleteSessionNamespace(connectionId: string, id: string): Promise; + // Command Stats Sample Methods - connectionId required for writes saveCommandStatsSamples( samples: Omit[], diff --git a/apps/api/src/session-explorer/__tests__/fake-valkey.ts b/apps/api/src/session-explorer/__tests__/fake-valkey.ts new file mode 100644 index 000000000..ff0b2ff83 --- /dev/null +++ b/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 | Record; +} + +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; + failKeys = new Map(); + movedKeys = new Set(); + + constructor( + public store: Map = new Map(), + options: Record = { host: 'seed', port: 7000 }, + ) { + this.options = options; + for (const name of WRITE_COMMANDS) { + (this as unknown as Record)[name] = () => { + throw new Error(`write command ${name} issued`); + }; + } + for (const name of READ_COMMANDS) { + (this as unknown as Record)[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) ?? []).slice(start, stop + 1).map((v) => Buffer.from(v)); + } + case 'hgetallBuffer': + return Object.fromEntries( + Object.entries((raw as Record) ?? {}).map(([k, v]) => [k, Buffer.from(v)]), + ); + case 'hscanBuffer': + return [ + Buffer.from('0'), + Object.entries((raw as Record) ?? {}).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 = {}; + 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; +} diff --git a/apps/api/src/session-explorer/__tests__/namespace-storage-contract.ts b/apps/api/src/session-explorer/__tests__/namespace-storage-contract.ts new file mode 100644 index 000000000..1a3e6afcf --- /dev/null +++ b/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 => ({ + 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([]); + }); +} diff --git a/apps/api/src/session-explorer/__tests__/namespace-storage.spec.ts b/apps/api/src/session-explorer/__tests__/namespace-storage.spec.ts new file mode 100644 index 000000000..112470784 --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/namespace-storage.spec.ts @@ -0,0 +1,38 @@ +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { SqliteAdapter } from '../../storage/adapters/sqlite.adapter'; +import { MemoryAdapter } from '../../storage/adapters/memory.adapter'; +import type { StoragePort } from '../../common/interfaces/storage-port.interface'; +import { runNamespaceStorageContract } from './namespace-storage-contract'; + +describe.each([ + ['memory', async () => { + const storage = new MemoryAdapter(); + await storage.initialize(); + return { storage: storage as StoragePort, cleanup: async () => storage.close() }; + }], + ['sqlite', async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'session-namespaces-')); + const storage = new SqliteAdapter({ filepath: path.join(dir, 'db.sqlite') }); + await storage.initialize(); + return { + storage: storage as StoragePort, + cleanup: async () => { + await storage.close(); + fs.rmSync(dir, { recursive: true, force: true }); + }, + }; + }], +])('%s adapter session namespace storage', (_name, make) => { + let storage: StoragePort; + let cleanup: () => Promise; + + beforeEach(async () => { + ({ storage, cleanup } = await make()); + }); + afterEach(async () => cleanup()); + + runNamespaceStorageContract(() => storage); +}); + diff --git a/apps/api/src/session-explorer/__tests__/namespace-validation.spec.ts b/apps/api/src/session-explorer/__tests__/namespace-validation.spec.ts new file mode 100644 index 000000000..075bff777 --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/namespace-validation.spec.ts @@ -0,0 +1,89 @@ +import { validateNamespaceInput } from '../namespace-validation'; + +const valid = { + name: 'Chat history', + pattern: 'message_store:*', + preset: 'langchain-community', + customSchema: null, + oversizeThresholdBytes: 1048576, +}; + +const errorsOf = (input: unknown): string[] => { + const result = validateNamespaceInput(input); + return result.ok ? [] : result.errors; +}; + +describe('validateNamespaceInput', () => { + it('accepts a valid input and trims the name', () => { + const result = validateNamespaceInput({ ...valid, name: ' Chat history ' }); + expect(result).toEqual({ ok: true, value: valid }); + }); + + it('defaults the oversize threshold when omitted', () => { + const { oversizeThresholdBytes: _omit, ...rest } = valid; + const result = validateNamespaceInput(rest); + expect(result.ok && result.value.oversizeThresholdBytes).toBe(1048576); + }); + + it.each(['*', '**', '?*', '[a]*', ''])('rejects bare wildcard pattern %p', (pattern) => { + expect(errorsOf({ ...valid, pattern })).toContain('Pattern must include a literal prefix'); + }); + + it.each(['app:*', 'message_store:*', 'a*'])('accepts pattern %p', (pattern) => { + expect(errorsOf({ ...valid, pattern })).toEqual([]); + }); + + it('accepts an escaped star as a literal prefix', () => { + expect(errorsOf({ ...valid, pattern: '\\*abc' })).toEqual([]); + }); + + it('trims the pattern and rejects inner whitespace', () => { + const result = validateNamespaceInput({ ...valid, pattern: ' app:* ' }); + expect(result.ok && result.value.pattern).toBe('app:*'); + expect(errorsOf({ ...valid, pattern: 'app: *' })).toContain('Pattern must not contain whitespace'); + }); + + it('rejects an empty name', () => { + expect(errorsOf({ ...valid, name: ' ' })).toContain('Name is required'); + }); + + it.each([0, -1, 1.5, '10', 1e300])('rejects threshold %p', (oversizeThresholdBytes) => { + expect(errorsOf({ ...valid, oversizeThresholdBytes })).toContain( + 'Oversize threshold must be a positive integer', + ); + }); + + it('rejects custom preset without a schema', () => { + expect(errorsOf({ ...valid, preset: 'custom', customSchema: null })).toContain( + 'Custom preset requires a schema', + ); + }); + + it('surfaces custom schema errors', () => { + const errors = errorsOf({ ...valid, preset: 'custom', customSchema: { nope: true } }); + expect(errors.length).toBeGreaterThan(0); + }); + + it('accepts a valid custom schema', () => { + const customSchema = { + valueType: 'list', + messages: { from: 'items' }, + fields: { role: 'type', content: 'data.content' }, + }; + const result = validateNamespaceInput({ ...valid, preset: 'custom', customSchema }); + expect(result.ok && result.value.customSchema).toEqual(customSchema); + }); + + it('rejects an unknown preset', () => { + expect(errorsOf({ ...valid, preset: 'mystery' })).toContain('Unknown preset'); + }); + + it('drops the schema for non-custom presets', () => { + const result = validateNamespaceInput({ ...valid, customSchema: { anything: 1 } }); + expect(result.ok && result.value.customSchema).toBeNull(); + }); + + it('rejects non-object input', () => { + expect(errorsOf(null)).toEqual(['Body must be an object']); + }); +}); diff --git a/apps/api/src/session-explorer/__tests__/session-explorer.controller.spec.ts b/apps/api/src/session-explorer/__tests__/session-explorer.controller.spec.ts new file mode 100644 index 000000000..254c3f024 --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/session-explorer.controller.spec.ts @@ -0,0 +1,458 @@ +import { Logger } from '@nestjs/common'; +import { FastifyAdapter, NestFastifyApplication } from '@nestjs/platform-fastify'; +import { Test } from '@nestjs/testing'; +import * as fs from 'fs'; +import * as path from 'path'; +import { Feature } from '@betterdb/shared'; +import type { CapturedKey, SessionNamespace } from '@betterdb/shared'; +import { LicenseGuard } from '@proprietary/licenses'; +import { LiveConnectionGuard } from '../../external-metrics/live-connection.guard'; +import { ConnectionRegistry } from '../../connections/connection-registry.service'; +import { SessionExplorerController } from '../session-explorer.controller'; +import { SessionExplorerService } from '../session-explorer.service'; +import { FakeEntry, FakeValkey, fakeDb } from './fake-valkey'; + +const human = (content: string) => JSON.stringify({ type: 'human', data: { content } }); +const ai = (content: string) => JSON.stringify({ type: 'ai', data: { content } }); + +function memoryStorage() { + const rows = new Map(); + const calls: string[] = []; + const storage = { + calls, + rows, + listSessionNamespaces: jest.fn(async (connectionId: string) => { + calls.push('listSessionNamespaces'); + return [...rows.values()].filter((r) => r.connectionId === connectionId); + }), + getSessionNamespace: jest.fn(async (connectionId: string, id: string) => { + calls.push('getSessionNamespace'); + const row = rows.get(id); + return row && row.connectionId === connectionId ? row : null; + }), + saveSessionNamespace: jest.fn(async (ns: SessionNamespace) => { + calls.push('saveSessionNamespace'); + rows.set(ns.id, ns); + return ns; + }), + deleteSessionNamespace: jest.fn(async (connectionId: string, id: string) => { + calls.push('deleteSessionNamespace'); + const row = rows.get(id); + if (!row || row.connectionId !== connectionId) return false; + rows.delete(id); + return true; + }), + }; + return storage; +} + +const namespace = (overrides: Partial): SessionNamespace => ({ + id: 'ns-raw', + connectionId: 'conn-1', + name: 'Raw chats', + pattern: 'chat:*', + preset: 'raw', + customSchema: null, + oversizeThresholdBytes: 1048576, + createdAt: 1, + updatedAt: 1, + ...overrides, +}); + +describe('SessionExplorerController', () => { + let app: NestFastifyApplication; + let storage: ReturnType; + let valkey: FakeValkey; + + const store = () => + new Map([ + ['chat:1', { type: 'string', pttl: 60000, memory: 120, value: 'hello SECRET-MARKER' }], + ['chat:2', { type: 'hash', pttl: -1, memory: 90, value: { a: 'b' } }], + [ + 'lc:good', + { type: 'list', pttl: 30000, memory: 200, value: [ai('hi there'), human('hello')] }, + ], + [ + 'lc:bad', + { type: 'list', pttl: 30000, memory: 200, value: [ai('fine'), 'not json'] }, + ], + ]); + + const headers = { 'x-connection-id': 'conn-1' }; + + beforeEach(async () => { + storage = memoryStorage(); + storage.rows.set('ns-raw', namespace({})); + storage.rows.set( + 'ns-lc', + namespace({ id: 'ns-lc', name: 'LangChain', pattern: 'lc:*', preset: 'langchain-community' }), + ); + valkey = new FakeValkey(store()); + const registry = { + get: () => fakeDb(valkey), + getConfig: () => ({ connectionType: 'direct' }), + }; + const moduleRef = await Test.createTestingModule({ + controllers: [SessionExplorerController], + providers: [ + SessionExplorerService, + { provide: ConnectionRegistry, useValue: registry }, + { provide: 'STORAGE_CLIENT', useValue: storage }, + ], + }) + .overrideGuard(LicenseGuard) + .useValue({ canActivate: () => true }) + .compile(); + app = moduleRef.createNestApplication(new FastifyAdapter()); + await app.init(); + await app.getHttpAdapter().getInstance().ready(); + }); + + afterEach(async () => { + await app.close(); + jest.restoreAllMocks(); + }); + + const input = { + name: 'Agent chats', + pattern: 'agent:*', + preset: 'raw', + customSchema: null, + oversizeThresholdBytes: 2048, + }; + + describe('namespace CRUD', () => { + it('lists namespaces for the connection', async () => { + const res = await app.inject({ method: 'GET', url: '/session-explorer/namespaces', headers }); + expect(res.statusCode).toBe(200); + expect(res.json().map((n: SessionNamespace) => n.id).sort()).toEqual(['ns-lc', 'ns-raw']); + }); + + it('creates a namespace bound to the connection', async () => { + const res = await app.inject({ + method: 'POST', + url: '/session-explorer/namespaces', + headers, + payload: input, + }); + expect(res.statusCode).toBe(201); + const body = res.json(); + expect(body).toMatchObject({ ...input, connectionId: 'conn-1' }); + expect(typeof body.id).toBe('string'); + expect(storage.rows.get(body.id)).toEqual(body); + }); + + it('rejects a pattern without a literal prefix with the validation errors', async () => { + const res = await app.inject({ + method: 'POST', + url: '/session-explorer/namespaces', + headers, + payload: { ...input, pattern: '*' }, + }); + expect(res.statusCode).toBe(400); + expect(res.json().errors).toContain('Pattern must include a literal prefix'); + expect(storage.saveSessionNamespace).not.toHaveBeenCalled(); + }); + + it('updates an existing namespace and keeps its identity', async () => { + const res = await app.inject({ + method: 'PUT', + url: '/session-explorer/namespaces/ns-raw', + headers, + payload: input, + }); + expect(res.statusCode).toBe(200); + expect(res.json()).toMatchObject({ ...input, id: 'ns-raw', connectionId: 'conn-1', createdAt: 1 }); + expect(res.json().updatedAt).toBeGreaterThan(1); + }); + + it('answers 404 when updating an unknown namespace', async () => { + const res = await app.inject({ + method: 'PUT', + url: '/session-explorer/namespaces/missing', + headers, + payload: input, + }); + expect(res.statusCode).toBe(404); + }); + + it('answers 404 for a namespace owned by another connection', async () => { + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions', + headers: { 'x-connection-id': 'conn-2' }, + }); + expect(res.statusCode).toBe(404); + }); + + it('deletes a namespace with 204 and answers 404 the second time', async () => { + const first = await app.inject({ + method: 'DELETE', + url: '/session-explorer/namespaces/ns-raw', + headers, + }); + expect(first.statusCode).toBe(204); + const second = await app.inject({ + method: 'DELETE', + url: '/session-explorer/namespaces/ns-raw', + headers, + }); + expect(second.statusCode).toBe(404); + }); + }); + + describe('preview', () => { + it('parses one sampled key without persisting anything', async () => { + valkey.scanPages = [['lc:good', 'lc:bad']]; + const res = await app.inject({ + method: 'POST', + url: '/session-explorer/namespaces/preview', + headers, + payload: { ...input, pattern: 'lc:*', preset: 'langchain-community' }, + }); + expect(res.statusCode).toBe(200); + const body = res.json(); + expect(body.sampleKey).toBe('lc:good'); + expect(body.view.parsed).toBe(true); + expect(body.view.key).toBe(body.sampleKey); + expect(storage.calls).toEqual([]); + expect(valkey.callsOf('scan')).toHaveLength(1); + }); + + it('answers nulls when no key matches', async () => { + valkey.scanPages = [[]]; + const res = await app.inject({ + method: 'POST', + url: '/session-explorer/namespaces/preview', + headers, + payload: input, + }); + expect(res.statusCode).toBe(200); + expect(res.json()).toEqual({ sampleKey: null, view: null }); + }); + + it('rejects an invalid preview body with 400', async () => { + const res = await app.inject({ + method: 'POST', + url: '/session-explorer/namespaces/preview', + headers, + payload: { ...input, pattern: '*' }, + }); + expect(res.statusCode).toBe(400); + expect(res.json().errors).toContain('Pattern must include a literal prefix'); + }); + }); + + describe('sessions', () => { + it('lists sessions of a namespace', async () => { + valkey.scanPages = [['chat:1', 'chat:2']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions', + headers, + }); + expect(res.statusCode).toBe(200); + const body = res.json(); + expect(body.sessions.map((s: { sessionId: string }) => s.sessionId).sort()).toEqual([ + 'chat:1', + 'chat:2', + ]); + expect(body).toMatchObject({ truncated: false, partial: false, failedNodes: 0 }); + }); + + it('returns every key unparsed with a reason for the raw preset', async () => { + valkey.scanPages = [['chat:1', 'chat:2']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1', + headers, + }); + expect(res.statusCode).toBe(200); + const body = res.json(); + expect(body.keys).toHaveLength(1); + for (const key of body.keys) { + expect(key).toMatchObject({ parsed: false, reason: 'No schema configured' }); + } + expect(body.complete).toBe(false); + }); + + it('marks a fully parsed session complete', async () => { + valkey.scanPages = [['lc:good', 'lc:bad']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-lc/sessions/lc%3Agood', + headers, + }); + expect(res.statusCode).toBe(200); + const body = res.json(); + expect(body.messages.map((m: { content: string }) => m.content)).toEqual([ + 'hello', + 'hi there', + ]); + expect(body).toMatchObject({ complete: true, unparsedKeys: 0, ttlMs: 30000, memoryBytes: 200 }); + }); + + it('marks a session with an unparsed key incomplete', async () => { + valkey.scanPages = [['lc:good', 'lc:bad']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-lc/sessions/lc%3Abad', + headers, + }); + expect(res.statusCode).toBe(200); + expect(res.json()).toMatchObject({ complete: false, unparsedKeys: 1 }); + }); + + it('answers 404 for a session with no keys without echoing its id', async () => { + valkey.scanPages = [['chat:1']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3Aghost', + headers, + }); + expect(res.statusCode).toBe(404); + expect(res.body).not.toContain('ghost'); + }); + }); + + describe('export', () => { + it('streams a JSON attachment holding every key as stored', async () => { + valkey.scanPages = [['chat:1', 'chat:2']]; + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1/export', + headers, + }); + expect(res.statusCode).toBe(200); + expect(res.headers['content-type']).toContain('application/json'); + expect(res.headers['content-disposition']).toBe('attachment; filename="session-chat_1.json"'); + const body = JSON.parse(res.body); + expect(body.namespace).toEqual({ name: 'Raw chats', pattern: 'chat:*', preset: 'raw' }); + expect(body.sessionId).toBe('chat:1'); + expect(typeof body.exportedAt).toBe('string'); + expect(body.keys).toEqual([ + expect.objectContaining({ key: 'chat:1', type: 'string', pttl: 60000, value: 'hello SECRET-MARKER' }), + ]); + expect(body.parsed.keys).toEqual([ + expect.objectContaining({ type: 'string', ttlMs: 60000, raw: 'hello SECRET-MARKER' }), + ]); + }); + }); + + describe('privacy', () => { + const contains = (args: unknown[]) => + args.some((a) => { + try { + return String(typeof a === 'string' ? a : JSON.stringify(a)).includes('SECRET-MARKER'); + } catch { + return false; + } + }); + + it('never logs session contents', async () => { + const spies = [ + ...(['log', 'error', 'warn', 'debug', 'verbose', 'fatal'] as const).map((m) => + jest.spyOn(Logger.prototype, m), + ), + ...(['log', 'error', 'warn', 'debug', 'verbose', 'fatal'] as const).map((m) => + jest.spyOn(Logger, m), + ), + ...(['log', 'info', 'warn', 'error', 'debug'] as const).map((m) => jest.spyOn(console, m)), + ]; + valkey.scanPages = [['chat:1']]; + await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1', + headers, + }); + await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1/export', + headers, + }); + for (const spy of spies) { + for (const call of spy.mock.calls) expect(contains(call)).toBe(false); + } + }); + + it('never persists session contents', async () => { + valkey.scanPages = [['chat:1']]; + await app.inject({ method: 'GET', url: '/session-explorer/namespaces/ns-raw/sessions', headers }); + await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1', + headers, + }); + await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/sessions/chat%3A1/export', + headers, + }); + expect(new Set(storage.calls)).toEqual(new Set(['getSessionNamespace'])); + }); + }); + + describe('oversize flags', () => { + const handler = SessionExplorerController.prototype.oversizeSessions; + + it('is gated by the license guard on the session oversize feature', () => { + expect(Reflect.getMetadata('requiredFeature', handler)).toBe(Feature.SESSION_OVERSIZE_FLAGS); + expect(Feature.SESSION_OVERSIZE_FLAGS).toBe('sessionOversizeFlags'); + expect(Reflect.getMetadata('__guards__', handler)).toEqual( + expect.arrayContaining([LiveConnectionGuard, LicenseGuard]), + ); + }); + + it('flags only sessions strictly above the threshold', async () => { + storage.rows.set('ns-raw', namespace({ oversizeThresholdBytes: 120 })); + valkey.scanPages = [['chat:1', 'chat:2']]; + valkey.store.set('chat:2', { type: 'string', pttl: -1, memory: 121, value: 'x' }); + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-raw/oversize', + headers, + }); + expect(res.statusCode).toBe(200); + expect(res.json()).toEqual({ + thresholdBytes: 120, + sessionIds: ['chat:2'], + unmeasuredSessionIds: [], + truncated: false, + partial: false, + failedNodes: 0, + }); + }); + + it('flags the captured oversize session at the default threshold', async () => { + const fixture = JSON.parse( + fs.readFileSync( + path.join(__dirname, '../../../test/fixtures/session-explorer/langchain-community@0.4.2.json'), + 'utf8', + ), + ) as { keys: CapturedKey[] }; + for (const k of fixture.keys) { + valkey.store.set(k.key, { + type: k.type, + pttl: k.pttl, + memory: Buffer.byteLength(JSON.stringify(k.value)), + value: k.value as string[], + }); + } + valkey.scanPages = [fixture.keys.map((k) => k.key)]; + storage.rows.set('ns-lc', namespace({ id: 'ns-lc', pattern: 'sxfix:lcc:*', preset: 'langchain-community' })); + const res = await app.inject({ + method: 'GET', + url: '/session-explorer/namespaces/ns-lc/oversize', + headers, + }); + expect(res.json()).toEqual({ + thresholdBytes: 1048576, + sessionIds: ['sxfix:lcc:oversize'], + unmeasuredSessionIds: [], + truncated: false, + partial: false, + failedNodes: 0, + }); + }); + }); +}); diff --git a/apps/api/src/session-explorer/__tests__/session-explorer.service.spec.ts b/apps/api/src/session-explorer/__tests__/session-explorer.service.spec.ts new file mode 100644 index 000000000..ee173c04c --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/session-explorer.service.spec.ts @@ -0,0 +1,152 @@ +import { NotFoundException } from '@nestjs/common'; +import type { CapturedKey, SessionNamespace } from '@betterdb/shared'; +import { SessionExplorerService } from '../session-explorer.service'; +import { keysForSession, scanSessions } from '../session-scanner'; +import { readSessionKeys } from '../session-reader'; + +jest.mock('../session-scanner', () => ({ + ...jest.requireActual('../session-scanner'), + keysForSession: jest.fn(), + scanSessions: jest.fn(), +})); +jest.mock('../session-reader', () => ({ + ...jest.requireActual('../session-reader'), + readSessionKeys: jest.fn(), +})); + +const keysFor = keysForSession as jest.MockedFunction; +const read = readSessionKeys as jest.MockedFunction; + +const ns: SessionNamespace = { + id: 'ns-1', + connectionId: 'conn-1', + name: 'LangChain', + pattern: 'lc:*', + preset: 'langchain-community', + customSchema: null, + oversizeThresholdBytes: 1048576, + createdAt: 1, + updatedAt: 1, +}; + +const message = JSON.stringify({ type: 'human', data: { content: 'abcdefgh' } }); + +const captured = (overrides: Partial = {}): CapturedKey => ({ + key: 'lc:1', + type: 'list', + pttl: 5000, + value: [message], + bytes: 64, + ...overrides, +}); + +describe('SessionExplorerService.getSession', () => { + const storage = { getSessionNamespace: jest.fn(async () => ns) }; + const registry = { get: jest.fn(() => ({})) }; + const service = new SessionExplorerService(registry as never, storage as never); + + beforeEach(() => { + keysFor.mockResolvedValue({ keys: ['lc:1'], truncated: false, failedNodes: 0 }); + read.mockResolvedValue({ keys: [captured()], truncated: false, failedNodes: 0, missingKeys: 0 }); + }); + + it('summarises a clean session as complete with a token estimate', async () => { + const detail = await service.getSession('conn-1', 'ns-1', 'lc:1'); + expect(detail).toMatchObject({ + sessionId: 'lc:1', + preset: 'langchain-community', + complete: true, + unparsedKeys: 0, + truncated: false, + failedNodes: 0, + missingKeys: 0, + estimatedTokens: 2, + ttlMs: 5000, + memoryBytes: 64, + }); + }); + + it('is incomplete when the key scan was capped', async () => { + keysFor.mockResolvedValue({ keys: ['lc:1'], truncated: true, failedNodes: 0 }); + expect(await service.getSession('conn-1', 'ns-1', 'lc:1')).toMatchObject({ + complete: false, + truncated: true, + }); + }); + + it('is incomplete when the read budget ran out', async () => { + read.mockResolvedValue({ keys: [captured()], truncated: true, failedNodes: 0, missingKeys: 0 }); + expect(await service.getSession('conn-1', 'ns-1', 'lc:1')).toMatchObject({ + complete: false, + truncated: true, + }); + }); + + it('is incomplete when a value was cut at the byte cap', async () => { + read.mockResolvedValue({ + keys: [captured({ truncated: true })], + truncated: false, + failedNodes: 0, + missingKeys: 0, + }); + expect(await service.getSession('conn-1', 'ns-1', 'lc:1')).toMatchObject({ complete: false }); + }); + + it('is incomplete when nodes failed or keys vanished', async () => { + keysFor.mockResolvedValue({ keys: ['lc:1', 'lc:2'], truncated: false, failedNodes: 1 }); + read.mockResolvedValue({ keys: [captured()], truncated: false, failedNodes: 1, missingKeys: 1 }); + expect(await service.getSession('conn-1', 'ns-1', 'lc:1')).toMatchObject({ + complete: false, + failedNodes: 2, + missingKeys: 1, + }); + }); + + it('answers not found when every key vanished before the read', async () => { + read.mockResolvedValue({ keys: [], truncated: false, failedNodes: 0, missingKeys: 1 }); + await expect(service.getSession('conn-1', 'ns-1', 'lc:1')).rejects.toBeInstanceOf( + NotFoundException, + ); + }); + + it('answers not found for an unknown namespace', async () => { + storage.getSessionNamespace.mockResolvedValueOnce(null as never); + await expect(service.getSession('conn-1', 'missing', 'lc:1')).rejects.toBeInstanceOf( + NotFoundException, + ); + }); +}); + +describe('SessionExplorerService.oversizeSessions', () => { + const scan = scanSessions as jest.MockedFunction; + const storage = { + getSessionNamespace: jest.fn(async () => ({ ...ns, oversizeThresholdBytes: 100 })), + }; + const service = new SessionExplorerService({ get: jest.fn(() => ({})) } as never, storage as never); + const summary = (sessionId: string, memoryBytes: number, unmeasuredKeys = 0) => ({ + sessionId, + keyCount: 1, + memoryBytes, + unmeasuredKeys, + idleSeconds: null, + ttlMs: null, + }); + + it('carries partial scans and unmeasured sessions into the response', async () => { + scan.mockResolvedValue({ + sessions: [summary('big', 500), summary('unknown', 10, 1), summary('small', 10)], + scannedKeys: 3, + truncated: false, + partial: true, + failedNodes: 1, + }); + expect(await service.oversizeSessions('conn-1', 'ns-1')).toEqual({ + thresholdBytes: 100, + sessionIds: ['big'], + unmeasuredSessionIds: ['unknown'], + truncated: false, + partial: true, + failedNodes: 1, + }); + }); +}); diff --git a/apps/api/src/session-explorer/__tests__/session-reader.spec.ts b/apps/api/src/session-explorer/__tests__/session-reader.spec.ts new file mode 100644 index 000000000..8c3bf9850 --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/session-reader.spec.ts @@ -0,0 +1,281 @@ +import Valkey from 'iovalkey'; +import calculateSlot from 'cluster-key-slot'; +import { readSessionKeys } from '../session-reader'; +import { FakeEntry, FakeValkey, fakeDb, masterNode } from './fake-valkey'; + +jest.mock('iovalkey', () => ({ __esModule: true, default: jest.fn() })); + +const ValkeyMock = Valkey as unknown as jest.Mock; +const MB = 1024 * 1024; + +const storeOf = (entries: Record) => new Map(Object.entries(entries)); + +const readOne = async (client: FakeValkey, key: string, opts = {}) => + (await readSessionKeys(fakeDb(client), [key], opts)).keys[0]; + +function clusterOf(nodes: Record, slotsByPort: Record) { + ValkeyMock.mockImplementation((opts: { port: number }) => nodes[opts.port]); + return fakeDb(new FakeValkey(), { + nodes: Object.keys(nodes).map((port) => + masterNode(port, `10.0.0.1:${port}`, [], slotsByPort[Number(port)]), + ), + }); +} + +describe('readSessionKeys', () => { + beforeEach(() => ValkeyMock.mockReset()); + + it('reads a small string whole', async () => { + const client = new FakeValkey(storeOf({ k: { type: 'string', value: 'hello', pttl: 500 } })); + const key = await readOne(client, 'k'); + expect(key).toMatchObject({ key: 'k', type: 'string', pttl: 500, value: 'hello' }); + expect(key.truncated).toBeFalsy(); + }); + + it('truncates a 3 MB string at a 2 MB cap without reading it whole', async () => { + const big = 'x'.repeat(3 * MB); + const client = new FakeValkey(storeOf({ k: { type: 'string', value: big, memory: 3 * MB } })); + + const key = await readOne(client, 'k', { byteCap: 2 * MB }); + + expect(key.truncated).toBe(true); + expect((key.value as string).length).toBeGreaterThanOrEqual(2 * MB - 4); + expect((key.value as string).length).toBeLessThanOrEqual(2 * MB); + expect(client.callsOf('pipeline.getBuffer')).toHaveLength(0); + expect(client.callsOf('pipeline.getrangeBuffer')[0].args).toEqual(['k', 0, 2 * MB - 1]); + }); + + it('keeps a multibyte character split by the cap out of the truncated string', async () => { + const client = new FakeValkey(storeOf({ k: { type: 'string', value: 'ab€€€' } })); + const key = await readOne(client, 'k', { byteCap: 6 }); + expect(key).toMatchObject({ value: 'ab€', truncated: true }); + }); + + it('falls back to base64 for a binary string', async () => { + const client = new FakeValkey( + storeOf({ k: { type: 'string', value: Buffer.from([0x80, 0x81, 0xff]) } }), + ); + const key = await readOne(client, 'k'); + expect(key.value).toEqual({ base64: Buffer.from([0x80, 0x81, 0xff]).toString('base64') }); + }); + + it('pipelines a session of small string keys in batches', async () => { + const entries: Record = {}; + const keys = Array.from({ length: 2500 }, (_, i) => `s:${i}`); + for (const key of keys) entries[key] = { type: 'string', value: 'v', memory: 50 }; + const client = new FakeValkey(storeOf(entries)); + + const result = await readSessionKeys(fakeDb(client), keys, { batch: 1000 }); + + expect(result.keys).toHaveLength(2500); + expect(client.pipelineExecs).toBe(6); + expect(client.directReads).toHaveLength(0); + }); + + it('reads a list over 500 items in windows', async () => { + const items = Array.from({ length: 1200 }, (_, i) => `m${i}`); + const client = new FakeValkey(storeOf({ l: { type: 'list', value: items } })); + + const key = await readOne(client, 'l'); + + expect(client.callsOf('pipeline.lrangeBuffer').map((c) => c.args.slice(1))).toEqual([[0, 499]]); + expect(client.callsOf('lrangeBuffer').map((c) => c.args.slice(1))).toEqual([ + [500, 999], + [1000, 1499], + ]); + expect(key.value).toEqual(items); + expect(key.truncated).toBeFalsy(); + }); + + it('stops reading a list at the byte cap', async () => { + const items = Array.from({ length: 50 }, () => 'y'.repeat(100)); + const client = new FakeValkey(storeOf({ l: { type: 'list', value: items } })); + + const key = await readOne(client, 'l', { byteCap: 1000, listWindow: 20 }); + + expect(key.truncated).toBe(true); + expect((key.value as string[]).length).toBe(10); + expect(client.callsOf('lrangeBuffer')).toHaveLength(0); + }); + + it('reads a hash as a record of strings', async () => { + const client = new FakeValkey( + storeOf({ h: { type: 'hash', value: { a: '1', b: '2' }, memory: 50 } }), + ); + const key = await readOne(client, 'h'); + expect(key.value).toEqual({ a: '1', b: '2' }); + expect(client.callsOf('pipeline.hgetallBuffer')).toHaveLength(1); + }); + + it('reads a hash larger than the cap with a bounded HSCAN', async () => { + const client = new FakeValkey( + storeOf({ h: { type: 'hash', value: { a: 'x'.repeat(60), b: 'y'.repeat(60) }, memory: 5000 } }), + ); + const key = await readOne(client, 'h', { byteCap: 100 }); + expect(client.callsOf('pipeline.hgetallBuffer')).toHaveLength(0); + expect(client.callsOf('hscanBuffer')).toHaveLength(1); + expect(key.truncated).toBe(true); + expect(Object.keys(key.value as object)).toEqual(['a']); + }); + + it('reads a sorted set as member and score pairs', async () => { + const client = new FakeValkey(storeOf({ z: { type: 'zset', value: ['m1', '1', 'm2', '2.5'] } })); + const key = await readOne(client, 'z'); + expect(key.value).toEqual([ + ['m1', 1], + ['m2', 2.5], + ]); + }); + + it('returns ReJSON-RL values as the raw JSON.GET string', async () => { + const client = new FakeValkey(storeOf({ j: { type: 'ReJSON-RL', value: '{"a":[1,2]}' } })); + const key = await readOne(client, 'j'); + expect(key.value).toBe('{"a":[1,2]}'); + expect(client.callsOf('pipeline.callBuffer')[0].args).toEqual(['JSON.GET', 'j']); + }); + + it('applies the byte cap to JSON values', async () => { + const doc = JSON.stringify({ text: 'z'.repeat(500) }); + const client = new FakeValkey(storeOf({ j: { type: 'ReJSON-RL', value: doc, memory: 300 } })); + const key = await readOne(client, 'j', { byteCap: 100 }); + expect(key.truncated).toBe(true); + expect((key.value as string).length).toBe(100); + }); + + it('does not fetch a JSON document far beyond the cap', async () => { + const client = new FakeValkey( + storeOf({ j: { type: 'ReJSON-RL', value: '{}', memory: 100 * MB } }), + ); + const key = await readOne(client, 'j', { byteCap: MB }); + expect(key).toMatchObject({ value: '', truncated: true }); + expect(client.callsOf('pipeline.callBuffer')).toHaveLength(0); + expect(client.callsOf('callBuffer')).toHaveLength(0); + }); + + it('sizes JSON with JSON.DEBUG MEMORY when MEMORY USAGE is unavailable', async () => { + const client = new FakeValkey( + storeOf({ j: { type: 'ReJSON-RL', value: '{"a":1}', memory: null, jsonMemory: 40 } }), + ); + const key = await readOne(client, 'j'); + expect(client.callsOf('pipeline.call')[0].args).toEqual(['JSON.DEBUG', 'MEMORY', 'j']); + expect(key).toMatchObject({ value: '{"a":1}' }); + expect(key.truncated).toBeFalsy(); + }); + + it('does not read JSON when no size is available', async () => { + const client = new FakeValkey( + storeOf({ j: { type: 'ReJSON-RL', value: '{"a":1}', memory: null, jsonMemory: null } }), + ); + const key = await readOne(client, 'j'); + expect(key).toMatchObject({ value: '', truncated: true }); + expect(client.calls.some((c) => c.args[0] === 'JSON.GET')).toBe(false); + }); + + it('omits a missing key', async () => { + const client = new FakeValkey(storeOf({ k: { type: 'string', value: 'v' } })); + const result = await readSessionKeys(fakeDb(client), ['gone', 'k']); + expect(result.keys.map((k) => k.key)).toEqual(['k']); + expect(result.missingKeys).toBe(0); + }); + + it('omits a key whose read fails without failing the request', async () => { + const client = new FakeValkey( + storeOf({ k1: { type: 'string', value: 'a' }, k2: { type: 'string', value: 'b' } }), + ); + client.failKeys.set('k1', 'WRONGTYPE Operation against a key holding the wrong kind of value'); + + const result = await readSessionKeys(fakeDb(client), ['k1', 'k2']); + + expect(result.keys.map((k) => k.key)).toEqual(['k2']); + expect(result.missingKeys).toBe(1); + }); + + it('records memory usage as bytes', async () => { + const client = new FakeValkey(storeOf({ k: { type: 'string', value: 'v', memory: 72 } })); + const key = await readOne(client, 'k'); + expect(key.bytes).toBe(72); + }); + + it('stops at the total byte budget and flags truncation', async () => { + const entries: Record = {}; + const keys = Array.from({ length: 10 }, (_, i) => `k${i}`); + for (const key of keys) entries[key] = { type: 'string', value: 'x'.repeat(100), memory: 100 }; + const client = new FakeValkey(storeOf(entries)); + + const result = await readSessionKeys(fakeDb(client), keys, { totalByteCap: 350 }); + + expect(result.truncated).toBe(true); + expect(result.keys.map((k) => k.key)).toEqual(['k0', 'k1', 'k2', 'k3']); + expect(result.keys[2].truncated).toBeFalsy(); + expect(result.keys[3]).toMatchObject({ value: 'x'.repeat(50), truncated: true }); + }); + + it('reads a key larger than the remaining budget up to the budget', async () => { + const client = new FakeValkey( + storeOf({ + big: { type: 'string', value: 'b'.repeat(500), memory: 500 }, + small: { type: 'string', value: 's', memory: 1 }, + }), + ); + + const result = await readSessionKeys(fakeDb(client), ['big', 'small'], { totalByteCap: 200 }); + + expect(result.truncated).toBe(true); + expect(result.keys).toEqual([ + expect.objectContaining({ key: 'big', value: 'b'.repeat(200), truncated: true }), + ]); + }); + + it('is not truncated when everything fits the budget', async () => { + const client = new FakeValkey(storeOf({ k: { type: 'string', value: 'v' } })); + const result = await readSessionKeys(fakeDb(client), ['k']); + expect(result).toMatchObject({ truncated: false, failedNodes: 0, missingKeys: 0 }); + }); + + it('routes each key to the primary that owns its slot', async () => { + const nodeA = new FakeValkey(storeOf({ a: { type: 'string', value: 'from-a' } })); + const nodeB = new FakeValkey(storeOf({ b: { type: 'string', value: 'from-b' } })); + const db = clusterOf( + { 7001: nodeA, 7002: nodeB }, + { 7001: [[calculateSlot('a'), calculateSlot('a')]], 7002: [[calculateSlot('b'), calculateSlot('b')]] }, + ); + + const result = await readSessionKeys(db, ['a', 'b']); + + expect(result.keys.map((k) => [k.key, k.value])).toEqual([ + ['a', 'from-a'], + ['b', 'from-b'], + ]); + expect(nodeA.callsOf('pipeline.type').map((c) => c.args[0])).toEqual(['a']); + expect(nodeB.callsOf('pipeline.type').map((c) => c.args[0])).toEqual(['b']); + expect(nodeA.disconnected && nodeB.disconnected).toBe(true); + }); + + it('omits a key answered with MOVED during slot migration', async () => { + const nodeA = new FakeValkey( + storeOf({ a: { type: 'string', value: '1' }, b: { type: 'string', value: '2' } }), + ); + nodeA.movedKeys.add('a'); + const db = clusterOf({ 7001: nodeA }, { 7001: [[0, 16383]] }); + + const result = await readSessionKeys(db, ['a', 'b']); + + expect(result.keys.map((k) => k.key)).toEqual(['b']); + expect(result.missingKeys).toBe(1); + }); + + it('reports keys owned by an unreachable primary', async () => { + const nodeA = new FakeValkey(storeOf({ a: { type: 'string', value: '1' } })); + const nodeB = new FakeValkey(storeOf({ b: { type: 'string', value: '2' } })); + nodeB.connectError = new Error('ECONNREFUSED'); + const db = clusterOf( + { 7001: nodeA, 7002: nodeB }, + { 7001: [[calculateSlot('a'), calculateSlot('a')]], 7002: [[calculateSlot('b'), calculateSlot('b')]] }, + ); + + const result = await readSessionKeys(db, ['a', 'b']); + + expect(result.keys.map((k) => k.key)).toEqual(['a']); + expect(result).toMatchObject({ failedNodes: 1, missingKeys: 1 }); + }); +}); diff --git a/apps/api/src/session-explorer/__tests__/session-scanner.spec.ts b/apps/api/src/session-explorer/__tests__/session-scanner.spec.ts new file mode 100644 index 000000000..c210e1147 --- /dev/null +++ b/apps/api/src/session-explorer/__tests__/session-scanner.spec.ts @@ -0,0 +1,331 @@ +import Valkey from 'iovalkey'; +import type { PresetDefinition } from '../presets/types'; +import { primaryClients } from '../primary-nodes'; +import { keysForSession, scanSessions } from '../session-scanner'; +import { FakeEntry, FakeValkey, fakeDb, masterNode } from './fake-valkey'; + +jest.mock('iovalkey', () => ({ __esModule: true, default: jest.fn() })); + +const ValkeyMock = Valkey as unknown as jest.Mock; + +const preset: PresetDefinition = { + id: 'custom', + defaultPattern: (prefix) => `${prefix}*`, + sessionIdOf: (key) => /^s:([^:]+):/.exec(key)?.[1] ?? null, + parseKey: () => { + throw new Error('not used'); + }, + mergeSession: () => [], +}; + +const keyRange = (from: number, count: number, session = 'a') => + Array.from({ length: count }, (_, i) => `s:${session}:${from + i}`); + +const storeOf = (entries: Record) => new Map(Object.entries(entries)); + +describe('scanSessions', () => { + beforeEach(() => ValkeyMock.mockReset()); + + it('stops at keyCap and sets truncated', async () => { + const client = new FakeValkey(); + client.scanPages = [keyRange(0, 5000), keyRange(5000, 5000), keyRange(10000, 5000)]; + for (const page of client.scanPages) { + for (const key of page) client.store.set(key, { type: 'string' }); + } + + const result = await scanSessions(fakeDb(client), 's:*', preset, { keyCap: 10000 }); + + expect(result.scannedKeys).toBe(10000); + expect(result.truncated).toBe(true); + expect(client.callsOf('scan')).toHaveLength(2); + expect(result.sessions).toEqual([expect.objectContaining({ sessionId: 'a', keyCount: 10000 })]); + }); + + it('applies the key cap as a total across primaries', async () => { + const nodes = [7001, 7002, 7003].map((port, n) => { + const node = new FakeValkey(); + node.scanPages = [keyRange(n * 5000, 5000)]; + for (const key of node.scanPages[0]) node.store.set(key, { type: 'string' }); + return node; + }); + const byPort: Record = { 7001: nodes[0], 7002: nodes[1], 7003: nodes[2] }; + ValkeyMock.mockImplementation((opts: { port: number }) => byPort[opts.port]); + const db = fakeDb(new FakeValkey(), { + nodes: [ + masterNode('a', '10.0.0.1:7001'), + masterNode('b', '10.0.0.2:7002'), + masterNode('c', '10.0.0.3:7003'), + ], + }); + + const result = await scanSessions(db, 's:*', preset, { keyCap: 10000 }); + + expect(result.scannedKeys).toBe(10000); + expect(result.truncated).toBe(true); + expect(nodes[2].callsOf('scan')).toHaveLength(0); + expect(nodes.every((n) => n.disconnected)).toBe(true); + }); + + it('is not truncated when the scan completes under the cap', async () => { + const client = new FakeValkey(storeOf({ 's:a:1': { type: 'string' } })); + const result = await scanSessions(fakeDb(client), 's:*', preset); + expect(result.truncated).toBe(false); + expect(result.partial).toBe(false); + expect(result.failedNodes).toBe(0); + }); + + it('never calls KEYS', async () => { + const client = new FakeValkey(storeOf({ 's:a:1': { type: 'string' } })); + await scanSessions(fakeDb(client), 's:*', preset); + expect(client.callsOf('keys')).toHaveLength(0); + expect(client.calls.some((c) => c.name.toLowerCase().includes('keys'))).toBe(false); + }); + + it('uses MATCH and COUNT 1000', async () => { + const client = new FakeValkey(storeOf({ 's:a:1': { type: 'string' } })); + await scanSessions(fakeDb(client), 's:*', preset); + expect(client.callsOf('scan')[0].args).toEqual(['0', 'MATCH', 's:*', 'COUNT', 1000]); + }); + + it('pipelines metadata', async () => { + const entries: Record = {}; + for (const key of keyRange(0, 2500)) entries[key] = { type: 'string' }; + const client = new FakeValkey(storeOf(entries)); + + const result = await scanSessions(fakeDb(client), 's:*', preset, { batch: 1000 }); + + expect(client.pipelineExecs).toBe(3); + expect(client.callsOf('type')).toHaveLength(0); + expect(client.callsOf('pipeline.type')).toHaveLength(2500); + expect(client.callsOf('pipeline.pttl')).toHaveLength(2500); + expect(client.callsOf('pipeline.memory')[0].args).toEqual(['USAGE', 's:a:0']); + expect(client.callsOf('pipeline.object')[0].args).toEqual(['IDLETIME', 's:a:0']); + expect(result.sessions[0].keyCount).toBe(2500); + }); + + it('skips keys that vanished between scan and metadata', async () => { + const client = new FakeValkey( + storeOf({ + 's:a:1': { type: 'string', memory: 10 }, + 's:a:2': { type: 'string', memory: 20 }, + 's:b:1': { type: 'string', memory: 5 }, + }), + ); + client.scanPages = [['s:a:1', 's:a:2', 's:a:gone', 's:c:gone', 's:b:1']]; + + const result = await scanSessions(fakeDb(client), 's:*', preset); + + expect(result.scannedKeys).toBe(5); + expect(result.sessions.map((s) => s.sessionId).sort()).toEqual(['a', 'b']); + const a = result.sessions.find((s) => s.sessionId === 'a'); + expect(a).toMatchObject({ keyCount: 2, memoryBytes: 30 }); + }); + + it('drops keys the preset does not recognise', async () => { + const client = new FakeValkey( + storeOf({ 's:a:1': { type: 'string' }, stray: { type: 'string' } }), + ); + const result = await scanSessions(fakeDb(client), '*', preset); + expect(result.sessions).toHaveLength(1); + expect(client.callsOf('pipeline.type').map((c) => c.args[0])).toEqual(['s:a:1']); + }); + + it('aggregates memory as sum and idle as min', async () => { + const client = new FakeValkey( + storeOf({ + 's:a:1': { type: 'string', memory: 100, idle: 50 }, + 's:a:2': { type: 'list', memory: 300, idle: 7 }, + }), + ); + const [session] = (await scanSessions(fakeDb(client), 's:*', preset)).sessions; + expect(session).toMatchObject({ memoryBytes: 400, idleSeconds: 7 }); + }); + + it('counts keys whose MEMORY USAGE failed instead of sizing them as zero', async () => { + const client = new FakeValkey( + storeOf({ + 's:a:1': { type: 'string', memory: 100 }, + 's:a:2': { type: 'string', memory: null }, + }), + ); + const [session] = (await scanSessions(fakeDb(client), 's:*', preset)).sessions; + expect(session).toMatchObject({ memoryBytes: 100, unmeasuredKeys: 1 }); + }); + + it('ttl is null when no key expires', async () => { + const client = new FakeValkey( + storeOf({ 's:a:1': { type: 'string', pttl: -1 }, 's:a:2': { type: 'string', pttl: -1 } }), + ); + const [session] = (await scanSessions(fakeDb(client), 's:*', preset)).sessions; + expect(session.ttlMs).toBeNull(); + }); + + it('ttl is the smallest positive PTTL', async () => { + const client = new FakeValkey( + storeOf({ + 's:a:1': { type: 'string', pttl: -1 }, + 's:a:2': { type: 'string', pttl: 9000 }, + 's:a:3': { type: 'string', pttl: 4000 }, + }), + ); + const [session] = (await scanSessions(fakeDb(client), 's:*', preset)).sessions; + expect(session.ttlMs).toBe(4000); + }); + + it('flags partial results when a primary fails', async () => { + const parent = new FakeValkey(new Map(), { host: 'seed', port: 7000, password: 'pw' }); + const nodeA = new FakeValkey(storeOf({ 's:a:1': { type: 'string' } })); + const nodeB = new FakeValkey(storeOf({ 's:b:1': { type: 'string' } })); + const nodeC = new FakeValkey(storeOf({ 's:c:1': { type: 'string' } })); + nodeC.connectError = new Error('ECONNREFUSED'); + const byPort: Record = { 7001: nodeA, 7002: nodeB, 7003: nodeC }; + ValkeyMock.mockImplementation((opts: { port: number }) => byPort[opts.port]); + + const db = fakeDb(parent, { + nodes: [ + masterNode('a', '10.0.0.1:7001'), + masterNode('b', '10.0.0.2:7002'), + masterNode('c', '10.0.0.3:7003'), + ], + }); + const result = await scanSessions(db, 's:*', preset); + + expect(result.partial).toBe(true); + expect(result.failedNodes).toBe(1); + expect(result.sessions.map((s) => s.sessionId).sort()).toEqual(['a', 'b']); + expect(ValkeyMock).toHaveBeenCalledWith(expect.objectContaining({ password: 'pw' })); + expect([nodeA, nodeB, nodeC].every((n) => n.disconnected)).toBe(true); + expect(parent.disconnected).toBe(false); + }); + + it('counts a primary whose scan fails as failed', async () => { + const parent = new FakeValkey(); + const nodeA = new FakeValkey(storeOf({ 's:a:1': { type: 'string' } })); + const nodeB = new FakeValkey(); + nodeB.scan = async () => { + throw new Error('LOADING'); + }; + const byPort: Record = { 7001: nodeA, 7002: nodeB }; + ValkeyMock.mockImplementation((opts: { port: number }) => byPort[opts.port]); + const db = fakeDb(parent, { + nodes: [masterNode('a', '10.0.0.1:7001'), masterNode('b', '10.0.0.2:7002')], + }); + + const result = await scanSessions(db, 's:*', preset); + + expect(result).toMatchObject({ partial: true, failedNodes: 1 }); + expect(result.sessions).toHaveLength(1); + }); + + it('rethrows when the only node fails', async () => { + const client = new FakeValkey(); + client.scan = async () => { + throw new Error('NOAUTH'); + }; + await expect(scanSessions(fakeDb(client), 's:*', preset)).rejects.toThrow('NOAUTH'); + expect(client.disconnected).toBe(false); + }); +}); + +describe('primaryClients', () => { + beforeEach(() => ValkeyMock.mockReset()); + + it('returns the parent client for standalone and never closes it', async () => { + const parent = new FakeValkey(); + const result = await primaryClients(fakeDb(parent)); + expect(result.clients).toEqual([parent]); + expect(result.failed).toBe(0); + await result.release(); + expect(parent.disconnected).toBe(false); + expect(ValkeyMock).not.toHaveBeenCalled(); + }); + + it('keeps only masters with slots and reuses the parent for myself', async () => { + const parent = new FakeValkey(); + const nodeB = new FakeValkey(); + ValkeyMock.mockImplementation(() => nodeB); + const replica = { ...masterNode('r', '10.0.0.9:7009'), flags: ['slave'], master: 'a' }; + const empty = { ...masterNode('e', '10.0.0.8:7008'), slots: [] }; + const db = fakeDb(parent, { + nodes: [ + masterNode('a', '10.0.0.1:7000', ['myself']), + masterNode('b', '10.0.0.2:7002'), + replica, + empty, + ], + }); + + const result = await primaryClients(db); + + expect(result.clients).toEqual([parent, nodeB]); + expect(ValkeyMock).toHaveBeenCalledTimes(1); + expect(ValkeyMock).toHaveBeenCalledWith( + expect.objectContaining({ host: '10.0.0.2', port: 7002, lazyConnect: true }), + ); + await result.release(); + expect(nodeB.disconnected).toBe(true); + expect(parent.disconnected).toBe(false); + }); +}); + +describe('keysForSession', () => { + beforeEach(() => ValkeyMock.mockReset()); + + it('scans every primary and returns only the session keys', async () => { + const parent = new FakeValkey(); + const nodeA = new FakeValkey( + storeOf({ 's:x:1': { type: 'string' }, 's:y:1': { type: 'string' } }), + ); + const nodeB = new FakeValkey(storeOf({ 's:x:2': { type: 'string' }, other: { type: 'string' } })); + const byPort: Record = { 7001: nodeA, 7002: nodeB }; + ValkeyMock.mockImplementation((opts: { port: number }) => byPort[opts.port]); + const db = fakeDb(parent, { + nodes: [masterNode('a', '10.0.0.1:7001'), masterNode('b', '10.0.0.2:7002')], + }); + + const result = await keysForSession(db, 's:*', preset, 'x'); + + expect(result.keys.sort()).toEqual(['s:x:1', 's:x:2']); + expect(result).toMatchObject({ truncated: false, failedNodes: 0 }); + expect(nodeA.disconnected && nodeB.disconnected).toBe(true); + }); + + it('respects the key cap', async () => { + const client = new FakeValkey(); + client.scanPages = [keyRange(0, 5000, 'x'), keyRange(5000, 5000, 'x')]; + const result = await keysForSession(fakeDb(client), 's:*', preset, 'x', 3000); + expect(result.keys).toHaveLength(3000); + expect(result.truncated).toBe(true); + expect(client.callsOf('scan')).toHaveLength(1); + }); + + it('finds session keys past the cap of non-matching keys', async () => { + const client = new FakeValkey(); + client.scanPages = [keyRange(0, 5000, 'y'), keyRange(5000, 5000, 'y'), ['s:x:1']]; + const result = await keysForSession(fakeDb(client), 's:*', preset, 'x', 3000); + expect(result).toMatchObject({ keys: ['s:x:1'], truncated: false }); + }); + + it('reports primaries that could not be reached or scanned', async () => { + const nodeA = new FakeValkey(storeOf({ 's:x:1': { type: 'string' } })); + const nodeB = new FakeValkey(); + nodeB.connectError = new Error('ECONNREFUSED'); + const nodeC = new FakeValkey(); + nodeC.scan = async () => { + throw new Error('LOADING'); + }; + const byPort: Record = { 7001: nodeA, 7002: nodeB, 7003: nodeC }; + ValkeyMock.mockImplementation((opts: { port: number }) => byPort[opts.port]); + const db = fakeDb(new FakeValkey(), { + nodes: [ + masterNode('a', '10.0.0.1:7001'), + masterNode('b', '10.0.0.2:7002'), + masterNode('c', '10.0.0.3:7003'), + ], + }); + + const result = await keysForSession(db, 's:*', preset, 'x'); + + expect(result).toEqual({ keys: ['s:x:1'], truncated: false, failedNodes: 2 }); + }); +}); diff --git a/apps/api/src/session-explorer/namespace-validation.ts b/apps/api/src/session-explorer/namespace-validation.ts new file mode 100644 index 000000000..4d703ef58 --- /dev/null +++ b/apps/api/src/session-explorer/namespace-validation.ts @@ -0,0 +1,68 @@ +import type { PresetId, SessionNamespaceInput } from '@betterdb/shared'; +import { validateCustomSchema } from './presets/custom'; + +export const DEFAULT_OVERSIZE_THRESHOLD_BYTES = 1048576; + +const PRESET_IDS: readonly PresetId[] = [ + 'langchain-community', + 'langchain-js', + 'llamaindex', + 'agent-cache', + 'custom', + 'raw', +]; + +const GLOB_METACHARACTERS = new Set(['*', '?', '[']); + +const hasLiteralPrefix = (pattern: string): boolean => + pattern.length > 0 && !GLOB_METACHARACTERS.has(pattern[0]); + +const isRecord = (v: unknown): v is Record => + typeof v === 'object' && v !== null && !Array.isArray(v); + +export function validateNamespaceInput( + input: unknown, +): { ok: true; value: SessionNamespaceInput } | { ok: false; errors: string[] } { + if (!isRecord(input)) return { ok: false, errors: ['Body must be an object'] }; + + const errors: string[] = []; + + const name = typeof input.name === 'string' ? input.name.trim() : ''; + if (name === '') errors.push('Name is required'); + + const pattern = typeof input.pattern === 'string' ? input.pattern.trim() : ''; + if (!hasLiteralPrefix(pattern)) errors.push('Pattern must include a literal prefix'); + if (/\s/.test(pattern)) errors.push('Pattern must not contain whitespace'); + + const preset = PRESET_IDS.find((id) => id === input.preset); + if (!preset) errors.push('Unknown preset'); + + const threshold = input.oversizeThresholdBytes ?? DEFAULT_OVERSIZE_THRESHOLD_BYTES; + const thresholdValid = + typeof threshold === 'number' && Number.isSafeInteger(threshold) && threshold > 0; + if (!thresholdValid) errors.push('Oversize threshold must be a positive integer'); + + let customSchema: SessionNamespaceInput['customSchema'] = null; + if (preset === 'custom') { + if (input.customSchema === null || input.customSchema === undefined) { + errors.push('Custom preset requires a schema'); + } else { + const result = validateCustomSchema(input.customSchema); + if (result.ok) customSchema = result.schema; + else errors.push(...result.errors); + } + } + + if (errors.length > 0 || !preset || !thresholdValid) return { ok: false, errors }; + + return { + ok: true, + value: { + name, + pattern, + preset, + customSchema, + oversizeThresholdBytes: threshold, + }, + }; +} diff --git a/apps/api/src/session-explorer/presets/text.ts b/apps/api/src/session-explorer/presets/text.ts index 4baca7ebf..afb550dde 100644 --- a/apps/api/src/session-explorer/presets/text.ts +++ b/apps/api/src/session-explorer/presets/text.ts @@ -15,8 +15,9 @@ const tryDecode = (buf: Buffer): string | null => { export function decodeValue( buf: Buffer, cap: number, + more = false, ): { value: string | { base64: string }; truncated: boolean } { - const truncated = buf.length > cap; + const truncated = more || buf.length > cap; const slice = truncated ? buf.subarray(0, cap) : buf; const full = tryDecode(slice); if (full !== null) { diff --git a/apps/api/src/session-explorer/primary-nodes.ts b/apps/api/src/session-explorer/primary-nodes.ts new file mode 100644 index 000000000..6d201d269 --- /dev/null +++ b/apps/api/src/session-explorer/primary-nodes.ts @@ -0,0 +1,112 @@ +import Valkey from 'iovalkey'; +import type { DatabasePort } from '../common/interfaces/database-port.interface'; +import type { ClusterNode } from '../common/types/metrics.types'; +import { tlsIdentityOptions } from '../database/adapters/tls-servername'; +import { parseNodeAddress } from '../topology/membership-diff'; + +const NODE_TIMEOUT_MS = 5000; + +export interface PrimaryRoute { + client: Valkey; + slots: number[][]; +} + +export interface PrimaryClients { + clients: Valkey[]; + failed: number; + routes: PrimaryRoute[] | null; + release: () => Promise; +} + +const isServingPrimary = (node: ClusterNode) => + node.flags.includes('master') && node.slots.length > 0; + +function nodeClient(parent: Valkey, node: ClusterNode): Valkey | null { + const address = parseNodeAddress(node.address); + if (!address) return null; + const { username, password, tls } = parent.options; + const { + servername: _servername, + checkServerIdentity: _checkServerIdentity, + ...nodeTls + } = tls ?? {}; + const client = new Valkey({ + host: address.host, + port: address.port, + username, + password, + tls: tls ? { ...nodeTls, ...tlsIdentityOptions(address.host, node.hostname) } : undefined, + lazyConnect: true, + enableOfflineQueue: false, + enableReadyCheck: false, + maxRetriesPerRequest: 0, + retryStrategy: () => null, + connectTimeout: NODE_TIMEOUT_MS, + commandTimeout: NODE_TIMEOUT_MS, + connectionName: 'BetterDB-Monitor-SessionExplorer', + }); + client.on('error', () => undefined); + return client; +} + +async function connectNode(parent: Valkey, node: ClusterNode): Promise { + let client: Valkey | null = null; + try { + client = nodeClient(parent, node); + if (!client) return null; + await client.connect(); + return client; + } catch { + client?.disconnect(); + return null; + } +} + +export async function primaryClients(db: DatabasePort): Promise { + const parent = db.getClient(); + if (!db.getCapabilities().clusterEnabled) { + return { clients: [parent], failed: 0, routes: null, release: async () => undefined }; + } + + const primaries = (await db.getClusterNodes()).filter(isServingPrimary); + const owned: Valkey[] = []; + const clients: Valkey[] = []; + const routes: PrimaryRoute[] = []; + let failed = 0; + + const connected = await Promise.all( + primaries.map((node) => + node.flags.includes('myself') ? Promise.resolve(parent) : connectNode(parent, node), + ), + ); + connected.forEach((client, i) => { + if (!client) { + failed++; + return; + } + clients.push(client); + routes.push({ client, slots: primaries[i].slots }); + if (client !== parent) owned.push(client); + }); + + return { + clients, + failed, + routes, + release: async () => { + for (const client of owned) client.disconnect(); + }, + }; +} + +export async function withPrimaries( + db: DatabasePort, + run: (primaries: PrimaryClients) => Promise, +): Promise { + const primaries = await primaryClients(db); + try { + return await run(primaries); + } finally { + await primaries.release(); + } +} diff --git a/apps/api/src/session-explorer/session-explorer.controller.ts b/apps/api/src/session-explorer/session-explorer.controller.ts new file mode 100644 index 000000000..2e28a2e6c --- /dev/null +++ b/apps/api/src/session-explorer/session-explorer.controller.ts @@ -0,0 +1,151 @@ +import { Readable } from 'stream'; +import { + Body, + Controller, + Delete, + Get, + HttpCode, + Param, + Post, + Put, + Res, + UseGuards, +} from '@nestjs/common'; +import { ApiHeader, ApiOperation, ApiTags } from '@nestjs/swagger'; +import type { FastifyReply } from 'fastify'; +import { Feature } from '@betterdb/shared'; +import type { + SessionDetail, + SessionListResponse, + SessionNamespace, + SessionOversizeResponse, + SessionPreview, +} from '@betterdb/shared'; +import { LicenseGuard } from '@proprietary/licenses'; +import { RequiresFeature } from '@proprietary/licenses/requires-feature.decorator'; +import { ConnectionId } from '../common/decorators/connection-id.decorator'; +import { ENV_DEFAULT_ID } from '../connections/connection-registry.service'; +import { + LiveConnectionGuard, + UseHeaderConnectionId, +} from '../external-metrics/live-connection.guard'; +import { SessionCapture, SessionExplorerService } from './session-explorer.service'; + +const MAX_FILENAME_ID = 100; + +const exportFilename = (sessionId: string): string => + `session-${sessionId.replace(/[^A-Za-z0-9._-]/g, '_').slice(0, MAX_FILENAME_ID)}.json`; + +function* exportChunks({ namespace, keys, detail }: SessionCapture): Generator { + const header = { + exportedAt: new Date().toISOString(), + namespace: { name: namespace.name, pattern: namespace.pattern, preset: namespace.preset }, + sessionId: detail.sessionId, + }; + yield `${JSON.stringify(header).slice(0, -1)},"keys":[`; + for (let i = 0; i < keys.length; i++) { + yield `${i > 0 ? ',' : ''}${JSON.stringify(keys[i])}`; + } + yield `],"parsed":${JSON.stringify(detail)}}`; +} + +@ApiTags('session-explorer') +@ApiHeader({ name: 'x-connection-id', required: false }) +@UseHeaderConnectionId() +@Controller('session-explorer') +export class SessionExplorerController { + constructor(private readonly service: SessionExplorerService) {} + + @Get('namespaces') + @ApiOperation({ summary: 'List session namespaces for a connection' }) + listNamespaces(@ConnectionId() connectionId?: string): Promise { + return this.service.listNamespaces(connectionId || ENV_DEFAULT_ID); + } + + @Post('namespaces') + @ApiOperation({ summary: 'Create a session namespace' }) + createNamespace( + @Body() body: unknown, + @ConnectionId() connectionId?: string, + ): Promise { + return this.service.createNamespace(connectionId || ENV_DEFAULT_ID, body); + } + + @Post('namespaces/preview') + @HttpCode(200) + @UseGuards(LiveConnectionGuard) + @ApiOperation({ summary: 'Parse one sampled key against an unsaved namespace' }) + preview(@Body() body: unknown, @ConnectionId() connectionId?: string): Promise { + return this.service.preview(connectionId || ENV_DEFAULT_ID, body); + } + + @Put('namespaces/:id') + @ApiOperation({ summary: 'Update a session namespace' }) + updateNamespace( + @Param('id') id: string, + @Body() body: unknown, + @ConnectionId() connectionId?: string, + ): Promise { + return this.service.updateNamespace(connectionId || ENV_DEFAULT_ID, id, body); + } + + @Delete('namespaces/:id') + @HttpCode(204) + @ApiOperation({ summary: 'Delete a session namespace' }) + deleteNamespace(@Param('id') id: string, @ConnectionId() connectionId?: string): Promise { + return this.service.deleteNamespace(connectionId || ENV_DEFAULT_ID, id); + } + + @Get('namespaces/:id/sessions') + @UseGuards(LiveConnectionGuard) + @ApiOperation({ summary: 'List sessions stored under a namespace' }) + listSessions( + @Param('id') id: string, + @ConnectionId() connectionId?: string, + ): Promise { + return this.service.listSessions(connectionId || ENV_DEFAULT_ID, id); + } + + @Get('namespaces/:id/oversize') + @UseGuards(LiveConnectionGuard, LicenseGuard) + @RequiresFeature(Feature.SESSION_OVERSIZE_FLAGS) + @ApiOperation({ summary: 'Sessions whose memory exceeds the namespace threshold' }) + oversizeSessions( + @Param('id') id: string, + @ConnectionId() connectionId?: string, + ): Promise { + return this.service.oversizeSessions(connectionId || ENV_DEFAULT_ID, id); + } + + @Get('namespaces/:id/sessions/:sessionId') + @UseGuards(LiveConnectionGuard) + @ApiOperation({ summary: 'Read and parse one session' }) + getSession( + @Param('id') id: string, + @Param('sessionId') sessionId: string, + @ConnectionId() connectionId?: string, + ): Promise { + return this.service.getSession(connectionId || ENV_DEFAULT_ID, id, sessionId); + } + + @Get('namespaces/:id/sessions/:sessionId/export') + @UseGuards(LiveConnectionGuard) + @ApiOperation({ summary: 'Export one session as stored, as a JSON attachment' }) + async exportSession( + @Param('id') id: string, + @Param('sessionId') sessionId: string, + @Res({ passthrough: false }) reply: FastifyReply, + @ConnectionId() connectionId?: string, + ): Promise { + const capture = await this.service.captureSession( + connectionId || ENV_DEFAULT_ID, + id, + sessionId, + ); + await reply + .status(200) + .header('content-type', 'application/json; charset=utf-8') + .header('content-disposition', `attachment; filename="${exportFilename(sessionId)}"`) + .send(Readable.from(exportChunks(capture))); + } +} diff --git a/apps/api/src/session-explorer/session-explorer.module.ts b/apps/api/src/session-explorer/session-explorer.module.ts new file mode 100644 index 000000000..44c031426 --- /dev/null +++ b/apps/api/src/session-explorer/session-explorer.module.ts @@ -0,0 +1,12 @@ +import { Module } from '@nestjs/common'; +import { StorageModule } from '../storage/storage.module'; +import { ConnectionsModule } from '../connections/connections.module'; +import { SessionExplorerService } from './session-explorer.service'; +import { SessionExplorerController } from './session-explorer.controller'; + +@Module({ + imports: [StorageModule, ConnectionsModule], + providers: [SessionExplorerService], + controllers: [SessionExplorerController], +}) +export class SessionExplorerModule {} diff --git a/apps/api/src/session-explorer/session-explorer.service.ts b/apps/api/src/session-explorer/session-explorer.service.ts new file mode 100644 index 000000000..0ccf04cd2 --- /dev/null +++ b/apps/api/src/session-explorer/session-explorer.service.ts @@ -0,0 +1,184 @@ +import { randomUUID } from 'crypto'; +import { BadRequestException, Inject, Injectable, NotFoundException } from '@nestjs/common'; +import type { + CapturedKey, + SessionDetail, + SessionListResponse, + SessionNamespace, + SessionNamespaceInput, + SessionOversizeResponse, + SessionPreview, +} from '@betterdb/shared'; +import type { StoragePort } from '../common/interfaces/storage-port.interface'; +import { ConnectionRegistry } from '../connections/connection-registry.service'; +import { validateNamespaceInput } from './namespace-validation'; +import { getPreset } from './presets/registry'; +import { estimateTokens } from './presets/text'; +import type { PresetDefinition } from './presets/types'; +import { withPrimaries } from './primary-nodes'; +import { keysForSession, scanSessions } from './session-scanner'; +import { readSessionKeys } from './session-reader'; + +const PREVIEW_SCAN_COUNT = 1000; + +export interface SessionCapture { + namespace: SessionNamespace; + keys: CapturedKey[]; + detail: SessionDetail; +} + +const minPositive = (values: number[]): number | null => { + const positive = values.filter((v) => v > 0); + return positive.length > 0 ? Math.min(...positive) : null; +}; + +@Injectable() +export class SessionExplorerService { + constructor( + private readonly registry: ConnectionRegistry, + @Inject('STORAGE_CLIENT') private readonly storage: StoragePort, + ) {} + + listNamespaces(connectionId: string): Promise { + return this.storage.listSessionNamespaces(connectionId); + } + + async createNamespace(connectionId: string, body: unknown): Promise { + const input = this.validate(body); + const now = Date.now(); + return this.storage.saveSessionNamespace({ + ...input, + id: randomUUID(), + connectionId, + createdAt: now, + updatedAt: now, + }); + } + + async updateNamespace(connectionId: string, id: string, body: unknown): Promise { + const input = this.validate(body); + const existing = await this.namespace(connectionId, id); + return this.storage.saveSessionNamespace({ + ...existing, + ...input, + updatedAt: Math.max(Date.now(), existing.updatedAt + 1), + }); + } + + async deleteNamespace(connectionId: string, id: string): Promise { + if (!(await this.storage.deleteSessionNamespace(connectionId, id))) { + throw new NotFoundException('Namespace not found'); + } + } + + async preview(connectionId: string, body: unknown): Promise { + const input = this.validate(body); + const preset = this.presetOf(input); + const db = this.registry.get(connectionId); + const sampleKey = await withPrimaries(db, async ({ clients }) => { + for (const client of clients) { + const [, page] = await client.scan('0', 'MATCH', input.pattern, 'COUNT', PREVIEW_SCAN_COUNT); + const key = page.find((k) => preset.sessionIdOf(k) !== null) ?? page[0]; + if (key !== undefined) return key; + } + return null; + }); + if (sampleKey === null) return { sampleKey: null, view: null }; + const read = await readSessionKeys(db, [sampleKey]); + const captured = read.keys[0]; + return { sampleKey, view: captured ? preset.parseKey(captured) : null }; + } + + async listSessions(connectionId: string, id: string): Promise { + const namespace = await this.namespace(connectionId, id); + return scanSessions(this.registry.get(connectionId), namespace.pattern, this.presetOf(namespace)); + } + + async oversizeSessions(connectionId: string, id: string): Promise { + const namespace = await this.namespace(connectionId, id); + const list = await scanSessions( + this.registry.get(connectionId), + namespace.pattern, + this.presetOf(namespace), + ); + const thresholdBytes = namespace.oversizeThresholdBytes; + const oversize = list.sessions.filter((s) => s.memoryBytes > thresholdBytes); + return { + thresholdBytes, + sessionIds: oversize.map((s) => s.sessionId), + unmeasuredSessionIds: list.sessions + .filter((s) => s.unmeasuredKeys > 0 && !oversize.includes(s)) + .map((s) => s.sessionId), + truncated: list.truncated, + partial: list.partial, + failedNodes: list.failedNodes, + }; + } + + async getSession(connectionId: string, id: string, sessionId: string): Promise { + return (await this.captureSession(connectionId, id, sessionId)).detail; + } + + async captureSession(connectionId: string, id: string, sessionId: string): Promise { + const namespace = await this.namespace(connectionId, id); + const preset = this.presetOf(namespace); + const db = this.registry.get(connectionId); + const scan = await keysForSession(db, namespace.pattern, preset, sessionId); + if (scan.keys.length === 0) throw new NotFoundException('Session not found'); + const read = await readSessionKeys(db, scan.keys); + if (read.keys.length === 0) throw new NotFoundException('Session not found'); + + const views = read.keys.map((k) => preset.parseKey(k)); + const messages = preset.mergeSession(views); + const unparsedKeys = views.filter((v) => !v.parsed).length; + const truncated = scan.truncated || read.truncated; + const failedNodes = scan.failedNodes + read.failedNodes; + const complete = + unparsedKeys === 0 && + !truncated && + failedNodes === 0 && + read.missingKeys === 0 && + read.keys.every((k) => !k.truncated); + + return { + namespace, + keys: read.keys, + detail: { + sessionId, + preset: namespace.preset, + keys: views, + messages, + estimatedTokens: estimateTokens(messages), + ttlMs: minPositive(read.keys.map((k) => k.pttl)), + memoryBytes: read.keys.reduce((sum, k) => sum + (k.bytes ?? 0), 0), + complete, + unparsedKeys, + truncated, + failedNodes, + missingKeys: read.missingKeys, + }, + }; + } + + private validate(body: unknown): SessionNamespaceInput { + const result = validateNamespaceInput(body); + if (!result.ok) { + throw new BadRequestException({ + statusCode: 400, + message: 'Invalid namespace', + errors: result.errors, + }); + } + return result.value; + } + + private async namespace(connectionId: string, id: string): Promise { + const namespace = await this.storage.getSessionNamespace(connectionId, id); + if (!namespace) throw new NotFoundException('Namespace not found'); + return namespace; + } + + private presetOf(ns: Pick): PresetDefinition { + return getPreset(ns.preset, ns.customSchema); + } +} diff --git a/apps/api/src/session-explorer/session-reader.ts b/apps/api/src/session-explorer/session-reader.ts new file mode 100644 index 000000000..23648b203 --- /dev/null +++ b/apps/api/src/session-explorer/session-reader.ts @@ -0,0 +1,430 @@ +import type Valkey from 'iovalkey'; +import type { ChainableCommander } from 'iovalkey'; +import calculateSlot from 'cluster-key-slot'; +import type { CapturedKey, ValkeyValue } from '@betterdb/shared'; +import type { DatabasePort } from '../common/interfaces/database-port.interface'; +import { decodeValue } from './presets/text'; +import { PrimaryClients, withPrimaries } from './primary-nodes'; + +export const DEFAULT_BYTE_CAP = 2 * 1024 * 1024; +export const DEFAULT_TOTAL_BYTE_CAP = 32 * 1024 * 1024; +export const DEFAULT_LIST_WINDOW = 500; +export const DEFAULT_READ_BATCH = 1000; +const JSON_MEMORY_HEADROOM = 4; + +type PipelineResult = [Error | null, unknown] | undefined; + +export interface SessionKeysRead { + keys: CapturedKey[]; + truncated: boolean; + failedNodes: number; + missingKeys: number; +} + +export interface ReadOptions { + byteCap?: number; + totalByteCap?: number; + listWindow?: number; + batch?: number; +} + +interface ReadOpts { + byteCap: number; + totalByteCap: number; + listWindow: number; + batch: number; +} + +interface Located { + key: string; + type: string; + pttl: number; + memory: number | null; +} + +interface ReadResult { + value: ValkeyValue; + truncated: boolean; + bytes: number; +} + +class KeyReadError extends Error {} + +const resultOf = (r: PipelineResult): unknown => { + if (!r || r[0]) throw new KeyReadError(); + return r[1]; +}; + +const numberOf = (r: PipelineResult): number | null => + r && !r[0] && typeof r[1] === 'number' ? r[1] : null; + +const decoded = (buf: Buffer, cap: number) => decodeValue(buf, cap).value; + +function decodeCut(buf: Buffer, cap: number, cut: boolean): ReadResult { + return { ...decodeValue(buf, cap, cut), bytes: Math.min(buf.length, cap) }; +} + +class Accumulator { + items: T[] = []; + bytes = 0; + full = false; + + constructor(private readonly cap: number) {} + + add(size: number, item: () => T): boolean { + if (this.bytes + size > this.cap) { + this.full = true; + return false; + } + this.bytes += size; + this.items.push(item()); + return true; + } +} + +function routeKeys( + primaries: PrimaryClients, + keys: string[], +): { groups: Map; unrouted: number } { + const groups = new Map(); + let unrouted = 0; + const add = (client: Valkey, key: string) => { + const list = groups.get(client) ?? []; + list.push(key); + groups.set(client, list); + }; + for (const key of keys) { + if (!primaries.routes) { + add(primaries.clients[0], key); + continue; + } + const slot = calculateSlot(key); + const route = primaries.routes.find((r) => + r.slots.some(([start, end]) => slot >= start && slot <= end), + ); + if (route) add(route.client, key); + else unrouted++; + } + return { groups, unrouted }; +} + +async function locate( + client: Valkey, + keys: string[], + batch: number, +): Promise<{ located: Located[]; missing: number }> { + const located: Located[] = []; + let missing = 0; + for (let start = 0; start < keys.length; start += batch) { + const slice = keys.slice(start, start + batch); + const pipeline = client.pipeline(); + for (const key of slice) { + pipeline.type(key); + pipeline.pttl(key); + pipeline.memory('USAGE', key); + } + const results = (await pipeline.exec()) ?? []; + slice.forEach((key, i) => { + const [type, pttl, memory] = results.slice(i * 3, i * 3 + 3) as PipelineResult[]; + if (!type || type[0] || typeof type[1] !== 'string') { + missing++; + return; + } + if (type[1] === 'none') return; + located.push({ key, type: type[1], pttl: numberOf(pttl) ?? -1, memory: numberOf(memory) }); + }); + } + return { located, missing }; +} + +function queueFirstRead(pipeline: ChainableCommander, at: Located, opts: ReadOpts): number { + const { key, memory } = at; + switch (at.type) { + case 'string': + pipeline.strlen(key); + pipeline.getrangeBuffer(key, 0, opts.byteCap - 1); + return 2; + case 'list': + pipeline.llen(key); + pipeline.lrangeBuffer(key, 0, opts.listWindow - 1); + return 2; + case 'hash': + if (memory === null || memory > opts.byteCap) return 0; + pipeline.hgetallBuffer(key); + return 1; + case 'zset': + pipeline.zcard(key); + pipeline.zrangeBuffer(key, 0, String(opts.listWindow - 1), 'WITHSCORES'); + return 2; + case 'ReJSON-RL': + if (memory === null) { + pipeline.call('JSON.DEBUG', 'MEMORY', key); + return 1; + } + if (memory > opts.byteCap * JSON_MEMORY_HEADROOM) return 0; + pipeline.callBuffer('JSON.GET', key); + return 1; + case 'stream': + pipeline.xrangeBuffer(key, '-', '+', 'COUNT', opts.listWindow + 1); + return 1; + default: + return 0; + } +} + +function readString(results: PipelineResult[], cap: number): ReadResult { + const length = resultOf(results[0]) as number; + const buf = resultOf(results[1]) as Buffer; + return decodeCut(buf, cap, length > buf.length); +} + +async function readList( + client: Valkey, + at: Located, + results: PipelineResult[], + cap: number, + opts: ReadOpts, +): Promise { + const length = resultOf(results[0]) as number; + let window = resultOf(results[1]) as Buffer[]; + const acc = new Accumulator(cap); + for (let start = 0; ; ) { + for (const item of window) { + if (!acc.add(item.length, () => decoded(item, cap))) break; + } + start += opts.listWindow; + if (acc.full || window.length < opts.listWindow || start >= length) break; + window = await client.lrangeBuffer(at.key, start, start + opts.listWindow - 1); + } + return { value: acc.items, truncated: acc.full, bytes: acc.bytes }; +} + +async function readHash( + client: Valkey, + at: Located, + results: PipelineResult[], + cap: number, + opts: ReadOpts, +): Promise { + const acc = new Accumulator<[string, ValkeyValue]>(cap); + if (results.length > 0) { + const all = resultOf(results[0]) as Record; + for (const [field, v] of Object.entries(all)) { + if (!acc.add(Buffer.byteLength(field) + v.length, () => [field, decoded(v, cap)])) break; + } + } else { + let cursor = '0'; + do { + const [next, flat] = await client.hscanBuffer(at.key, cursor, 'COUNT', opts.listWindow); + cursor = next.toString(); + for (let i = 0; i + 1 < flat.length; i += 2) { + const [field, v] = [flat[i], flat[i + 1]]; + if (!acc.add(field.length + v.length, () => [field.toString(), decoded(v, cap)])) break; + } + } while (cursor !== '0' && !acc.full); + } + return { value: Object.fromEntries(acc.items), truncated: acc.full, bytes: acc.bytes }; +} + +async function readZset( + client: Valkey, + at: Located, + results: PipelineResult[], + cap: number, + opts: ReadOpts, +): Promise { + const length = resultOf(results[0]) as number; + let flat = resultOf(results[1]) as Buffer[]; + const acc = new Accumulator<[ValkeyValue, number]>(cap); + for (let start = 0; ; ) { + for (let i = 0; i + 1 < flat.length; i += 2) { + const [member, score] = [flat[i], flat[i + 1]]; + if (!acc.add(member.length, () => [decoded(member, cap), Number(score.toString())])) break; + } + start += opts.listWindow; + if (acc.full || flat.length < opts.listWindow * 2 || start >= length) break; + flat = await client.zrangeBuffer( + at.key, + start, + String(start + opts.listWindow - 1), + 'WITHSCORES', + ); + } + return { value: acc.items, truncated: acc.full, bytes: acc.bytes }; +} + +const UNREAD_JSON: ReadResult = { value: '', truncated: true, bytes: 0 }; + +async function readJson( + client: Valkey, + at: Located, + results: PipelineResult[], + cap: number, + opts: ReadOpts, +): Promise { + const limit = opts.byteCap * JSON_MEMORY_HEADROOM; + let buf: Buffer | null; + if (at.memory === null) { + const size = numberOf(results[0]); + if (size === null || size > limit) return UNREAD_JSON; + buf = (await client.callBuffer('JSON.GET', at.key)) as Buffer | null; + } else if (at.memory > limit) { + return UNREAD_JSON; + } else { + buf = resultOf(results[0]) as Buffer | null; + } + if (buf === null) return null; + return { ...decodeValue(buf, cap), bytes: Math.min(buf.length, cap) }; +} + +function readStream(results: PipelineResult[], cap: number, opts: ReadOpts): ReadResult { + const entries = resultOf(results[0]) as Array<[Buffer, Buffer[]]>; + const acc = new Accumulator<[string, Record]>(cap); + for (const [id, flat] of entries.slice(0, opts.listWindow)) { + const fields: Record = {}; + let size = 0; + for (let i = 0; i + 1 < flat.length; i += 2) { + size += flat[i].length + flat[i + 1].length; + fields[flat[i].toString()] = decoded(flat[i + 1], cap); + } + if (!acc.add(size, () => [id.toString(), fields])) break; + } + return { + value: acc.items, + truncated: acc.full || entries.length > opts.listWindow, + bytes: acc.bytes, + }; +} + +function finishRead( + client: Valkey, + at: Located, + results: PipelineResult[], + cap: number, + opts: ReadOpts, +): Promise | ReadResult | null { + switch (at.type) { + case 'string': + return readString(results, cap); + case 'list': + return readList(client, at, results, cap, opts); + case 'hash': + return readHash(client, at, results, cap, opts); + case 'zset': + return readZset(client, at, results, cap, opts); + case 'ReJSON-RL': + return readJson(client, at, results, cap, opts); + case 'stream': + return readStream(results, cap, opts); + default: + return { value: null, truncated: false, bytes: 0 }; + } +} + +const estimate = (at: Located, opts: ReadOpts) => + Math.max(1, Math.min(at.memory ?? opts.byteCap, opts.byteCap)); + +interface ReadState { + remaining: number; + exhausted: boolean; + missing: number; + captured: Map; +} + +async function readNode( + client: Valkey, + located: Located[], + opts: ReadOpts, + state: ReadState, +): Promise { + let next = 0; + while (next < located.length) { + if (state.remaining <= 0) { + state.exhausted = true; + return; + } + const batch: Located[] = []; + let reserved = 0; + while (next < located.length && batch.length < opts.batch) { + const cost = estimate(located[next], opts); + if (batch.length > 0 && reserved + cost > state.remaining) break; + reserved += cost; + batch.push(located[next++]); + } + + const pipeline = client.pipeline(); + const counts = batch.map((at) => queueFirstRead(pipeline, at, opts)); + const results = counts.some((n) => n > 0) ? ((await pipeline.exec()) ?? []) : []; + + let offset = 0; + for (const [i, at] of batch.entries()) { + const own = results.slice(offset, offset + counts[i]) as PipelineResult[]; + offset += counts[i]; + try { + const read = await finishRead(client, at, own, Math.min(opts.byteCap, state.remaining), opts); + if (!read) continue; + state.remaining -= read.bytes; + state.captured.set(at.key, { + key: at.key, + type: at.type, + pttl: at.pttl, + value: read.value, + ...(read.truncated ? { truncated: true } : {}), + ...(at.memory !== null ? { bytes: at.memory } : {}), + }); + } catch { + state.missing++; + } + } + } +} + +export async function readSessionKeys( + db: DatabasePort, + keys: string[], + options: ReadOptions = {}, +): Promise { + const opts: ReadOpts = { + byteCap: options.byteCap ?? DEFAULT_BYTE_CAP, + totalByteCap: options.totalByteCap ?? DEFAULT_TOTAL_BYTE_CAP, + listWindow: options.listWindow ?? DEFAULT_LIST_WINDOW, + batch: options.batch ?? DEFAULT_READ_BATCH, + }; + return withPrimaries(db, async (primaries) => { + const { groups, unrouted } = routeKeys(primaries, keys); + const state: ReadState = { + remaining: opts.totalByteCap, + exhausted: false, + missing: unrouted, + captured: new Map(), + }; + let failedNodes = primaries.failed; + let lastError: unknown; + let succeeded = 0; + + for (const [client, nodeKeys] of groups) { + if (state.exhausted) break; + try { + const { located, missing } = await locate(client, nodeKeys, opts.batch); + state.missing += missing; + await readNode(client, located, opts, state); + succeeded++; + } catch (err) { + failedNodes++; + lastError = err; + } + } + + if (succeeded === 0 && lastError !== undefined) throw lastError; + + return { + keys: keys.flatMap((key) => { + const captured = state.captured.get(key); + return captured ? [captured] : []; + }), + truncated: state.exhausted, + failedNodes, + missingKeys: state.missing, + }; + }); +} diff --git a/apps/api/src/session-explorer/session-scanner.ts b/apps/api/src/session-explorer/session-scanner.ts new file mode 100644 index 000000000..6c484eaad --- /dev/null +++ b/apps/api/src/session-explorer/session-scanner.ts @@ -0,0 +1,215 @@ +import type Valkey from 'iovalkey'; +import type { SessionListResponse, SessionSummary } from '@betterdb/shared'; +import type { DatabasePort } from '../common/interfaces/database-port.interface'; +import type { PresetDefinition } from './presets/types'; +import { withPrimaries } from './primary-nodes'; + +export const DEFAULT_KEY_CAP = 10000; +export const DEFAULT_BATCH = 1000; +const SCAN_COUNT = 1000; + +type PipelineResult = [Error | null, unknown] | undefined; + +interface KeyMeta { + memory: number | null; + idle: number | null; + pttl: number; +} + +interface ScanOutcome { + keys: string[]; + exhausted: boolean; +} + +interface NodeKeys { + client: Valkey; + keys: string[]; +} + +interface ScanAllOutcome { + nodes: NodeKeys[]; + scannedKeys: number; + truncated: boolean; + failed: number; +} + +export interface SessionKeysResult { + keys: string[]; + truncated: boolean; + failedNodes: number; +} + +type KeyFilter = (key: string) => boolean; + +const acceptAll: KeyFilter = () => true; + +async function scanNode( + client: Valkey, + pattern: string, + limit: number, + accept: KeyFilter, +): Promise { + const seen = new Set(); + let cursor = '0'; + do { + const [next, page] = await client.scan(cursor, 'MATCH', pattern, 'COUNT', SCAN_COUNT); + cursor = next; + for (const key of page) { + if (seen.size >= limit) break; + if (accept(key)) seen.add(key); + } + } while (cursor !== '0' && seen.size < limit); + return { keys: [...seen], exhausted: cursor === '0' }; +} + +async function scanAll( + clients: Valkey[], + pattern: string, + keyCap: number, + accept: KeyFilter = acceptAll, +): Promise { + const nodes: NodeKeys[] = []; + let scannedKeys = 0; + let truncated = false; + let failed = 0; + let lastError: unknown; + + for (const [index, client] of clients.entries()) { + if (scannedKeys >= keyCap) { + truncated = true; + break; + } + try { + const { keys, exhausted } = await scanNode(client, pattern, keyCap - scannedKeys, accept); + scannedKeys += keys.length; + if (!exhausted || (scannedKeys >= keyCap && index < clients.length - 1)) truncated = true; + nodes.push({ client, keys }); + } catch (err) { + failed++; + lastError = err; + } + } + + if (nodes.length === 0 && lastError !== undefined) throw lastError; + return { nodes, scannedKeys, truncated, failed }; +} + +const numberOf = (r: PipelineResult): number | null => + r && !r[0] && typeof r[1] === 'number' ? r[1] : null; + +async function fetchMeta( + client: Valkey, + keys: string[], + batch: number, +): Promise> { + const meta = new Map(); + for (let start = 0; start < keys.length; start += batch) { + const slice = keys.slice(start, start + batch); + const pipeline = client.pipeline(); + for (const key of slice) { + pipeline.type(key); + pipeline.pttl(key); + pipeline.memory('USAGE', key); + pipeline.object('IDLETIME', key); + } + const results = (await pipeline.exec()) ?? []; + slice.forEach((key, i) => { + const [type, pttl, memory, idle] = results.slice(i * 4, i * 4 + 4) as PipelineResult[]; + if (!type || type[0] || type[1] === 'none') return; + const ttl = numberOf(pttl); + if (ttl === -2) return; + meta.set(key, { memory: numberOf(memory), idle: numberOf(idle), pttl: ttl ?? -1 }); + }); + } + return meta; +} + +const minOrNull = (a: number | null, b: number | null) => + a === null ? b : b === null ? a : Math.min(a, b); + +function summarise(sessionId: string, keys: KeyMeta[]): SessionSummary { + return keys.reduce( + (acc, k) => ({ + ...acc, + memoryBytes: acc.memoryBytes + (k.memory ?? 0), + unmeasuredKeys: acc.unmeasuredKeys + (k.memory === null ? 1 : 0), + idleSeconds: minOrNull(acc.idleSeconds, k.idle), + ttlMs: k.pttl > 0 ? minOrNull(acc.ttlMs, k.pttl) : acc.ttlMs, + }), + { + sessionId, + keyCount: keys.length, + memoryBytes: 0, + unmeasuredKeys: 0, + idleSeconds: null, + ttlMs: null, + }, + ); +} + +export async function scanSessions( + db: DatabasePort, + pattern: string, + preset: PresetDefinition, + opts: { keyCap?: number; batch?: number } = {}, +): Promise { + const keyCap = opts.keyCap ?? DEFAULT_KEY_CAP; + const batch = opts.batch ?? DEFAULT_BATCH; + + return withPrimaries(db, async ({ clients, failed: unreachable }) => { + const scan = await scanAll(clients, pattern, keyCap); + const grouped = new Map(); + let failed = unreachable + scan.failed; + let lastError: unknown; + let succeeded = 0; + + for (const { client, keys } of scan.nodes) { + try { + const sessionKeys = keys.filter((key) => preset.sessionIdOf(key) !== null); + const meta = await fetchMeta(client, sessionKeys, batch); + for (const [key, m] of meta) { + const sessionId = preset.sessionIdOf(key) as string; + const list = grouped.get(sessionId) ?? []; + list.push(m); + grouped.set(sessionId, list); + } + succeeded++; + } catch (err) { + failed++; + lastError = err; + } + } + + if (succeeded === 0 && lastError !== undefined) throw lastError; + + return { + sessions: [...grouped].map(([id, keys]) => summarise(id, keys)), + scannedKeys: scan.scannedKeys, + truncated: scan.truncated, + partial: failed > 0, + failedNodes: failed, + }; + }); +} + +export async function keysForSession( + db: DatabasePort, + pattern: string, + preset: PresetDefinition, + sessionId: string, + keyCap: number = DEFAULT_KEY_CAP, +): Promise { + return withPrimaries(db, async ({ clients, failed }) => { + const scan = await scanAll( + clients, + pattern, + keyCap, + (key) => preset.sessionIdOf(key) === sessionId, + ); + return { + keys: scan.nodes.flatMap(({ keys }) => keys), + truncated: scan.truncated, + failedNodes: failed + scan.failed, + }; + }); +} diff --git a/apps/api/src/storage/adapters/memory.adapter.ts b/apps/api/src/storage/adapters/memory.adapter.ts index df935fd08..7a2d2874e 100644 --- a/apps/api/src/storage/adapters/memory.adapter.ts +++ b/apps/api/src/storage/adapters/memory.adapter.ts @@ -100,6 +100,7 @@ import type { ReadinessDimension, ReadinessDimensionKey, ScalingReadinessSettings, + SessionNamespace, } from '@betterdb/shared'; const NULL_SUB_DISCRIMINATOR = '__betterdb_null__'; @@ -1208,6 +1209,36 @@ export class MemoryAdapter implements StoragePort, RawDatabaseHandleProvider { return settings; } + private sessionNamespaces = new Map(); + + async listSessionNamespaces(connectionId: string): Promise { + return [...this.sessionNamespaces.values()] + .filter((n) => n.connectionId === connectionId) + .sort((a, b) => a.createdAt - b.createdAt) + .map((n) => ({ ...n })); + } + + async getSessionNamespace(connectionId: string, id: string): Promise { + const ns = this.sessionNamespaces.get(id); + return ns && ns.connectionId === connectionId ? { ...ns } : null; + } + + async saveSessionNamespace(ns: SessionNamespace): Promise { + const existing = this.sessionNamespaces.get(ns.id); + if (existing && existing.connectionId !== ns.connectionId) { + throw new Error('Session namespace id belongs to another connection'); + } + const stored = { ...ns, createdAt: existing ? existing.createdAt : ns.createdAt }; + this.sessionNamespaces.set(ns.id, stored); + return { ...stored }; + } + + async deleteSessionNamespace(connectionId: string, id: string): Promise { + const ns = this.sessionNamespaces.get(id); + if (!ns || ns.connectionId !== connectionId) return false; + return this.sessionNamespaces.delete(id); + } + // Command Stats Sample Methods private commandStatsSamples: StoredCommandStatsSample[] = []; diff --git a/apps/api/src/storage/adapters/postgres.adapter.ts b/apps/api/src/storage/adapters/postgres.adapter.ts index 0bb67d145..654558fab 100644 --- a/apps/api/src/storage/adapters/postgres.adapter.ts +++ b/apps/api/src/storage/adapters/postgres.adapter.ts @@ -10,6 +10,7 @@ import type { import type { ReadinessDimension, ScalingReadinessSettings, + SessionNamespace, } from '@betterdb/shared'; import type { RawDatabaseHandle, RawDatabaseHandleProvider } from '../raw-database-handle'; import { @@ -1496,6 +1497,20 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { updated_at BIGINT NOT NULL ); + CREATE TABLE IF NOT EXISTS session_explorer_namespaces ( + id TEXT PRIMARY KEY, + connection_id TEXT NOT NULL, + name TEXT NOT NULL, + pattern TEXT NOT NULL, + preset TEXT NOT NULL, + custom_schema JSONB, + oversize_threshold_bytes BIGINT NOT NULL DEFAULT 1048576, + created_at BIGINT NOT NULL, + updated_at BIGINT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_session_explorer_namespaces_conn + ON session_explorer_namespaces(connection_id); + CREATE TABLE IF NOT EXISTS webhooks ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name VARCHAR(255) NOT NULL, @@ -3784,6 +3799,78 @@ export class PostgresAdapter implements StoragePort, RawDatabaseHandleProvider { return settings; } + private mapSessionNamespaceRow(row: any): SessionNamespace { + return { + id: row.id, + connectionId: row.connection_id, + name: row.name, + pattern: row.pattern, + preset: row.preset, + customSchema: row.custom_schema ?? null, + oversizeThresholdBytes: Number(row.oversize_threshold_bytes), + createdAt: Number(row.created_at), + updatedAt: Number(row.updated_at), + }; + } + + async listSessionNamespaces(connectionId: string): Promise { + if (!this.pool) throw new Error('Database not initialized'); + const result = await this.pool.query( + 'SELECT * FROM session_explorer_namespaces WHERE connection_id = $1 ORDER BY created_at ASC, id ASC', + [connectionId], + ); + return result.rows.map((row) => this.mapSessionNamespaceRow(row)); + } + + async getSessionNamespace(connectionId: string, id: string): Promise { + if (!this.pool) throw new Error('Database not initialized'); + const result = await this.pool.query( + 'SELECT * FROM session_explorer_namespaces WHERE connection_id = $1 AND id = $2', + [connectionId, id], + ); + return result.rows.length > 0 ? this.mapSessionNamespaceRow(result.rows[0]) : null; + } + + async saveSessionNamespace(ns: SessionNamespace): Promise { + if (!this.pool) throw new Error('Database not initialized'); + await this.pool.query( + `INSERT INTO session_explorer_namespaces + (id, connection_id, name, pattern, preset, custom_schema, oversize_threshold_bytes, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + ON CONFLICT (id) DO UPDATE SET + name = EXCLUDED.name, + pattern = EXCLUDED.pattern, + preset = EXCLUDED.preset, + custom_schema = EXCLUDED.custom_schema, + oversize_threshold_bytes = EXCLUDED.oversize_threshold_bytes, + updated_at = EXCLUDED.updated_at + WHERE session_explorer_namespaces.connection_id = EXCLUDED.connection_id`, + [ + ns.id, + ns.connectionId, + ns.name, + ns.pattern, + ns.preset, + ns.customSchema ? JSON.stringify(ns.customSchema) : null, + ns.oversizeThresholdBytes, + ns.createdAt, + ns.updatedAt, + ], + ); + const saved = await this.getSessionNamespace(ns.connectionId, ns.id); + if (!saved) throw new Error('Session namespace id belongs to another connection'); + return saved; + } + + async deleteSessionNamespace(connectionId: string, id: string): Promise { + if (!this.pool) throw new Error('Database not initialized'); + const result = await this.pool.query( + 'DELETE FROM session_explorer_namespaces WHERE connection_id = $1 AND id = $2', + [connectionId, id], + ); + return (result.rowCount ?? 0) > 0; + } + // Command Stats Sample Methods async saveCommandStatsSamples( samples: Omit[], diff --git a/apps/api/src/storage/adapters/sqlite.adapter.ts b/apps/api/src/storage/adapters/sqlite.adapter.ts index 0b6dfe2b6..c10417b39 100644 --- a/apps/api/src/storage/adapters/sqlite.adapter.ts +++ b/apps/api/src/storage/adapters/sqlite.adapter.ts @@ -13,6 +13,7 @@ import type { ReadinessDimension, ReadinessDimensionKey, ScalingReadinessSettings, + SessionNamespace, } from '@betterdb/shared'; import type { RawDatabaseHandle, RawDatabaseHandleProvider } from '../raw-database-handle'; import { @@ -1779,6 +1780,20 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { updated_at INTEGER NOT NULL ); + CREATE TABLE IF NOT EXISTS session_explorer_namespaces ( + id TEXT PRIMARY KEY, + connection_id TEXT NOT NULL, + name TEXT NOT NULL, + pattern TEXT NOT NULL, + preset TEXT NOT NULL, + custom_schema TEXT, + oversize_threshold_bytes INTEGER NOT NULL DEFAULT 1048576, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_session_explorer_namespaces_conn + ON session_explorer_namespaces(connection_id); + CREATE TABLE IF NOT EXISTS webhooks ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -3802,6 +3817,88 @@ export class SqliteAdapter implements StoragePort, RawDatabaseHandleProvider { return settings; } + private mapSessionNamespaceRow(row: { + id: string; + connection_id: string; + name: string; + pattern: string; + preset: string; + custom_schema: string | null; + oversize_threshold_bytes: number; + created_at: number; + updated_at: number; + }): SessionNamespace { + return { + id: row.id, + connectionId: row.connection_id, + name: row.name, + pattern: row.pattern, + preset: row.preset as SessionNamespace['preset'], + customSchema: row.custom_schema ? JSON.parse(row.custom_schema) : null, + oversizeThresholdBytes: row.oversize_threshold_bytes, + createdAt: row.created_at, + updatedAt: row.updated_at, + }; + } + + async listSessionNamespaces(connectionId: string): Promise { + if (!this.db) throw new Error('Database not initialized'); + const rows = this.db + .prepare( + 'SELECT * FROM session_explorer_namespaces WHERE connection_id = ? ORDER BY created_at ASC, id ASC', + ) + .all(connectionId) as Parameters[0][]; + return rows.map((row) => this.mapSessionNamespaceRow(row)); + } + + async getSessionNamespace(connectionId: string, id: string): Promise { + if (!this.db) throw new Error('Database not initialized'); + const row = this.db + .prepare('SELECT * FROM session_explorer_namespaces WHERE connection_id = ? AND id = ?') + .get(connectionId, id) as Parameters[0] | undefined; + return row ? this.mapSessionNamespaceRow(row) : null; + } + + async saveSessionNamespace(ns: SessionNamespace): Promise { + if (!this.db) throw new Error('Database not initialized'); + this.db + .prepare( + `INSERT INTO session_explorer_namespaces + (id, connection_id, name, pattern, preset, custom_schema, oversize_threshold_bytes, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + name = excluded.name, + pattern = excluded.pattern, + preset = excluded.preset, + custom_schema = excluded.custom_schema, + oversize_threshold_bytes = excluded.oversize_threshold_bytes, + updated_at = excluded.updated_at + WHERE session_explorer_namespaces.connection_id = excluded.connection_id`, + ) + .run( + ns.id, + ns.connectionId, + ns.name, + ns.pattern, + ns.preset, + ns.customSchema ? JSON.stringify(ns.customSchema) : null, + ns.oversizeThresholdBytes, + ns.createdAt, + ns.updatedAt, + ); + const saved = await this.getSessionNamespace(ns.connectionId, ns.id); + if (!saved) throw new Error('Session namespace id belongs to another connection'); + return saved; + } + + async deleteSessionNamespace(connectionId: string, id: string): Promise { + if (!this.db) throw new Error('Database not initialized'); + const result = this.db + .prepare('DELETE FROM session_explorer_namespaces WHERE connection_id = ? AND id = ?') + .run(connectionId, id); + return result.changes > 0; + } + // Command Stats Sample Methods async saveCommandStatsSamples( samples: Omit[], diff --git a/apps/api/test/session-namespace-storage-postgres.e2e-spec.ts b/apps/api/test/session-namespace-storage-postgres.e2e-spec.ts new file mode 100644 index 000000000..867669a44 --- /dev/null +++ b/apps/api/test/session-namespace-storage-postgres.e2e-spec.ts @@ -0,0 +1,83 @@ +import { execFileSync } from 'child_process'; +import { Pool } from 'pg'; +import { PostgresAdapter } from '../src/storage/adapters/postgres.adapter'; +import { runNamespaceStorageContract } from '../src/session-explorer/__tests__/namespace-storage-contract'; + +const DSN = + process.env.SESSION_NAMESPACE_POSTGRES_TEST_DSN ?? + process.env.SCALING_READINESS_POSTGRES_TEST_DSN ?? + 'postgres://betterdb:devpassword@localhost:5433/betterdb'; + +const SCHEMA = `session_namespace_contract_${process.pid}_${Date.now()}`; + +function postgresRequired(): boolean { + const flag = process.env.REQUIRE_POSTGRES_TESTS; + if (flag === 'true') { + return true; + } + if (flag === 'false') { + return false; + } + return process.env.CI === 'true'; +} + +function isReachable(connectionString: string): boolean { + const { hostname, port } = new URL(connectionString); + const probe = ` + const socket = require('net').createConnection({ + host: process.env.PROBE_HOST, + port: Number(process.env.PROBE_PORT), + }); + socket.setTimeout(1500); + socket.on('connect', () => { socket.destroy(); process.exit(0); }); + socket.on('timeout', () => { socket.destroy(); process.exit(1); }); + socket.on('error', () => { process.exit(1); }); + `; + try { + execFileSync(process.execPath, ['-e', probe], { + stdio: 'ignore', + timeout: 3000, + env: { ...process.env, PROBE_HOST: hostname, PROBE_PORT: port || '5432' }, + }); + return true; + } catch { + return false; + } +} + +const reachable = isReachable(DSN); +const describePostgres = reachable ? describe : describe.skip; + +if (reachable === false && postgresRequired() === true) { + describe('session namespace storage — postgres (live Postgres required)', () => { + it('has a reachable PostgreSQL', () => { + throw new Error( + `PostgreSQL is required here (CI or REQUIRE_POSTGRES_TESTS=true) but ${new URL(DSN).host} is unreachable`, + ); + }); + }); +} + +describePostgres('session namespace storage — postgres', () => { + let adminPool: Pool; + let adapter: PostgresAdapter; + + beforeAll(async () => { + adminPool = new Pool({ connectionString: DSN }); + await adminPool.query(`CREATE SCHEMA ${SCHEMA}`); + adapter = new PostgresAdapter({ connectionString: DSN, schema: SCHEMA }); + await adapter.initialize(); + }); + + beforeEach(async () => { + await adminPool.query(`DELETE FROM ${SCHEMA}.session_explorer_namespaces`); + }); + + afterAll(async () => { + await adapter.close(); + await adminPool.query(`DROP SCHEMA IF EXISTS ${SCHEMA} CASCADE`); + await adminPool.end(); + }); + + runNamespaceStorageContract(() => adapter); +}); diff --git a/packages/shared/src/types/session-explorer.types.ts b/packages/shared/src/types/session-explorer.types.ts index e44a38f8c..29671a317 100644 --- a/packages/shared/src/types/session-explorer.types.ts +++ b/packages/shared/src/types/session-explorer.types.ts @@ -61,6 +61,7 @@ export interface SessionSummary { sessionId: string; keyCount: number; memoryBytes: number; + unmeasuredKeys: number; idleSeconds: number | null; ttlMs: number | null; } @@ -100,4 +101,31 @@ export interface SessionDetail { estimatedTokens: number; ttlMs: number | null; memoryBytes: number; + complete: boolean; + unparsedKeys: number; + truncated: boolean; + failedNodes: number; + missingKeys: number; +} + +export interface SessionOversizeResponse { + thresholdBytes: number; + sessionIds: string[]; + unmeasuredSessionIds: string[]; + truncated: boolean; + partial: boolean; + failedNodes: number; +} + +export interface SessionPreview { + sampleKey: string | null; + view: KeyView | null; +} + +export interface SessionExport { + exportedAt: string; + namespace: Pick; + sessionId: string; + keys: CapturedKey[]; + parsed: SessionDetail; } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index c558df61a..3ab608eed 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -183,6 +183,9 @@ importers: class-validator: specifier: ^0.15.1 version: 0.15.1 + cluster-key-slot: + specifier: ^1.1.2 + version: 1.1.2 cron: specifier: ^4.4.0 version: 4.4.0