adaptation related to meta-core api changes
This commit is contained in:
Pierre
@@ -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<void>;
|
||||
}
|
||||
|
||||
// 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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
private async handleSSEEvent(e: SSEEvent): Promise<void> {
|
||||
// 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<FileEventConsumerOptions>
|
||||
): FileEventConsumer {
|
||||
return new FileEventConsumer({
|
||||
kvClient,
|
||||
metaCoreApiUrl,
|
||||
pipeline,
|
||||
...options
|
||||
});
|
||||
|
||||
@@ -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 (`<ms>-<seq>`).
|
||||
* 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> | 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> | 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<Omit<SSEClientOptions, 'onGap'>> & Pick<SSEClientOptions, 'onGap'>;
|
||||
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<void> {
|
||||
await this.loadCursor();
|
||||
this.startFlushTimer();
|
||||
await this.runLoop();
|
||||
}
|
||||
|
||||
/** Graceful shutdown — closes the open connection and flushes the cursor. */
|
||||
async stop(): Promise<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
this.abort = new AbortController();
|
||||
const headers: Record<string, string> = {
|
||||
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<void> {
|
||||
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<void> {
|
||||
return new Promise(resolve => setTimeout(resolve, ms));
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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<RedisKVClient> {
|
||||
private async createRedisClient(redisUrl: string, apiUrl?: string): Promise<RedisKVClient> {
|
||||
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;
|
||||
|
||||
@@ -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<void> {
|
||||
// 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<string, string> = {};
|
||||
for (const pair of pairs) {
|
||||
// pair.key is "file:<hash>/<field>"; 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<void> {
|
||||
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<number> {
|
||||
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<any | null> {
|
||||
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<any | null> {
|
||||
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<string[]> {
|
||||
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<boolean> {
|
||||
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<unknown> {
|
||||
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<unknown> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<void> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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<any | null> {
|
||||
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<any | null> {
|
||||
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<void> {
|
||||
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<number> {
|
||||
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<string[]> {
|
||||
if (this.writer) {
|
||||
return this.writer.getAllHashIds();
|
||||
}
|
||||
const indexKey = this.buildKey('file:__index__');
|
||||
return await this.redis.smembers(indexKey);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user