From b1d8a56c53effa310710adefb1b8ed518a708566 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 24 May 2026 17:26:41 +0200 Subject: [PATCH] adaptation related to meta-core api changes --- .../src/events/FileEventConsumer.ts | 121 ++++--- .../src/events/SSEEventClient.ts | 295 ++++++++++++++++++ packages/meta-sort-core/src/index.ts | 18 +- packages/meta-sort-core/src/kv/KVManager.ts | 24 +- .../src/kv/MetaCoreApiWriter.ts | 232 ++++++++++++++ packages/meta-sort-core/src/kv/RedisClient.ts | 94 +++++- 6 files changed, 703 insertions(+), 81 deletions(-) create mode 100644 packages/meta-sort-core/src/events/SSEEventClient.ts create mode 100644 packages/meta-sort-core/src/kv/MetaCoreApiWriter.ts diff --git a/packages/meta-sort-core/src/events/FileEventConsumer.ts b/packages/meta-sort-core/src/events/FileEventConsumer.ts index 26812a5..bf20e28 100644 --- a/packages/meta-sort-core/src/events/FileEventConsumer.ts +++ b/packages/meta-sort-core/src/events/FileEventConsumer.ts @@ -1,14 +1,20 @@ /** * FileEventConsumer * - * Consumes file events from meta-core via Redis Stream. + * Consumes file events from meta-core via SSE (Server-Sent Events). * When events are received, they are forwarded to the StreamingPipeline. * - * This enables meta-sort to run without direct filesystem access - it receives + * This enables meta-sort to run without direct filesystem access — it receives * file discovery events from meta-core and processes files via WebDAV. + * + * Previously consumed `file:events` via Redis Streams XREADGROUP. The + * api-mediated-access migration (PR C) routes this through meta-core's HTTP + * /api/events/files endpoint instead, so consumers no longer need direct + * Redis access. Cursor durability is now the consumer's responsibility — + * see SSEEventClient. */ -import type { RedisKVClient, StreamMessage } from '../kv/RedisClient.js'; +import { SSEEventClient, type SSEEvent } from './SSEEventClient.js'; // Pipeline interface (minimal, matching StreamingPipeline) interface Pipeline { @@ -18,13 +24,9 @@ interface Pipeline { reset(): Promise; } -// Redis Stream configuration -const EVENTS_STREAM = 'file:events'; -const CONSUMER_GROUP = 'meta-sort-processor'; - export interface FileEventConsumerOptions { - /** Redis KV client */ - kvClient: RedisKVClient; + /** meta-core API base URL (e.g. http://metacore-app:9000). */ + metaCoreApiUrl: string; /** Pipeline to receive events */ pipeline: Pipeline; @@ -32,115 +34,104 @@ export interface FileEventConsumerOptions { /** Base path for files (e.g., /files). Paths from meta-core are relative and need this prefix. */ filesPath?: string; - /** Block timeout in milliseconds for stream reads */ - blockMs?: number; + /** Where to persist the SSE cursor. Defaults to /meta-core/cursors/meta-sort-files.cursor. */ + cursorPath?: string; } export class FileEventConsumer { - private kvClient: RedisKVClient; private pipeline: Pipeline; private filesPath: string; - private blockMs: number; + private sse: SSEEventClient; private isRunning = false; constructor(options: FileEventConsumerOptions) { - this.kvClient = options.kvClient; this.pipeline = options.pipeline; this.filesPath = options.filesPath ?? '/files'; - this.blockMs = options.blockMs ?? 5000; + + const apiUrl = options.metaCoreApiUrl.replace(/\/+$/, ''); + const cursorPath = options.cursorPath ?? '/meta-core/cursors/meta-sort-files.cursor'; + + this.sse = new SSEEventClient({ + url: `${apiUrl}/api/events/files`, + cursorPath, + logTag: '[FileEventConsumer]', + onEvent: (e) => this.handleSSEEvent(e), + // gap on file:events means we missed retention. Easiest correct + // response: ask meta-core for a watcher rescan so the StreamingPipeline + // can rebuild from a known-good baseline. + onGap: () => { + console.warn('[FileEventConsumer] Stream cursor trimmed — pipeline will rebuild from next reset event'); + }, + }); } /** - * Start consuming events from Redis Stream + * Start consuming events from the meta-core SSE endpoint. */ async start(): Promise { if (this.isRunning) { console.log('[FileEventConsumer] Already running'); return; } - this.isRunning = true; - console.log(`[FileEventConsumer] Starting stream consumer for ${EVENTS_STREAM}...`); - - // Initialize consumer group - await this.kvClient.initStreamConsumer(EVENTS_STREAM, CONSUMER_GROUP); - - // Process any pending entries from crashed consumers - await this.kvClient.processPendingEntries( - EVENTS_STREAM, - CONSUMER_GROUP, - 30000, // 30 second idle threshold - async (message) => this.handleMessage(message) - ); - - // Start consumer loop in background (don't await - runs indefinitely) - this.kvClient.startStreamConsumer( - EVENTS_STREAM, - CONSUMER_GROUP, - async (message) => this.handleMessage(message), - this.blockMs - ).catch(error => { - console.error('[FileEventConsumer] Stream consumer error:', error); + console.log('[FileEventConsumer] Starting SSE stream consumer for /api/events/files...'); + // Fire-and-forget. SSEEventClient.start() returns only when stopped. + this.sse.start().catch(error => { + console.error('[FileEventConsumer] Stream consumer terminated:', error); }); - - console.log('[FileEventConsumer] Stream consumer started'); + console.log('[FileEventConsumer] SSE stream consumer started'); } /** * Stop consuming events */ - stop(): void { + async stop(): Promise { if (!this.isRunning) { return; } - this.isRunning = false; - this.kvClient.stopStreamConsumer(); + await this.sse.stop(); console.log('[FileEventConsumer] Stopped'); } - /** - * Handle a stream message - */ - private async handleMessage(message: StreamMessage): Promise { + private async handleSSEEvent(e: SSEEvent): Promise { // Convert relative path to absolute path const toAbsolutePath = (relativePath: string): string => { - // If path is already absolute, return as-is - if (relativePath.startsWith('/')) { - return relativePath; - } - // Otherwise, prepend filesPath + if (!relativePath) return relativePath; + if (relativePath.startsWith('/')) return relativePath; return `${this.filesPath}/${relativePath}`; }; - switch (message.type) { + const path: string = e.data?.path ?? ''; + const midhash256: string | undefined = e.data?.midhash256 ?? undefined; + const oldPath: string | undefined = e.data?.oldPath ?? undefined; + const watcherId: string | undefined = e.data?.watcherId ?? undefined; + + switch (e.event) { case 'reset': - console.log(`[FileEventConsumer] Received reset event from watcher: ${message.watcherId || 'all'}`); + console.log(`[FileEventConsumer] Received reset event from watcher: ${watcherId || 'all'}`); await this.pipeline.reset(); break; case 'add': - this.pipeline.handleFileAdded(toAbsolutePath(message.path), message.midhash256); + if (path) this.pipeline.handleFileAdded(toAbsolutePath(path), midhash256); break; case 'change': - this.pipeline.handleFileChanged(toAbsolutePath(message.path), message.midhash256); + if (path) this.pipeline.handleFileChanged(toAbsolutePath(path), midhash256); break; case 'delete': - this.pipeline.handleFileDeleted(toAbsolutePath(message.path)); + if (path) this.pipeline.handleFileDeleted(toAbsolutePath(path)); break; case 'rename': - // Handle rename as delete + add - if (message.oldPath) { - this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath)); - } - this.pipeline.handleFileAdded(toAbsolutePath(message.path), message.midhash256); + if (oldPath) this.pipeline.handleFileDeleted(toAbsolutePath(oldPath)); + if (path) this.pipeline.handleFileAdded(toAbsolutePath(path), midhash256); break; default: - console.warn(`[FileEventConsumer] Unknown event type: ${message.type}`); + console.warn(`[FileEventConsumer] Unknown event type: ${e.event}`); } } } @@ -149,12 +140,12 @@ export class FileEventConsumer { * Create a FileEventConsumer with default configuration */ export function createFileEventConsumer( - kvClient: RedisKVClient, + metaCoreApiUrl: string, pipeline: Pipeline, options?: Partial ): FileEventConsumer { return new FileEventConsumer({ - kvClient, + metaCoreApiUrl, pipeline, ...options }); diff --git a/packages/meta-sort-core/src/events/SSEEventClient.ts b/packages/meta-sort-core/src/events/SSEEventClient.ts new file mode 100644 index 0000000..2f9de22 --- /dev/null +++ b/packages/meta-sort-core/src/events/SSEEventClient.ts @@ -0,0 +1,295 @@ +/** + * SSEEventClient — HTTP Server-Sent Events client for meta-core event streams. + * + * Replaces the Redis Streams XREADGROUP loop. Each consumer service holds + * one long-lived HTTP connection to meta-core's /api/events/{files,meta} + * endpoint and receives JSON events as the underlying Redis Stream is + * appended to. + * + * Cursor semantics (see meta-core docs/api-mediated-access.md): + * - The SSE wire id is the opaque Redis Stream entry ID (`-`). + * We treat it as a black-box string. + * - On reconnect we send Last-Event-ID so meta-core resumes from the + * entry AFTER our cursor. + * - We persist the cursor to disk so the loop survives a process + * restart. The cursor is written every flushIntervalMs and on graceful + * stop. + * - On a `gap` event the server is telling us our cursor was trimmed + * out of retention. Caller-supplied onGap handles it (typically a + * re-bootstrap from a full HTTP fetch). If the caller doesn't handle + * gaps, we just continue from the new cursor. + */ +import { promises as fs } from 'fs'; +import { dirname } from 'path'; + +/** One SSE event parsed off the wire. */ +export interface SSEEvent { + /** Opaque cursor (Redis Stream entry ID under the hood). */ + id: string; + /** Event type — `add` / `change` / `delete` / `rename` / `reset` for + * file:events; `set` / `del` / `expire` for meta:events; `gap` for + * trim notifications; arbitrary otherwise. */ + event: string; + /** Parsed JSON payload from the `data:` line. */ + data: any; +} + +export interface SSEClientOptions { + /** Full URL to subscribe to (e.g. http://metacore-app:9000/api/events/files). */ + url: string; + + /** Path where the cursor is persisted between runs. + * Set to null to keep the cursor in-memory only (at-most-once on crash). */ + cursorPath: string | null; + + /** Handler for every non-gap event. */ + onEvent: (e: SSEEvent) => Promise | void; + + /** Optional handler for the synthetic `gap` event. If omitted we just + * log and continue from the resumeFrom cursor inside the payload. */ + onGap?: (payload: { requested: string; resumeFrom: string; reason: string }) => Promise | void; + + /** Logger prefix used for console.log lines. */ + logTag?: string; + + /** Initial reconnect backoff in ms (doubles per failure, capped). Default 500. */ + reconnectBaseMs?: number; + + /** Max reconnect backoff in ms. Default 30_000. */ + reconnectMaxMs?: number; + + /** How often to flush the in-memory cursor to disk. Default 2000. */ + flushIntervalMs?: number; +} + +export class SSEEventClient { + private readonly opts: Required> & Pick; + private cursor: string = ''; + private cursorDirty = false; + private abort: AbortController | null = null; + private flushTimer: NodeJS.Timeout | null = null; + private stopped = false; + private backoff = 0; + + constructor(options: SSEClientOptions) { + this.opts = { + url: options.url, + cursorPath: options.cursorPath, + onEvent: options.onEvent, + onGap: options.onGap, + logTag: options.logTag ?? '[SSE]', + reconnectBaseMs: options.reconnectBaseMs ?? 500, + reconnectMaxMs: options.reconnectMaxMs ?? 30_000, + flushIntervalMs: options.flushIntervalMs ?? 2000, + }; + this.backoff = this.opts.reconnectBaseMs; + } + + /** + * Read the persisted cursor (if any) and begin streaming. The returned + * promise resolves on stop()/permanent error; callers should not await + * it directly — fire-and-forget and call stop() to shut down. + */ + async start(): Promise { + await this.loadCursor(); + this.startFlushTimer(); + await this.runLoop(); + } + + /** Graceful shutdown — closes the open connection and flushes the cursor. */ + async stop(): Promise { + this.stopped = true; + if (this.abort) { + this.abort.abort(); + this.abort = null; + } + if (this.flushTimer) { + clearInterval(this.flushTimer); + this.flushTimer = null; + } + await this.flushCursor(); + } + + /** Currently-known cursor — useful for tests. */ + getCursor(): string { + return this.cursor; + } + + private async loadCursor(): Promise { + if (!this.opts.cursorPath) return; + try { + const data = await fs.readFile(this.opts.cursorPath, 'utf-8'); + const trimmed = data.trim(); + if (trimmed) { + this.cursor = trimmed; + console.log(`${this.opts.logTag} Resuming from cursor ${this.cursor}`); + } + } catch (err: any) { + if (err.code !== 'ENOENT') { + console.warn(`${this.opts.logTag} Could not read cursor file ${this.opts.cursorPath}: ${err.message}`); + } + // ENOENT is expected first time; start at "$" (server-default). + } + } + + private startFlushTimer(): void { + this.flushTimer = setInterval(() => { + if (this.cursorDirty) { + this.flushCursor().catch(err => + console.warn(`${this.opts.logTag} Cursor flush failed: ${err.message}`) + ); + } + }, this.opts.flushIntervalMs); + } + + private async flushCursor(): Promise { + if (!this.opts.cursorPath || !this.cursorDirty) return; + this.cursorDirty = false; + try { + await fs.mkdir(dirname(this.opts.cursorPath), { recursive: true }); + await fs.writeFile(this.opts.cursorPath, this.cursor, 'utf-8'); + } catch (err: any) { + // Re-mark dirty so the next tick retries. + this.cursorDirty = true; + throw err; + } + } + + private async runLoop(): Promise { + while (!this.stopped) { + try { + await this.connect(); + // connect() only returns on graceful end-of-stream — reset + // backoff so a long-running connection that finally + // disconnects reconnects quickly. + this.backoff = this.opts.reconnectBaseMs; + } catch (err: any) { + if (this.stopped) return; + console.warn(`${this.opts.logTag} Stream error, reconnecting in ${this.backoff}ms: ${err?.message ?? err}`); + await this.sleep(this.backoff); + this.backoff = Math.min(this.backoff * 2, this.opts.reconnectMaxMs); + } + } + } + + private async connect(): Promise { + this.abort = new AbortController(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache', + }; + if (this.cursor) { + headers['Last-Event-ID'] = this.cursor; + } + + const response = await fetch(this.opts.url, { + method: 'GET', + headers, + signal: this.abort.signal, + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status} ${response.statusText}`); + } + if (!response.body) { + throw new Error('Response has no body'); + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder('utf-8'); + let buffer = ''; + + try { + while (!this.stopped) { + const { value, done } = await reader.read(); + if (done) { + return; + } + buffer += decoder.decode(value, { stream: true }); + // Each SSE event ends with a blank line ("\n\n"). Split on + // that boundary and process each chunk. + let idx: number; + while ((idx = buffer.indexOf('\n\n')) !== -1) { + const rawChunk = buffer.slice(0, idx); + buffer = buffer.slice(idx + 2); + await this.handleChunk(rawChunk); + } + } + } finally { + try { reader.releaseLock(); } catch { /* nothing useful */ } + } + } + + private async handleChunk(chunk: string): Promise { + let id = ''; + let event = ''; + const dataLines: string[] = []; + for (const rawLine of chunk.split('\n')) { + if (!rawLine || rawLine.startsWith(':')) { + continue; // comment / heartbeat — ignore + } + const colon = rawLine.indexOf(':'); + if (colon === -1) continue; + const field = rawLine.slice(0, colon); + // SSE spec: ": " (single space after the colon) is stripped; if no space, keep as-is. + let value = rawLine.slice(colon + 1); + if (value.startsWith(' ')) value = value.slice(1); + + switch (field) { + case 'id': + id = value; + break; + case 'event': + event = value; + break; + case 'data': + dataLines.push(value); + break; + // ignore other fields (retry, etc.) + } + } + + if (dataLines.length === 0) { + return; // bare heartbeat / no payload + } + + let data: any; + try { + data = JSON.parse(dataLines.join('\n')); + } catch (err: any) { + console.warn(`${this.opts.logTag} Skipping unparseable event ${id || '(no id)'}: ${err.message}`); + return; + } + + const ev: SSEEvent = { id, event: event || 'message', data }; + + if (ev.event === 'gap') { + if (this.opts.onGap) { + await this.opts.onGap(ev.data); + } else { + console.warn(`${this.opts.logTag} Gap event (cursor ${ev.data?.requested} trimmed): resumed from ${ev.data?.resumeFrom}`); + } + if (typeof ev.data?.resumeFrom === 'string' && ev.data.resumeFrom) { + this.cursor = ev.data.resumeFrom; + this.cursorDirty = true; + } + return; + } + + try { + await this.opts.onEvent(ev); + } catch (err: any) { + // Handler error doesn't tear down the connection — log and move on. + console.error(`${this.opts.logTag} onEvent handler failed for ${ev.id}: ${err?.stack ?? err}`); + } + + if (ev.id) { + this.cursor = ev.id; + this.cursorDirty = true; + } + } + + private sleep(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)); + } +} diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index a000c96..824f0cf 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -104,7 +104,7 @@ process.on('SIGINT', async () => { // Stop File Event Consumer if running if (fileEventConsumer) { console.log('[Shutdown] Stopping File Event Consumer...'); - fileEventConsumer.stop(); + await fileEventConsumer.stop(); } // Stop Container Manager if running @@ -256,21 +256,21 @@ process.on('SIGINT', async () => { console.log(`[Cache] midhash256 cache: ${hits} hits, ${misses} misses (${hitRate}% hit rate)`); } - // Connect to meta-core for file events via Redis Stream - // KV client must be available - if (!kvClient) { - console.error('[Startup] ERROR: KV client not available. Cannot consume file events.'); - console.error('[Startup] Make sure meta-core is running and Redis is accessible.'); + // Connect to meta-core for file events via SSE + const leaderInfoForSSE = kvManager?.getLeaderInfo(); + const metaCoreApiUrl = leaderInfoForSSE?.apiUrl; + if (!metaCoreApiUrl) { + console.error('[Startup] ERROR: meta-core apiUrl not available from leader info. Cannot consume file events.'); process.exit(1); } - console.log('[Startup] Starting File Event Consumer (Redis Stream)...'); + console.log(`[Startup] Starting File Event Consumer (SSE → ${metaCoreApiUrl}/api/events/files)...`); fileEventConsumer = new FileEventConsumer({ - kvClient: kvClient as RedisKVClient, + metaCoreApiUrl, pipeline: pipeline, filesPath: config.FILES_PATH, }); await fileEventConsumer.start(); - console.log('[Startup] FileEventConsumer started, consuming from file:events stream'); + console.log('[Startup] FileEventConsumer started, consuming via SSE'); } catch (error) { console.error('[Startup] Error during startup:', error); diff --git a/packages/meta-sort-core/src/kv/KVManager.ts b/packages/meta-sort-core/src/kv/KVManager.ts index cd7b60b..5d2d7d3 100644 --- a/packages/meta-sort-core/src/kv/KVManager.ts +++ b/packages/meta-sort-core/src/kv/KVManager.ts @@ -73,13 +73,17 @@ export class KVManager { } /** - * Create Redis client from URL + * Create Redis client from URL. When apiUrl is supplied, the resulting + * client routes writes through meta-core's HTTP API (PR B of the + * api-mediated-access lockdown) while keeping reads + stream consumption + * on the direct Redis connection. */ - private async createRedisClient(redisUrl: string): Promise { + private async createRedisClient(redisUrl: string, apiUrl?: string): Promise { const client = new RedisKVClient({ url: redisUrl, timeout: 30000, - prefix: '' // No prefix for shared access + prefix: '', // No prefix for shared access + metaCoreApiUrl: apiUrl, }); await client.connect(); @@ -121,8 +125,16 @@ export class KVManager { // Wait for leader and connect try { const leaderInfo = await this.leaderClient.waitForLeader(30000); - console.log(`[KVManager] Connecting to Redis at ${leaderInfo.redisUrl}...`); - this.kvClient = await this.createRedisClient(leaderInfo.redisUrl); + if (leaderInfo.redisUrl) { + // api-mediated-access PR D removes redisUrl from leader info. + // Until that lands we still log it but services boot fine + // when it's absent — all writes/reads route through HTTP. + console.warn('[KVManager] Leader info still publishes redisUrl; direct Redis access is being deprecated, prefer meta-core HTTP'); + console.log(`[KVManager] Connecting to Redis at ${leaderInfo.redisUrl}...`); + } else { + console.log('[KVManager] redisUrl absent in leader info — running in HTTP-only mode'); + } + this.kvClient = await this.createRedisClient(leaderInfo.redisUrl, leaderInfo.apiUrl); // Configure WebDAV client with leader's WebDAV URL (prefer internal for container-to-container) const webdavUrl = leaderInfo.webdavUrlInternal ?? leaderInfo.webdavUrl; @@ -169,7 +181,7 @@ export class KVManager { try { const leaderInfo = await this.leaderClient.waitForLeader(30000); console.log(`[KVManager] Reconnecting to Redis at ${leaderInfo.redisUrl}...`); - this.kvClient = await this.createRedisClient(leaderInfo.redisUrl); + this.kvClient = await this.createRedisClient(leaderInfo.redisUrl, leaderInfo.apiUrl); // Reconfigure WebDAV client with new leader's WebDAV URL (prefer internal for container-to-container) const webdavUrl = leaderInfo.webdavUrlInternal ?? leaderInfo.webdavUrl; diff --git a/packages/meta-sort-core/src/kv/MetaCoreApiWriter.ts b/packages/meta-sort-core/src/kv/MetaCoreApiWriter.ts new file mode 100644 index 0000000..c2057f3 --- /dev/null +++ b/packages/meta-sort-core/src/kv/MetaCoreApiWriter.ts @@ -0,0 +1,232 @@ +/** + * MetaCoreApiWriter — HTTP writer for metadata, replacing direct Redis writes. + * + * meta-core owns the Redis schema (UUID-rooted, reverse-indexed by CID). Any + * service writing directly to Redis risks drifting the schema; the lockdown + * plan (docs/api-mediated-access.md in meta-core) routes every write through + * meta-core's HTTP /meta/{hash} family. + * + * This writer covers the three operations meta-sort actually performs in + * production: + * - setMetadataFlat → PATCH /meta/{hash} (merge semantics) + * - setMetadataProperty → PUT /meta/{hash}/{key} + * - deleteMetadataFlat → DELETE /meta/{hash} + * + * Reads and stream consumption stay on Redis in PR B; PR C migrates those. + */ +import { + flattenMetadata, + buildFilePrefix, +} from './MetadataUtils.js'; + +export interface MetaCoreApiWriterConfig { + /** Base URL for meta-core's HTTP API (e.g. http://metacore-app:9000) */ + apiUrl: string; + + /** Optional per-request timeout in ms (default: 30000) */ + timeoutMs?: number; +} + +export class MetaCoreApiWriter { + private apiUrl: string; + private timeoutMs: number; + + constructor(config: MetaCoreApiWriterConfig) { + // Trim trailing slash so we can join paths predictably. + this.apiUrl = config.apiUrl.replace(/\/+$/, ''); + this.timeoutMs = config.timeoutMs ?? 30000; + } + + /** + * PATCH /meta/{hash}. + * + * meta-core's PATCH calls MergeMetadataFlat — merge semantics, matches + * what setMetadataFlat used to do via Redis pipeline of SETs. + */ + async setMetadataFlat( + hashId: string, + metadata: any, + excludeFields: string[] = [] + ): Promise { + // Reuse the same flattening logic the Redis path used so the wire + // payload matches what handler.MergeMetadataFlat expects (flat map + // of "a/b/c" → string value). + const prefix = buildFilePrefix(hashId); + const pairs = flattenMetadata(metadata, prefix, excludeFields); + if (pairs.length === 0) return; + + const flat: Record = {}; + for (const pair of pairs) { + // pair.key is "file:/"; strip the prefix+slash. + const field = pair.key.slice(prefix.length + 1); + if (field) { + flat[field] = pair.value; + } + } + + await this.fetchJson('PATCH', `/meta/${encodeURIComponent(hashId)}`, flat); + } + + /** + * PUT /meta/{hash}/{key}. + * + * meta-core accepts either JSON {"value":"..."} or plain text; we use + * the JSON form to avoid ambiguity with values that happen to look like + * JSON themselves. + */ + async setMetadataProperty(hashId: string, property: string, value: string): Promise { + await this.fetchJson( + 'PUT', + `/meta/${encodeURIComponent(hashId)}/${this.encodePropertyPath(property)}`, + { value } + ); + } + + /** + * DELETE /meta/{hash}. + * + * Removes every property and unregisters the hashId from file:__index__. + */ + async deleteMetadataFlat(hashId: string): Promise { + const body = await this.fetchJson( + 'DELETE', + `/meta/${encodeURIComponent(hashId)}`, + null + ); + // meta-core's response is {success, hashId, deleted: N}. + if (body && typeof body === 'object' && typeof (body as any).deleted === 'number') { + return (body as any).deleted; + } + return 0; + } + + /** + * GET /meta/{hash}. Returns the flat metadata map (or null if absent). + */ + async getMetadataFlat(hashId: string): Promise { + const path = `/meta/${encodeURIComponent(hashId)}`; + const body = await this.fetchJsonAllowing404('GET', path); + if (body === null) return null; + // meta-core wraps the flat map under `metadata`. + const flat = (body as any)?.metadata; + if (!flat || typeof flat !== 'object') return null; + return flat; + } + + /** + * GET /meta/{hash}/{propertyPath}. Returns the raw string value (or null + * if absent). meta-core sends back text/plain. + */ + async getMetadata(hashId: string, propertyPath: string): Promise { + const path = `/meta/${encodeURIComponent(hashId)}/${this.encodePropertyPath(propertyPath)}`; + const url = `${this.apiUrl}${path}`; + const response = await fetch(url, { + method: 'GET', + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'text/plain' }, + }); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`[MetaCoreApiWriter] GET ${path} → ${response.status} ${response.statusText}`); + } + const text = await response.text(); + if (!text) return null; + // Mirror the legacy RedisClient: try JSON-parse first, fall back to raw string. + try { + return JSON.parse(text); + } catch { + return text; + } + } + + /** + * GET /meta. Returns the list of all hashIds known to meta-core. + */ + async getAllHashIds(): Promise { + const body = await this.fetchJson('GET', '/meta', null); + const ids = (body as any)?.hashIds; + return Array.isArray(ids) ? ids : []; + } + + /** + * GET /health on meta-core. Used as a write-path health probe. + */ + async health(): Promise { + try { + const response = await fetch(`${this.apiUrl}/health`, { + method: 'GET', + signal: AbortSignal.timeout(Math.min(this.timeoutMs, 5000)), + headers: { Accept: 'application/json' }, + }); + return response.ok; + } catch { + return false; + } + } + + private async fetchJsonAllowing404(method: string, path: string): Promise { + const url = `${this.apiUrl}${path}`; + const response = await fetch(url, { + method, + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'application/json' }, + }); + if (response.status === 404) return null; + if (!response.ok) { + throw new Error(`[MetaCoreApiWriter] ${method} ${path} → ${response.status} ${response.statusText}`); + } + const text = await response.text(); + if (!text) return null; + try { return JSON.parse(text); } catch { return text; } + } + + /** + * Encode a slash-containing property path safely for the URL while still + * letting meta-core's `{key:.*}` route match the multi-segment key. + */ + private encodePropertyPath(property: string): string { + return property + .split('/') + .map(segment => encodeURIComponent(segment)) + .join('/'); + } + + private async fetchJson( + method: string, + path: string, + body: unknown + ): Promise { + const url = `${this.apiUrl}${path}`; + const init: RequestInit = { + method, + signal: AbortSignal.timeout(this.timeoutMs), + headers: { Accept: 'application/json' }, + }; + if (body !== null && body !== undefined) { + init.headers = { ...init.headers, 'Content-Type': 'application/json' }; + init.body = JSON.stringify(body); + } + + const response = await fetch(url, init); + if (!response.ok) { + // Surface meta-core's error envelope when it's structured. + let detail = response.statusText; + try { + const errBody = await response.text(); + if (errBody) detail = `${detail}: ${errBody}`; + } catch { + /* ignore body-read errors; statusText still useful */ + } + throw new Error(`[MetaCoreApiWriter] ${method} ${path} → ${response.status} ${detail}`); + } + + // 204 / empty body — return null so callers don't choke on JSON parse. + const text = await response.text(); + if (!text) return null; + try { + return JSON.parse(text); + } catch { + return text; + } + } +} diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index 85c253c..ec55555 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -21,6 +21,7 @@ import { buildFilePrefix, buildPropertyKey, } from './MetadataUtils.js'; +import { MetaCoreApiWriter } from './MetaCoreApiWriter.js'; // ESM/CJS interop for ioredis const Redis = (IORedis as any).default ?? IORedis; @@ -50,6 +51,18 @@ interface RedisClientConfig { url: string; timeout?: number; prefix?: string; + + /** + * meta-core HTTP API base URL. When set, write methods + * (setMetadataFlat / setMetadataProperty / deleteMetadataFlat) route + * through the /meta/{hash} HTTP family instead of writing to Redis + * directly. Reads and stream consumption still use the Redis connection. + * + * Wiring this is the meta-sort half of the api-mediated-access lockdown + * (see meta-core docs/api-mediated-access.md, PR B). Leaving it unset + * preserves the legacy direct-write path for tests / one-off scripts. + */ + metaCoreApiUrl?: string; } export class RedisKVClient implements IKVClient { @@ -58,6 +71,7 @@ export class RedisKVClient implements IKVClient { private timeout: number; private prefix: string; private isConnected: boolean = false; + private writer: MetaCoreApiWriter | null = null; // Stream consumer state private streamConsumerRunning = false; @@ -68,9 +82,32 @@ export class RedisKVClient implements IKVClient { this.timeout = config.timeout || 30000; this.prefix = config.prefix || ''; + if (config.metaCoreApiUrl) { + this.writer = new MetaCoreApiWriter({ + apiUrl: config.metaCoreApiUrl, + timeoutMs: this.timeout, + }); + console.log(`[Redis] Writes routed through meta-core API at ${config.metaCoreApiUrl}`); + } + // Unique consumer name for this instance this.consumerName = `${os.hostname()}-${process.pid}`; + // HTTP-only mode: no Redis URL AND a writer is configured. Create a + // disconnected stub Redis so accidental fallthrough into a Redis + // method throws immediately instead of silently connecting to + // localhost. + if (!config.url && this.writer) { + // ioredis requires a config object; lazyConnect prevents the + // automatic connect. Any caller that tries to use the Redis + // path (e.g. unmigrated stream code) will fail loudly rather + // than silently misbehaving. + this.redis = new Redis({ host: '127.0.0.1', port: 0, lazyConnect: true, enableOfflineQueue: false }); + this.isConnected = false; + console.log('[Redis] HTTP-only mode — no direct Redis connection'); + return; + } + // Parse Redis URL and create client this.redis = new Redis(config.url, { commandTimeout: this.timeout, @@ -102,10 +139,22 @@ export class RedisKVClient implements IKVClient { } /** - * Connect to Redis (call before using other methods) + * Connect to Redis (call before using other methods). + * + * In HTTP-only mode (no Redis URL + writer configured), connect is a + * no-op — there's no underlying socket to open. */ async connect(): Promise { if (this.isConnected) return; + if (this.writer && !this.redis.options.host) { + // HTTP-only mode stub — nothing to connect. + return; + } + // Skip explicit connect when host is 127.0.0.1 with port 0 (HTTP-only stub). + const opts = this.redis.options as { host?: string; port?: number }; + if (opts && opts.port === 0) { + return; + } await this.redis.connect(); } @@ -163,6 +212,11 @@ export class RedisKVClient implements IKVClient { } async health(): Promise { + if (this.writer) { + // Probe meta-core /health rather than Redis PING; that's the + // surface the application actually depends on after the lockdown. + return this.writer.health(); + } try { const result = await this.redis.ping(); return result === 'PONG'; @@ -271,6 +325,12 @@ export class RedisKVClient implements IKVClient { metadata: any, excludeFields: string[] = [] ): Promise { + if (this.writer) { + // PATCH /meta/{hash} — meta-core handles the flat-key fanout and + // the file:__index__ membership atomically server-side. + return this.writer.setMetadataFlat(hashId, metadata, excludeFields); + } + const prefix = buildFilePrefix(hashId); const pairs = flattenMetadata(metadata, prefix, excludeFields); @@ -296,6 +356,20 @@ export class RedisKVClient implements IKVClient { } async getMetadataFlat(hashId: string): Promise { + if (this.writer) { + // GET /meta/{hash} via HTTP — returns the flat map directly, + // so we don't need a Redis SCAN/MGET round trip. Reconstruct + // the nested structure consumers expect. + const flat = await this.writer.getMetadataFlat(hashId); + if (!flat) return null; + const prefix = buildFilePrefix(hashId); + const pairs: KeyValuePair[] = Object.entries(flat).map(([field, value]) => ({ + key: `${prefix}/${field}`, + value: value as string, + })); + return reconstructMetadata(pairs, prefix); + } + const keyPrefix = this.buildKey(`file:${hashId}/`); // Scan for all keys with this prefix @@ -329,6 +403,13 @@ export class RedisKVClient implements IKVClient { } async getMetadata(hashId: string, propertyPath: string): Promise { + if (this.writer) { + // GET /meta/{hash}/{prop} via HTTP; meta-core returns text/plain + // and the writer mirrors the JSON-or-raw fallback the Redis path + // uses below. + return this.writer.getMetadata(hashId, propertyPath); + } + const key = this.buildKey(`file:${hashId}/${propertyPath}`); const value = await this.redis.get(key); @@ -346,6 +427,10 @@ export class RedisKVClient implements IKVClient { * Writes to flat key: file:{hashId}/{property} */ async setMetadataProperty(hashId: string, property: string, value: string): Promise { + if (this.writer) { + return this.writer.setMetadataProperty(hashId, property, value); + } + const key = this.buildKey(`file:${hashId}/${property}`); await this.redis.set(key, value); @@ -354,6 +439,10 @@ export class RedisKVClient implements IKVClient { } async deleteMetadataFlat(hashId: string): Promise { + if (this.writer) { + return this.writer.deleteMetadataFlat(hashId); + } + const keyPrefix = this.buildKey(`file:${hashId}/`); // Scan and delete all keys with this prefix @@ -378,6 +467,9 @@ export class RedisKVClient implements IKVClient { } async getAllHashIds(): Promise { + if (this.writer) { + return this.writer.getAllHashIds(); + } const indexKey = this.buildKey('file:__index__'); return await this.redis.smembers(indexKey); }