From d1ba5b8cc5223d2e3cbb2d5c8e9325fa6f51b1e3 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 8 Feb 2026 21:54:10 +0100 Subject: [PATCH] new flat KV structure --- .../src/api/UnifiedAPIServer.ts | 15 +- .../src/events/FileEventConsumer.ts | 16 +- packages/meta-sort-core/src/kv/RedisClient.ts | 139 ++++++++++++------ 3 files changed, 107 insertions(+), 63 deletions(-) diff --git a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts index acb09f4..8470d7d 100644 --- a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts +++ b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts @@ -829,25 +829,22 @@ export class UnifiedAPIServer { try { console.log('Clearing all metadata via API'); - // Get all file keys and delete them + // Get all file keys and delete them using flat key deletion const hashIds = await this.kvClient.getAllHashIds(); let deletedCount = 0; for (const hashId of hashIds) { try { - await this.kvClient.delete(`/file/${hashId}`); + // Use deleteMetadataFlat to delete all flat keys for this file + await this.kvClient.deleteMetadataFlat(hashId); deletedCount++; } catch (err) { - console.warn(`Failed to delete key /file/${hashId}:`, err); + console.warn(`Failed to delete metadata for ${hashId}:`, err); } } - // Also delete the index key - try { - await this.kvClient.delete('/file/__index__'); - } catch (err) { - // Index might not exist - } + // Note: deleteMetadataFlat already removes from file:__index__ + // No need to delete the index separately console.log(`Cleared ${deletedCount} file metadata entries`); diff --git a/packages/meta-sort-core/src/events/FileEventConsumer.ts b/packages/meta-sort-core/src/events/FileEventConsumer.ts index 7d10358..26812a5 100644 --- a/packages/meta-sort-core/src/events/FileEventConsumer.ts +++ b/packages/meta-sort-core/src/events/FileEventConsumer.ts @@ -15,6 +15,7 @@ interface Pipeline { handleFileAdded(filePath: string, midhash256?: string): void; handleFileChanged(filePath: string, midhash256?: string): void; handleFileDeleted(filePath: string): void; + reset(): Promise; } // Redis Stream configuration @@ -112,19 +113,22 @@ export class FileEventConsumer { return `${this.filesPath}/${relativePath}`; }; - const absolutePath = toAbsolutePath(message.path); - switch (message.type) { + case 'reset': + console.log(`[FileEventConsumer] Received reset event from watcher: ${message.watcherId || 'all'}`); + await this.pipeline.reset(); + break; + case 'add': - this.pipeline.handleFileAdded(absolutePath, message.midhash256); + this.pipeline.handleFileAdded(toAbsolutePath(message.path), message.midhash256); break; case 'change': - this.pipeline.handleFileChanged(absolutePath, message.midhash256); + this.pipeline.handleFileChanged(toAbsolutePath(message.path), message.midhash256); break; case 'delete': - this.pipeline.handleFileDeleted(absolutePath); + this.pipeline.handleFileDeleted(toAbsolutePath(message.path)); break; case 'rename': @@ -132,7 +136,7 @@ export class FileEventConsumer { if (message.oldPath) { this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath)); } - this.pipeline.handleFileAdded(absolutePath, message.midhash256); + this.pipeline.handleFileAdded(toAbsolutePath(message.path), message.midhash256); break; default: diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index 61c69cc..2983d9b 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -1,15 +1,15 @@ /** * Redis KV Client - Implements IKVClient interface using Redis * - * Uses Redis hashes for efficient nested key storage: - * - Each file's metadata is stored as a Redis hash - * - Hash key: file:{hashId} - * - Hash field: property/path (e.g., "video/codec") + * Uses flat keys for file metadata storage: + * - Each file property is stored as a separate Redis key + * - Key format: file:{hashId}/{property} (e.g., "file:abc123/video/codec") + * - Index set: file:__index__ contains all hashIds * - * Benefits of Redis hashes: - * - Single command to get all file metadata (HGETALL) - * - Single command to set multiple properties (HMSET) - * - Atomic operations on file metadata + * Benefits of flat keys: + * - Enables Redis keyspace notifications for field-level changes + * - Allows services to subscribe to specific field patterns (e.g., "file:*\/tmdb") + * - Compatible with meta:events stream for real-time updates */ import * as IORedis from 'ioredis'; @@ -28,15 +28,16 @@ type RedisInstance = InstanceType; /** * Stream message from Redis Streams - * File events from meta-core: add, change, delete, rename + * File events from meta-core: add, change, delete, rename, reset */ export interface StreamMessage { id: string; - type: 'add' | 'change' | 'delete' | 'rename'; + type: 'add' | 'change' | 'delete' | 'rename' | 'reset'; path: string; size?: string; midhash256?: string; // midhash256 CID computed by meta-core oldPath?: string; + watcherId?: string; // For reset events timestamp: string; } @@ -53,6 +54,7 @@ interface RedisClientConfig { export class RedisKVClient implements IKVClient { private redis: RedisInstance; + private streamRedis: RedisInstance | null = null; // Separate connection for blocking stream consumer private timeout: number; private prefix: string; private isConnected: boolean = false; @@ -256,12 +258,13 @@ export class RedisKVClient implements IKVClient { } // ======================================================================== - // High-Level Metadata Operations (Using Redis Hashes) + // High-Level Metadata Operations (Using Flat Keys) // ======================================================================== /** - * Store file metadata using Redis hash - * More efficient than individual keys - single HMSET command + * Store file metadata using flat keys + * Each property is stored as a separate key: file:{hashId}/{property} + * Enables keyspace notifications for field-level changes */ async setMetadataFlat( hashId: string, @@ -273,49 +276,61 @@ export class RedisKVClient implements IKVClient { if (pairs.length === 0) return; - // Convert to Redis hash format - // Hash key: file:{hashId} - // Hash fields: property paths (without /file/{hashId} prefix) - const hashKey = this.buildKey(`file:${hashId}`); - const hashData: Record = {}; + // Use pipeline for efficient batch SET operations + const pipeline = this.redis.pipeline(); for (const pair of pairs) { // Extract field name by removing the prefix const field = pair.key.slice(prefix.length + 1); // +1 for trailing / if (field) { - hashData[field] = pair.value; + // Key format: file:{hashId}/{property} + const redisKey = this.buildKey(`file:${hashId}/${field}`); + pipeline.set(redisKey, pair.value); } } - if (Object.keys(hashData).length > 0) { - await this.redis.hmset(hashKey, hashData); - } + await pipeline.exec(); - // Also store in key index for getAllHashIds + // Maintain index set for getAllHashIds await this.redis.sadd(this.buildKey('file:__index__'), hashId); } async getMetadataFlat(hashId: string): Promise { - const hashKey = this.buildKey(`file:${hashId}`); - const hashData = await this.redis.hgetall(hashKey); + const keyPrefix = this.buildKey(`file:${hashId}/`); - if (!hashData || Object.keys(hashData).length === 0) { - return null; - } + // Scan for all keys with this prefix + const keys: string[] = []; + let cursor = '0'; + do { + const [nextCursor, batch] = await this.redis.scan( + cursor, + 'MATCH', + `${keyPrefix}*`, + 'COUNT', + 1000 + ); + cursor = nextCursor; + keys.push(...batch); + } while (cursor !== '0'); - // Convert Redis hash back to KeyValuePair format for reconstruction + if (keys.length === 0) return null; + + // Get all values + const values = await this.redis.mget(...keys); + + // Reconstruct as KeyValuePair format const prefix = buildFilePrefix(hashId); - const pairs: KeyValuePair[] = Object.entries(hashData).map(([field, value]) => ({ - key: `${prefix}/${field}`, - value: value as string - })); + const pairs: KeyValuePair[] = keys.map((key, i) => { + const field = key.slice(keyPrefix.length); + return { key: `${prefix}/${field}`, value: values[i] as string }; + }).filter(p => p.value !== null); return reconstructMetadata(pairs, prefix); } async getMetadata(hashId: string, propertyPath: string): Promise { - const hashKey = this.buildKey(`file:${hashId}`); - const value = await this.redis.hget(hashKey, propertyPath); + const key = this.buildKey(`file:${hashId}/${propertyPath}`); + const value = await this.redis.get(key); if (value === null) return null; @@ -328,24 +343,38 @@ export class RedisKVClient implements IKVClient { /** * Set a single metadata property for a file - * Writes directly to Redis hash field + * Writes to flat key: file:{hashId}/{property} */ async setMetadataProperty(hashId: string, property: string, value: string): Promise { - const hashKey = this.buildKey(`file:${hashId}`); - await this.redis.hset(hashKey, property, value); + const key = this.buildKey(`file:${hashId}/${property}`); + await this.redis.set(key, value); // Also ensure the file is in the index await this.redis.sadd(this.buildKey('file:__index__'), hashId); } async deleteMetadataFlat(hashId: string): Promise { - const hashKey = this.buildKey(`file:${hashId}`); - const fieldCount = await this.redis.hlen(hashKey); + const keyPrefix = this.buildKey(`file:${hashId}/`); + + // Scan and delete all keys with this prefix + let deletedCount = 0; + let cursor = '0'; + do { + const [nextCursor, keys] = await this.redis.scan( + cursor, + 'MATCH', + `${keyPrefix}*`, + 'COUNT', + 1000 + ); + cursor = nextCursor; + if (keys.length > 0) { + deletedCount += await this.redis.del(...keys); + } + } while (cursor !== '0'); - await this.redis.del(hashKey); await this.redis.srem(this.buildKey('file:__index__'), hashId); - - return fieldCount; + return deletedCount; } async getAllHashIds(): Promise { @@ -498,6 +527,9 @@ export class RedisKVClient implements IKVClient { * Start stream consumer loop * Reads messages using XREADGROUP and calls the handler * + * IMPORTANT: Uses a duplicate connection for the blocking XREADGROUP command + * to avoid blocking regular Redis operations on the main connection. + * * @param stream - Stream name * @param group - Consumer group name * @param onMessage - Handler for each message @@ -517,14 +549,18 @@ export class RedisKVClient implements IKVClient { this.streamConsumerRunning = true; this.streamConsumerAbort = new AbortController(); - console.log(`[Redis] Starting stream consumer: ${stream} (group: ${group}, consumer: ${this.consumerName})`); + // Create a duplicate connection for blocking operations + // This prevents XREADGROUP BLOCK from blocking regular Redis commands + this.streamRedis = this.redis.duplicate(); + console.log(`[Redis] Starting stream consumer: ${stream} (group: ${group}, consumer: ${this.consumerName}) [using dedicated connection]`); // Consumer loop while (this.streamConsumerRunning && !this.streamConsumerAbort.signal.aborted) { try { // XREADGROUP blocks waiting for new messages // '>' means only read new messages (not already delivered to this consumer) - const result = await (this.redis.call( + // Use streamRedis (duplicate connection) for blocking call + const result = await (this.streamRedis.call( 'XREADGROUP', 'GROUP', group, this.consumerName, @@ -545,7 +581,7 @@ export class RedisKVClient implements IKVClient { const message = this.parseStreamEntry(id, fields); if (message) { await onMessage(message); - // ACK the message after successful processing + // ACK the message on main connection (non-blocking) await this.redis.xack(stream, group, id); } } catch (error: any) { @@ -564,6 +600,11 @@ export class RedisKVClient implements IKVClient { } } + // Cleanup duplicate connection + if (this.streamRedis) { + await this.streamRedis.quit(); + this.streamRedis = null; + } console.log('[Redis] Stream consumer stopped'); } @@ -596,7 +637,8 @@ export class RedisKVClient implements IKVClient { return null; } - if (!fieldMap.path) { + // Reset events don't require a path + if (!fieldMap.path && fieldMap.type !== 'reset') { console.warn(`[Redis] Invalid stream entry ${id}: missing path for ${fieldMap.type} event`); return null; } @@ -604,10 +646,11 @@ export class RedisKVClient implements IKVClient { return { id, type: fieldMap.type as StreamMessage['type'], - path: fieldMap.path, + path: fieldMap.path || '', size: fieldMap.size, midhash256: fieldMap.midhash256, oldPath: fieldMap.oldPath, + watcherId: fieldMap.watcherId, timestamp: fieldMap.timestamp || '0', }; }