From 942c06634a979b101ee86165380c1db9e80fdd17 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sun, 8 Feb 2026 21:53:52 +0100 Subject: [PATCH] new flat KV structure --- packages/meta-dup-core/src/DuplicateIndex.ts | 81 +++++++ .../meta-dup-core/src/MetaEventConsumer.ts | 221 ++++++++++++++++++ packages/meta-dup-core/src/index.ts | 33 ++- packages/meta-dup-core/src/kv/RedisClient.ts | 34 +++ 4 files changed, 368 insertions(+), 1 deletion(-) create mode 100644 packages/meta-dup-core/src/MetaEventConsumer.ts diff --git a/packages/meta-dup-core/src/DuplicateIndex.ts b/packages/meta-dup-core/src/DuplicateIndex.ts index 5458b89..79ca243 100644 --- a/packages/meta-dup-core/src/DuplicateIndex.ts +++ b/packages/meta-dup-core/src/DuplicateIndex.ts @@ -415,4 +415,85 @@ export class DuplicateIndex { } } } + + /** + * Update title index for a specific file when metadata changes + * Called by MetaEventConsumer when title-related fields are updated + */ + async updateTitleForFile(hashId: string, redisClient: RedisClient): Promise { + const currentInfo = this.fileInfo.get(hashId); + const oldTitleKey = currentInfo ? this.getTitleKeyFromInfo(currentInfo) : null; + + // Fetch updated metadata from Redis + const metadata = await redisClient.getFileByHashId(hashId); + + if (!metadata) { + // File deleted - remove from title index + if (oldTitleKey) { + this.removeFromTitleIndex(hashId, oldTitleKey); + logger.debug(`Removed deleted file ${hashId} from title index`); + } + return; + } + + const newTitleKey = this.getTitleKey(metadata); + + // Only update if title key changed + if (oldTitleKey !== newTitleKey) { + if (oldTitleKey) { + this.removeFromTitleIndex(hashId, oldTitleKey); + } + if (newTitleKey) { + this.addToTitleIndex(hashId, newTitleKey); + } + // Update cached file info + this.addFile(hashId, metadata); + logger.debug(`Title updated for ${hashId}: "${oldTitleKey}" -> "${newTitleKey}"`); + } + + this.lastUpdated = new Date(); + } + + /** + * Get normalized title key from FileInfo + */ + private getTitleKeyFromInfo(info: FileInfo): string | null { + if (!info.title) return null; + + let key = info.title.toLowerCase().trim(); + + if (info.year) { + key += `:${info.year}`; + } + if (info.season !== undefined) { + key += `:s${info.season}`; + } + if (info.episode !== undefined) { + key += `:e${info.episode}`; + } + + return key; + } + + /** + * Remove a file from the title index by key + */ + private removeFromTitleIndex(hashId: string, titleKey: string): void { + const titleSet = this.byTitle.get(titleKey); + if (titleSet) { + titleSet.delete(hashId); + if (titleSet.size === 0) { + this.byTitle.delete(titleKey); + } + } + } + + /** + * Add a file to the title index by key + */ + private addToTitleIndex(hashId: string, titleKey: string): void { + const titleSet = this.byTitle.get(titleKey) ?? new Set(); + titleSet.add(hashId); + this.byTitle.set(titleKey, titleSet); + } } diff --git a/packages/meta-dup-core/src/MetaEventConsumer.ts b/packages/meta-dup-core/src/MetaEventConsumer.ts new file mode 100644 index 0000000..405c041 --- /dev/null +++ b/packages/meta-dup-core/src/MetaEventConsumer.ts @@ -0,0 +1,221 @@ +/** + * MetaEventConsumer - Consumes meta:events stream for title updates + * + * meta-core publishes metadata changes to the meta:events stream via MetaPublisher. + * This consumer listens for title-related field changes and updates the DuplicateIndex + * to keep title duplicate detection current as plugins like TMDB update metadata. + * + * Event format from meta-core: + * {type: "set"|"del", key: "file:{hashId}/{field}", ts: ...} + */ + +import { Redis } from 'ioredis'; +import { Logger } from 'tslog'; + +const logger = new Logger({ name: 'MetaEventConsumer' }); + +const META_EVENTS_STREAM = 'meta:events'; +const CONSUMER_GROUP = 'meta-dup-meta-consumer'; + +// Fields that affect title duplicate detection +const TITLE_FIELDS = [ + 'title', 'titles/eng', 'originalTitle', 'fileName', + 'year', 'movieYear', 'season', 'episode', 'tmdb' +]; + +// Debounce time in ms for rapid updates (multiple fields updated at once) +const DEBOUNCE_MS = 500; + +export class MetaEventConsumer { + private redis: Redis; + private consumerName: string; + private running = false; + private pendingUpdates: Map = new Map(); + private onTitleChange: (hashId: string) => Promise; + + constructor( + redis: Redis, + consumerName: string, + onTitleChange: (hashId: string) => Promise + ) { + this.redis = redis; + this.consumerName = consumerName; + this.onTitleChange = onTitleChange; + } + + /** + * Start consuming meta:events stream + */ + async start(): Promise { + if (this.running) { + return; + } + + // Create consumer group at '$' (only new messages) + // We don't need historical events - the initial rebuild already loaded all titles + try { + await this.redis.xgroup('CREATE', META_EVENTS_STREAM, CONSUMER_GROUP, '$', 'MKSTREAM'); + logger.info(`Created consumer group '${CONSUMER_GROUP}'`); + } catch (error: any) { + if (error.message?.includes('BUSYGROUP')) { + logger.debug(`Consumer group '${CONSUMER_GROUP}' already exists`); + } else { + throw error; + } + } + + this.running = true; + logger.info(`Starting meta:events consumer (consumer: ${this.consumerName})`); + + // Start consume loop + this.consumeLoop().catch(error => { + if (this.running) { + logger.error('meta:events consumer loop error:', error); + } + }); + } + + /** + * Stop the consumer + */ + stop(): void { + if (!this.running) { + return; + } + + logger.info('Stopping meta:events consumer...'); + this.running = false; + + // Clear pending debounced updates + for (const timeout of this.pendingUpdates.values()) { + clearTimeout(timeout); + } + this.pendingUpdates.clear(); + } + + /** + * Main consume loop - reads from stream and processes messages + */ + private async consumeLoop(): Promise { + while (this.running) { + try { + // XREADGROUP with 5s block timeout + const result = await this.redis.call( + 'XREADGROUP', + 'GROUP', CONSUMER_GROUP, + this.consumerName, + 'BLOCK', 5000, + 'COUNT', 100, + 'STREAMS', META_EVENTS_STREAM, + '>' + ) as Array<[string, Array<[string, string[]]>]> | null; + + if (!result || result.length === 0) { + continue; + } + + // Process each stream's messages + for (const [streamName, entries] of result) { + for (const [id, fields] of entries) { + try { + await this.processMessage(id, fields); + // ACK after successful processing + await this.redis.xack(META_EVENTS_STREAM, CONSUMER_GROUP, id); + } catch (error: any) { + logger.error(`Error processing meta event ${id}:`, error.message); + // Still ACK to prevent blocking on bad messages + await this.redis.xack(META_EVENTS_STREAM, CONSUMER_GROUP, id); + } + } + } + } catch (error: any) { + if (!this.running) { + break; + } + logger.error('Error reading meta:events:', error.message); + // Brief pause before retrying + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + + logger.info('meta:events consumer stopped'); + } + + /** + * Process a single message from the stream + */ + private async processMessage(id: string, fields: string[]): Promise { + // Parse fields array into map + const fieldMap: Record = {}; + for (let i = 0; i < fields.length; i += 2) { + fieldMap[fields[i]] = fields[i + 1]; + } + + const type = fieldMap.type; + const key = fieldMap.key; + + if (!type || !key) { + return; + } + + // Only care about 'set' events (new/updated values) + if (type !== 'set') { + return; + } + + // Check if this is a title-related field + if (!this.isTitleField(key)) { + return; + } + + // Extract hashId from key format: file:{hashId}/{field} + const hashId = this.extractHashId(key); + if (!hashId) { + return; + } + + logger.debug(`Title field changed: ${key}`); + + // Debounce: multiple fields may be updated at once (e.g., title + year from TMDB) + // Wait a short time to batch updates for the same file + this.debounceUpdate(hashId); + } + + /** + * Check if a key represents a title-related field + */ + private isTitleField(key: string): boolean { + return TITLE_FIELDS.some(field => key.endsWith(`/${field}`)); + } + + /** + * Extract hashId from key format: file:{hashId}/{field} + */ + private extractHashId(key: string): string | null { + const match = key.match(/^file:([^/]+)\//); + return match ? match[1] : null; + } + + /** + * Debounce updates for the same file + */ + private debounceUpdate(hashId: string): void { + // Clear existing timeout if any + const existing = this.pendingUpdates.get(hashId); + if (existing) { + clearTimeout(existing); + } + + // Set new timeout + const timeout = setTimeout(async () => { + this.pendingUpdates.delete(hashId); + try { + await this.onTitleChange(hashId); + } catch (error: any) { + logger.error(`Error updating title for ${hashId}:`, error.message); + } + }, DEBOUNCE_MS); + + this.pendingUpdates.set(hashId, timeout); + } +} diff --git a/packages/meta-dup-core/src/index.ts b/packages/meta-dup-core/src/index.ts index e1fff54..30e40de 100644 --- a/packages/meta-dup-core/src/index.ts +++ b/packages/meta-dup-core/src/index.ts @@ -10,11 +10,14 @@ */ import 'dotenv/config'; +import * as os from 'os'; +import { Redis } from 'ioredis'; import { Logger } from 'tslog'; import { KVManager } from './kv/KVManager.js'; import { RedisClient, StreamMessage } from './kv/RedisClient.js'; import { DuplicateIndex } from './DuplicateIndex.js'; import { APIServer } from './api/APIServer.js'; +import { MetaEventConsumer } from './MetaEventConsumer.js'; const logger = new Logger({ name: 'meta-dup' }); @@ -169,16 +172,26 @@ async function main(): Promise { let redisClient: RedisClient | null = null; let duplicateIndex: DuplicateIndex | null = null; let apiServer: APIServer | null = null; + let metaEventConsumer: MetaEventConsumer | null = null; + let metaRedis: Redis | null = null; // Graceful shutdown const shutdown = async (signal: string): Promise => { logger.info(`Received ${signal}, shutting down...`); try { - // Stop stream consumer first + // Stop stream consumers first + if (metaEventConsumer) { + metaEventConsumer.stop(); + } if (redisClient) { redisClient.stopStreamConsumer(); } + // Close meta:events Redis connection + if (metaRedis) { + await metaRedis.quit(); + metaRedis = null; + } if (apiServer) await apiServer.stop(); await kvManager.stop(); logger.info('Shutdown complete'); @@ -239,6 +252,24 @@ async function main(): Promise { logger.error('Stream consumer error:', error); }); + // Start meta:events consumer for title updates + // This catches metadata changes from plugins like TMDB + metaRedis = redisClient.createDuplicateConnection(); + if (metaRedis) { + const consumerName = `${os.hostname()}-${process.pid}-meta`; + metaEventConsumer = new MetaEventConsumer( + metaRedis, + consumerName, + async (hashId: string) => { + await duplicateIndex!.updateTitleForFile(hashId, redisClient!); + } + ); + await metaEventConsumer.start(); + logger.info('meta:events consumer started for title updates'); + } else { + logger.warn('Could not create meta:events consumer - title updates will not be tracked'); + } + // Initialize API server apiServer = new APIServer(duplicateIndex, { port: config.apiPort, diff --git a/packages/meta-dup-core/src/kv/RedisClient.ts b/packages/meta-dup-core/src/kv/RedisClient.ts index 93db7a9..b75889a 100644 --- a/packages/meta-dup-core/src/kv/RedisClient.ts +++ b/packages/meta-dup-core/src/kv/RedisClient.ts @@ -730,4 +730,38 @@ export class RedisClient implements Partial { getRawClient(): Redis | null { return this.client; } + + /** + * Create a duplicate Redis connection for blocking stream consumers. + * Each blocking consumer needs its own connection since XREADGROUP blocks. + * + * The caller is responsible for closing this connection. + */ + createDuplicateConnection(): Redis | null { + if (!this.isConnected) { + logger.warn('Cannot create duplicate connection: not connected'); + return null; + } + + try { + const connection = new Redis(this.config.url, { + retryStrategy: (times: number) => { + if (times > 10) { + return null; + } + return Math.min(times * 500, 5000); + }, + maxRetriesPerRequest: 3, + }); + + connection.on('error', (err: Error) => { + logger.error('Duplicate Redis connection error:', err.message); + }); + + return connection; + } catch (error) { + logger.error('Failed to create duplicate connection:', error); + return null; + } + } }