diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 787fed1..9f1c702 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -19,13 +19,9 @@ import type { } from './types.js'; import type { TaskQueueType } from '../plugin-engine/types.js'; import type { IKVClient } from '../kv/IKVClient.js'; -import { hasStreamSupport, type ExtendedContainerTask } from '../types/ExtendedInterfaces.js'; +import { type ExtendedContainerTask } from '../types/ExtendedInterfaces.js'; import { config } from '../config/EnvConfig.js'; -// Redis Streams for reliable event delivery -const EVENTS_STREAM = 'meta-sort:events'; -const STREAM_MAXLEN = 10000; - /** * Scheduler events */ @@ -193,6 +189,7 @@ export class ContainerPluginScheduler extends EventEmitter { /** * Mark a plugin as completed for a file and notify waiters + * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream) */ private markPluginCompleted(fileHash: string, pluginId: string, filePath?: string): void { // Track completion @@ -210,37 +207,6 @@ export class ContainerPluginScheduler extends EventEmitter { } this.dependencyWaiters.delete(key); } - - // Publish to Redis for external subscribers (e.g., meta-stremio) - this.publishPluginComplete(fileHash, pluginId, filePath || ''); - } - - /** - * Publish plugin completion event to Redis Streams - * - * External services (like meta-fuse, meta-stremio) consume this stream - * to be notified when specific plugins complete for a file. - */ - private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise { - if (!this.kvClient || !hasStreamSupport(this.kvClient)) { - return; - } - - try { - const payload = JSON.stringify({ - fileHash, - pluginId, - filePath, - timestamp: Date.now() - }); - await this.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { - type: 'plugin:complete', - payload, - timestamp: Date.now().toString() - }); - } catch (error) { - // Silent fail - stream publish is optional - } } /** diff --git a/packages/meta-sort-core/src/events/EventSubscriber.ts b/packages/meta-sort-core/src/events/EventSubscriber.ts deleted file mode 100644 index f1f3d33..0000000 --- a/packages/meta-sort-core/src/events/EventSubscriber.ts +++ /dev/null @@ -1,270 +0,0 @@ -/** - * EventSubscriber - * - * Subscribes to 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 - * file discovery events from meta-core and processes files via WebDAV. - */ - -import { EventSource } from 'eventsource'; - -// File event types matching meta-core's watcher -interface FileEvent { - type: 'add' | 'change' | 'delete' | 'rename'; - path: string; // Relative to FILES_PATH - size?: number; - timestamp: number; - partialHash?: string; // Hash of first 64KB - oldPath?: string; // For rename events -} - -// Pipeline interface (minimal, matching StreamingPipeline) -interface Pipeline { - handleFileAdded(filePath: string): void; - handleFileChanged(filePath: string): void; - handleFileDeleted(filePath: string): void; -} - -export interface EventSubscriberOptions { - /** URL of meta-core API (e.g., http://meta-core or http://localhost:8083) */ - metaCoreUrl: string; - - /** Pipeline to receive events */ - pipeline: Pipeline; - - /** Base path for files (e.g., /files). Paths from meta-core are relative and need this prefix. */ - filesPath?: string; - - /** Whether to request an initial scan when connecting */ - requestInitialScan?: boolean; - - /** Reconnect delay in milliseconds */ - reconnectDelayMs?: number; - - /** Maximum reconnect attempts (0 = unlimited) */ - maxReconnectAttempts?: number; -} - -export class EventSubscriber { - private options: Required; - private eventSource: EventSource | null = null; - private isRunning = false; - private reconnectAttempts = 0; - private reconnectTimer: NodeJS.Timeout | null = null; - - constructor(options: EventSubscriberOptions) { - this.options = { - filesPath: '/files', - requestInitialScan: true, - reconnectDelayMs: 5000, - maxReconnectAttempts: 0, // Unlimited by default - ...options - }; - } - - /** - * Start subscribing to events from meta-core - */ - async start(): Promise { - if (this.isRunning) { - console.log('[EventSubscriber] Already running'); - return; - } - - this.isRunning = true; - console.log(`[EventSubscriber] Connecting to ${this.options.metaCoreUrl}/api/events/subscribe`); - - await this.connect(); - } - - /** - * Stop subscribing to events - */ - async stop(): Promise { - this.isRunning = false; - - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; - } - - if (this.eventSource) { - this.eventSource.close(); - this.eventSource = null; - } - - console.log('[EventSubscriber] Stopped'); - } - - // NOTE: triggerScan() has been removed - meta-core handles scanning automatically - // To trigger a scan manually, call meta-core's /api/scan/trigger endpoint directly - - /** - * Get watcher status from meta-core - */ - async getStatus(): Promise<{ - status: string; - scanning: boolean; - lastScan: number; - fileCount: number; - } | null> { - try { - const response = await fetch(`${this.options.metaCoreUrl}/api/scan/status`); - - if (!response.ok) { - return null; - } - - return await response.json(); - } catch (error) { - console.error('[EventSubscriber] Failed to get status:', error); - return null; - } - } - - /** - * Poll for events since a timestamp (fallback for SSE) - */ - async pollEvents(sinceMs: number, limit = 100): Promise { - try { - const url = `${this.options.metaCoreUrl}/api/events/poll?since=${sinceMs}&limit=${limit}`; - const response = await fetch(url); - - if (!response.ok) { - return []; - } - - const data = await response.json(); - return data.events || []; - } catch (error) { - console.error('[EventSubscriber] Failed to poll events:', error); - return []; - } - } - - private async connect(): Promise { - if (!this.isRunning) return; - - const url = `${this.options.metaCoreUrl}/api/events/subscribe`; - - try { - this.eventSource = new EventSource(url); - - this.eventSource.onopen = () => { - console.log('[EventSubscriber] Connected to meta-core SSE'); - this.reconnectAttempts = 0; - // NOTE: Initial scan is handled automatically by meta-core (Architecture V3) - }; - - this.eventSource.onerror = (error) => { - console.error('[EventSubscriber] SSE connection error:', error); - this.scheduleReconnect(); - }; - - // Handle file events - this.eventSource.addEventListener('file', (event: MessageEvent) => { - try { - const fileEvent: FileEvent = JSON.parse(event.data); - this.handleEvent(fileEvent); - } catch (error) { - console.error('[EventSubscriber] Failed to parse event:', error); - } - }); - - // Handle connected event - this.eventSource.addEventListener('connected', () => { - console.log('[EventSubscriber] Received connected event'); - }); - - } catch (error) { - console.error('[EventSubscriber] Failed to connect:', error); - this.scheduleReconnect(); - } - } - - private handleEvent(event: FileEvent): void { - const { pipeline, filesPath } = this.options; - - // 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 - return `${filesPath}/${relativePath}`; - }; - - const absolutePath = toAbsolutePath(event.path); - - switch (event.type) { - case 'add': - pipeline.handleFileAdded(absolutePath); - break; - - case 'change': - pipeline.handleFileChanged(absolutePath); - break; - - case 'delete': - pipeline.handleFileDeleted(absolutePath); - break; - - case 'rename': - // Handle rename as delete + add - if (event.oldPath) { - pipeline.handleFileDeleted(toAbsolutePath(event.oldPath)); - } - pipeline.handleFileAdded(absolutePath); - break; - - default: - console.warn(`[EventSubscriber] Unknown event type: ${event.type}`); - } - } - - private scheduleReconnect(): void { - if (!this.isRunning) return; - - // Close existing connection - if (this.eventSource) { - this.eventSource.close(); - this.eventSource = null; - } - - // Check reconnect limits - if (this.options.maxReconnectAttempts > 0 && - this.reconnectAttempts >= this.options.maxReconnectAttempts) { - console.error('[EventSubscriber] Max reconnect attempts reached, giving up'); - this.isRunning = false; - return; - } - - this.reconnectAttempts++; - const delay = this.options.reconnectDelayMs * Math.min(this.reconnectAttempts, 6); // Max 30s - - console.log(`[EventSubscriber] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`); - - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - this.connect(); - }, delay); - } -} - -/** - * Create an EventSubscriber with default configuration - */ -export function createEventSubscriber( - metaCoreUrl: string, - pipeline: Pipeline, - options?: Partial -): EventSubscriber { - return new EventSubscriber({ - metaCoreUrl, - pipeline, - ...options - }); -} diff --git a/packages/meta-sort-core/src/events/FileEventConsumer.ts b/packages/meta-sort-core/src/events/FileEventConsumer.ts new file mode 100644 index 0000000..3fd6ea4 --- /dev/null +++ b/packages/meta-sort-core/src/events/FileEventConsumer.ts @@ -0,0 +1,157 @@ +/** + * FileEventConsumer + * + * Consumes file events from meta-core via Redis Stream. + * When events are received, they are forwarded to the StreamingPipeline. + * + * This enables meta-sort to run without direct filesystem access - it receives + * file discovery events from meta-core and processes files via WebDAV. + */ + +import type { RedisKVClient, StreamMessage } from '../kv/RedisClient.js'; + +// Pipeline interface (minimal, matching StreamingPipeline) +interface Pipeline { + handleFileAdded(filePath: string): void; + handleFileChanged(filePath: string): void; + handleFileDeleted(filePath: string): void; +} + +// Redis Stream configuration +const EVENTS_STREAM = 'file:events'; +const CONSUMER_GROUP = 'meta-sort-processor'; + +export interface FileEventConsumerOptions { + /** Redis KV client */ + kvClient: RedisKVClient; + + /** Pipeline to receive events */ + pipeline: Pipeline; + + /** 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; +} + +export class FileEventConsumer { + private kvClient: RedisKVClient; + private pipeline: Pipeline; + private filesPath: string; + private blockMs: number; + private isRunning = false; + + constructor(options: FileEventConsumerOptions) { + this.kvClient = options.kvClient; + this.pipeline = options.pipeline; + this.filesPath = options.filesPath ?? '/files'; + this.blockMs = options.blockMs ?? 5000; + } + + /** + * Start consuming events from Redis Stream + */ + 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] Stream consumer started'); + } + + /** + * Stop consuming events + */ + stop(): void { + if (!this.isRunning) { + return; + } + + this.isRunning = false; + this.kvClient.stopStreamConsumer(); + console.log('[FileEventConsumer] Stopped'); + } + + /** + * Handle a stream message + */ + private async handleMessage(message: StreamMessage): 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 + return `${this.filesPath}/${relativePath}`; + }; + + const absolutePath = toAbsolutePath(message.path); + + switch (message.type) { + case 'add': + this.pipeline.handleFileAdded(absolutePath); + break; + + case 'change': + this.pipeline.handleFileChanged(absolutePath); + break; + + case 'delete': + this.pipeline.handleFileDeleted(absolutePath); + break; + + case 'rename': + // Handle rename as delete + add + if (message.oldPath) { + this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath)); + } + this.pipeline.handleFileAdded(absolutePath); + break; + + default: + console.warn(`[FileEventConsumer] Unknown event type: ${message.type}`); + } + } +} + +/** + * Create a FileEventConsumer with default configuration + */ +export function createFileEventConsumer( + kvClient: RedisKVClient, + pipeline: Pipeline, + options?: Partial +): FileEventConsumer { + return new FileEventConsumer({ + kvClient, + pipeline, + ...options + }); +} diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index 25c1ce6..06f02b1 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -11,7 +11,8 @@ import {KVManager} from "./kv/KVManager.js"; import {getPluginManagerInstance, initializePluginManager} from "./logic/fileProcessor/FileProcessorPiscina.js"; import {ContainerManager} from "./container-plugins/ContainerManager.js"; import {ContainerPluginScheduler} from "./container-plugins/ContainerPluginScheduler.js"; -import {EventSubscriber} from "./events/EventSubscriber.js"; +import {FileEventConsumer} from "./events/FileEventConsumer.js"; +import {RedisKVClient} from "./kv/RedisClient.js"; try { // Get current user information @@ -39,8 +40,8 @@ let kvManager: KVManager | null = null; let containerManager: ContainerManager | null = null; let containerPluginScheduler: ContainerPluginScheduler | null = null; -// Event Subscriber for receiving events from meta-core (Architecture V3) -let eventSubscriber: EventSubscriber | null = null; +// File Event Consumer for receiving events from meta-core via Redis Stream +let fileEventConsumer: FileEventConsumer | null = null; // Calculate concurrency for pipeline stages const cpuCount = os.cpus().length; @@ -102,10 +103,10 @@ process.on('SIGINT', async () => { await apiServer.stop(); } - // Stop Event Subscriber if running - if (eventSubscriber) { - console.log('[Shutdown] Stopping Event Subscriber...'); - await eventSubscriber.stop(); + // Stop File Event Consumer if running + if (fileEventConsumer) { + console.log('[Shutdown] Stopping File Event Consumer...'); + fileEventConsumer.stop(); } // Stop Container Manager if running @@ -246,24 +247,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 SSE (Architecture V3) - // Get meta-core URL from leader discovery - const leaderInfo = kvManager?.getLeaderInfo(); - if (!leaderInfo) { - console.error('[Startup] ERROR: No leader info available. Cannot connect to meta-core.'); - console.error('[Startup] Make sure meta-core is running and kv-leader.info exists.'); + // 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.'); process.exit(1); } - const metaCoreUrl = leaderInfo.apiUrl; - console.log(`[Startup] Connecting to meta-core at ${metaCoreUrl}...`); - eventSubscriber = new EventSubscriber({ - metaCoreUrl: metaCoreUrl, + console.log('[Startup] Starting File Event Consumer (Redis Stream)...'); + fileEventConsumer = new FileEventConsumer({ + kvClient: kvClient as RedisKVClient, pipeline: pipeline, - requestInitialScan: true, // Request initial scan from meta-core - reconnectDelayMs: 5000, + filesPath: config.FILES_PATH, }); - await eventSubscriber.start(); - console.log('[Startup] EventSubscriber connected to meta-core'); + await fileEventConsumer.start(); + console.log('[Startup] FileEventConsumer started, consuming from file:events stream'); } catch (error) { console.error('[Startup] Error during startup:', error); diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index b43be6f..18cbdff 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -13,6 +13,7 @@ */ import * as IORedis from 'ioredis'; +import * as os from 'os'; import type { IKVClient, KeyValuePair } from './IKVClient.js'; import { flattenMetadata, @@ -25,6 +26,25 @@ import { const Redis = (IORedis as any).default ?? IORedis; type RedisInstance = InstanceType; +/** + * Stream message from Redis Streams + * File events from meta-core: add, change, delete, rename + */ +export interface StreamMessage { + id: string; + type: 'add' | 'change' | 'delete' | 'rename'; + path: string; + size?: string; + partialHash?: string; + oldPath?: string; + timestamp: string; +} + +/** + * Stream consumer callback + */ +export type StreamMessageHandler = (message: StreamMessage) => Promise; + interface RedisClientConfig { url: string; timeout?: number; @@ -37,10 +57,18 @@ export class RedisKVClient implements IKVClient { private prefix: string; private isConnected: boolean = false; + // Stream consumer state + private streamConsumerRunning = false; + private streamConsumerAbort: AbortController | null = null; + private consumerName: string; + constructor(config: RedisClientConfig) { this.timeout = config.timeout || 30000; this.prefix = config.prefix || ''; + // Unique consumer name for this instance + this.consumerName = `${os.hostname()}-${process.pid}`; + // Parse Redis URL and create client this.redis = new Redis(config.url, { commandTimeout: this.timeout, @@ -371,11 +399,226 @@ export class RedisKVClient implements IKVClient { return result as string; } + // ======================================================================== + // Redis Streams Consumer Methods + // ======================================================================== + + /** + * Initialize stream consumer group + * Creates the consumer group at position 0 to read all historical events + * + * @param stream - Stream name (e.g., 'file:events') + * @param group - Consumer group name (e.g., 'meta-sort-processor') + */ + async initStreamConsumer(stream: string, group: string): Promise { + try { + // Create consumer group at position 0 (read all historical events) + // MKSTREAM creates the stream if it doesn't exist + await this.redis.xgroup('CREATE', stream, group, '0', 'MKSTREAM'); + console.log(`[Redis] Created consumer group '${group}' for stream '${stream}'`); + } catch (error: any) { + // BUSYGROUP means group already exists - that's fine + if (error.message?.includes('BUSYGROUP')) { + console.log(`[Redis] Consumer group '${group}' already exists`); + } else { + throw error; + } + } + } + + /** + * Process pending entries from crashed consumers + * Uses XAUTOCLAIM to claim entries that have been idle too long + * + * @param stream - Stream name + * @param group - Consumer group name + * @param minIdleTime - Minimum idle time in ms before claiming (default: 30000) + * @param onMessage - Handler for each message + */ + async processPendingEntries( + stream: string, + group: string, + minIdleTime: number = 30000, + onMessage: StreamMessageHandler + ): Promise { + let processed = 0; + let cursor = '0-0'; + + try { + while (true) { + // XAUTOCLAIM claims idle entries and returns them + const result = await this.redis.xautoclaim( + stream, + group, + this.consumerName, + minIdleTime, + cursor, + 'COUNT', + 100 + ) as [string, Array<[string, string[]]>, string[]]; + + const [nextCursor, entries] = result; + cursor = nextCursor; + + if (!entries || entries.length === 0) { + break; + } + + for (const [id, fields] of entries) { + try { + const message = this.parseStreamEntry(id, fields); + if (message) { + await onMessage(message); + // ACK the message after successful processing + await this.redis.xack(stream, group, id); + processed++; + } + } catch (error: any) { + console.error(`[Redis] Error processing pending entry ${id}:`, error.message); + } + } + + // If cursor is 0-0, we've processed all pending entries + if (cursor === '0-0') { + break; + } + } + + if (processed > 0) { + console.log(`[Redis] Processed ${processed} pending stream entries`); + } + } catch (error: any) { + console.error('[Redis] Error processing pending entries:', error.message); + } + + return processed; + } + + /** + * Start stream consumer loop + * Reads messages using XREADGROUP and calls the handler + * + * @param stream - Stream name + * @param group - Consumer group name + * @param onMessage - Handler for each message + * @param blockMs - Block timeout in ms (default: 5000) + */ + async startStreamConsumer( + stream: string, + group: string, + onMessage: StreamMessageHandler, + blockMs: number = 5000 + ): Promise { + if (this.streamConsumerRunning) { + console.warn('[Redis] Stream consumer already running'); + return; + } + + this.streamConsumerRunning = true; + this.streamConsumerAbort = new AbortController(); + + console.log(`[Redis] Starting stream consumer: ${stream} (group: ${group}, consumer: ${this.consumerName})`); + + // 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( + 'XREADGROUP', + 'GROUP', group, + this.consumerName, + 'BLOCK', blockMs, + 'COUNT', 10, + 'STREAMS', stream, + '>' + ) as Promise]> | null>); + + if (!result || result.length === 0) { + continue; // Timeout, loop again + } + + // Process each stream's messages + for (const [streamName, entries] of result) { + for (const [id, fields] of entries) { + try { + const message = this.parseStreamEntry(id, fields); + if (message) { + await onMessage(message); + // ACK the message after successful processing + await this.redis.xack(stream, group, id); + } + } catch (error: any) { + console.error(`[Redis] Error processing stream entry ${id}:`, error.message); + // Don't ACK on error - message will be reprocessed + } + } + } + } catch (error: any) { + if (this.streamConsumerAbort?.signal.aborted) { + break; + } + console.error('[Redis] Stream consumer error:', error.message); + // Brief pause before retrying + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + + console.log('[Redis] Stream consumer stopped'); + } + + /** + * Stop the stream consumer loop + */ + stopStreamConsumer(): void { + if (!this.streamConsumerRunning) { + return; + } + + console.log('[Redis] Stopping stream consumer...'); + this.streamConsumerRunning = false; + this.streamConsumerAbort?.abort(); + this.streamConsumerAbort = null; + } + + /** + * Parse a stream entry into a StreamMessage + */ + private parseStreamEntry(id: string, fields: string[]): StreamMessage | null { + // Fields come as flat array: ['type', 'add', 'path', '...', 'timestamp', '123'] + const fieldMap: Record = {}; + for (let i = 0; i < fields.length; i += 2) { + fieldMap[fields[i]] = fields[i + 1]; + } + + if (!fieldMap.type) { + console.warn(`[Redis] Invalid stream entry ${id}: missing type`); + return null; + } + + if (!fieldMap.path) { + console.warn(`[Redis] Invalid stream entry ${id}: missing path for ${fieldMap.type} event`); + return null; + } + + return { + id, + type: fieldMap.type as StreamMessage['type'], + path: fieldMap.path, + size: fieldMap.size, + partialHash: fieldMap.partialHash, + oldPath: fieldMap.oldPath, + timestamp: fieldMap.timestamp || '0', + }; + } + // ======================================================================== // Lifecycle // ======================================================================== async close(): Promise { + // Stop stream consumer first + this.stopStreamConsumer(); await this.redis.quit(); this.isConnected = false; } diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index f0536d4..eefa1fa 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -12,12 +12,8 @@ import {VirtualFileSystem} from "../api/VirtualFileSystem.js"; import {performanceMetrics} from "../metrics/PerformanceMetrics.js"; import {UnifiedProcessingStateManager} from "./UnifiedProcessingStateManager.js"; import type {IKVClient} from "../kv/IKVClient.js"; -import { hasStreamSupport, type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js"; +import { type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js"; import * as webdav from '../webdav/WebdavClient.js'; - -// Redis Streams for reliable event delivery -const EVENTS_STREAM = 'meta-sort:events'; -const STREAM_MAXLEN = 10000; import PQueue from "p-queue"; import os from 'os'; import { TaskScheduler, createTaskScheduler, MetadataNodeKVStore } from "../plugin-engine/index.js"; @@ -205,7 +201,8 @@ export class WatchedFileProcessor implements FileProcessorInterface { } /** - * Set up TaskScheduler event listeners for VFS integration and Redis pub/sub + * Set up TaskScheduler event listeners for VFS integration + * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream) */ private setupTaskSchedulerEvents(): void { if (!this.taskScheduler) return; @@ -218,9 +215,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { if (task.pluginId === 'filename-parser') { this.updateVFSForFile(task.fileHash, task.filePath); } - - // Publish plugin completion event to Redis - this.publishPluginComplete(task.fileHash, task.pluginId, task.filePath); }); // When all tasks for a file complete @@ -238,32 +232,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { }); } - /** - * Publish plugin completion event to Redis Streams - */ - private async publishPluginComplete(fileHash: string, pluginId: string, filePath: string): Promise { - const kvClient = this.getKVClient(); - if (!kvClient || !hasStreamSupport(kvClient)) { - return; - } - - try { - const payload = JSON.stringify({ - fileHash, - pluginId, - filePath, - timestamp: Date.now() - }); - await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { - type: 'plugin:complete', - payload, - timestamp: Date.now().toString() - }); - } catch (error) { - // Silent fail - stream publish is optional - } - } - /** * Update VFS for a file with current metadata */ diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 05cbbc4..9886b6b 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -2,22 +2,6 @@ import * as path from 'path'; import PQueue from 'p-queue'; import {PipelineConfig} from './PipelineConfig.js'; import {performanceMetrics} from '../../metrics/PerformanceMetrics.js'; -import { hasStreamSupport } from '../../types/ExtendedInterfaces.js'; - -// Redis Streams for reliable event delivery to meta-fuse -// Single stream with type field for all event types (batch, reset, plugin:complete) -const EVENTS_STREAM = 'meta-sort:events'; - -// Maximum stream length (approximate, ~10000 entries provides hours of history) -const STREAM_MAXLEN = 10000; - -// Batch update interval (5 seconds) -const BATCH_INTERVAL_MS = 5000; - -interface BatchChange { - action: 'add' | 'update' | 'remove'; - hashId: string; -} /** * Streaming Pipeline for file processing @@ -26,7 +10,7 @@ interface BatchChange { * * With midhash256 being instant (< 1s), files appear in VFS immediately * - * Publishes batched updates to meta-fuse via Redis Streams every 5 seconds. + * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream) */ export class StreamingPipeline { private validationQueue: PQueue; @@ -39,10 +23,6 @@ export class StreamingPipeline { private fastProcessedCount = 0; private backgroundProcessedCount = 0; - // Batch buffer for stream notifications - private batchBuffer: BatchChange[] = []; - private batchTimer: NodeJS.Timeout | null = null; - constructor(config: PipelineConfig) { this.config = config; @@ -65,72 +45,6 @@ export class StreamingPipeline { }); console.log(`[Pipeline] Initialized with concurrency: validation=${config.validationConcurrency}, fastQueue=${config.fastQueueConcurrency}, backgroundQueue=${config.backgroundQueueConcurrency}`); - - // Start batch publish timer - this.startBatchTimer(); - } - - /** - * Start the batch publish timer - * Publishes accumulated changes to meta-fuse every 5 seconds - */ - private startBatchTimer(): void { - this.batchTimer = setInterval(() => { - this.flushBatch().catch(err => { - console.error('[Pipeline] Failed to flush batch:', err.message); - }); - }, BATCH_INTERVAL_MS); - } - - /** - * Stop the batch publish timer - */ - stopBatchTimer(): void { - if (this.batchTimer) { - clearInterval(this.batchTimer); - this.batchTimer = null; - } - } - - /** - * Queue a change for batched notification - */ - private queueChange(action: 'add' | 'update' | 'remove', hashId: string): void { - // Deduplicate: remove existing entries for same hashId - this.batchBuffer = this.batchBuffer.filter(c => c.hashId !== hashId); - this.batchBuffer.push({ action, hashId }); - } - - /** - * Flush the batch buffer and publish to Redis Streams - */ - private async flushBatch(): Promise { - if (this.batchBuffer.length === 0) { - return; - } - - const changes = [...this.batchBuffer]; - this.batchBuffer = []; - - try { - const kvClient = this.config.kvClient; - if (kvClient && hasStreamSupport(kvClient)) { - const payload = JSON.stringify({ - timestamp: Date.now(), - changes - }); - await kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { - type: 'batch', - payload, - timestamp: Date.now().toString() - }); - console.log(`[Pipeline] Published batch update to stream: ${changes.length} changes`); - } - } catch (error: any) { - // Put changes back in buffer on failure - this.batchBuffer = [...changes, ...this.batchBuffer]; - console.warn(`[Pipeline] Failed to publish batch: ${error.message}`); - } } /** @@ -233,11 +147,6 @@ export class StreamingPipeline { // Add to VFS immediately (file appears with permanent midhash256 ID) this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata); - // Queue for pub/sub notification to meta-fuse - if (metadata.cid_midhash256) { - this.queueChange('add', metadata.cid_midhash256); - } - // Notify Stremio addon to add video (incremental update like FUSE) if (metadata.cid_midhash256) { this.notifyStremioAdd(metadata.cid_midhash256).catch(err => { @@ -346,20 +255,12 @@ export class StreamingPipeline { * Handle file deleted event (from chokidar) */ async handleFileDeleted(filePath: string): Promise { - // Get hashId before removal for pub/sub notification - const metadata = this.config.fileProcessor.getDatabase().get(filePath); - const hashId = (metadata as any)?.cid_midhash256; - // Remove from state and call cleanup if (this.config.fileProcessor.deleteFile) { this.config.fileProcessor.deleteFile(filePath); } this.config.stateManager.removeFile(filePath); - - // Queue for pub/sub notification to meta-fuse - if (hashId) { - this.queueChange('remove', hashId); - } + // Note: meta-core handles event publishing to meta-fuse via file:events stream } /** @@ -412,7 +313,7 @@ export class StreamingPipeline { /** * Reset pipeline counters and clear state manager * Called when triggering a fresh scan to reset statistics - * Publishes reset event to notify meta-fuse to clear its VFS + * Note: meta-fuse listens to file:events stream from meta-core for updates */ async reset(): Promise { console.log('[Pipeline] Resetting counters and state...'); @@ -423,30 +324,9 @@ export class StreamingPipeline { this.fastProcessedCount = 0; this.backgroundProcessedCount = 0; - // Clear batch buffer - this.batchBuffer = []; - // Clear state manager (discovered, processing, done states) this.config.stateManager.clear(); - // Publish reset event to Redis Streams to notify meta-fuse - if (this.config.kvClient && hasStreamSupport(this.config.kvClient)) { - try { - const payload = JSON.stringify({ - timestamp: Date.now(), - action: 'reset' - }); - await this.config.kvClient.xadd(EVENTS_STREAM, STREAM_MAXLEN, { - type: 'reset', - payload, - timestamp: Date.now().toString() - }); - console.log('[Pipeline] Reset event published to stream'); - } catch (error) { - console.error('[Pipeline] Failed to publish reset event:', error); - } - } - console.log('[Pipeline] Reset complete'); }