diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index 51f2f5b..f4e3579 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -16,11 +16,11 @@ import { APIServer } from './api/APIServer.js'; const logger = new Logger({ name: 'meta-fuse' }); -// Redis Streams for reliable event delivery from meta-sort -const EVENTS_STREAM = 'meta-sort:events'; +// Redis Streams for reliable event delivery from meta-core +const EVENTS_STREAM = 'file:events'; const CONSUMER_GROUP = 'meta-fuse-vfs'; -// Plugins that provide metadata needed for VFS virtual paths +// Plugins that provide metadata needed for VFS virtual paths (for legacy plugin:complete events) // When these complete, we should update the VFS entry for that file const VFS_RELEVANT_PLUGINS = new Set(['filename-parser', 'tmdb', 'jellyfin-nfo']); @@ -47,11 +47,42 @@ interface ResetPayload { /** * Handle a stream message from Redis Streams * Routes messages by type to appropriate handlers + * + * Event types: + * - add/change/delete/rename: Direct file events from meta-core (path-based) + * - batch/reset/plugin:complete: Legacy events from meta-sort (hashId-based) */ async function handleStreamMessage(message: StreamMessage, vfs: VirtualFileSystem): Promise { try { switch (message.type) { + // Direct file events from meta-core + case 'add': { + logger.debug(`File added: ${message.path}`); + await vfs.onFileEventByPath(message.path!, 'add'); + break; + } + + case 'change': { + logger.debug(`File changed: ${message.path}`); + await vfs.onFileEventByPath(message.path!, 'change'); + break; + } + + case 'delete': { + logger.debug(`File deleted: ${message.path}`); + await vfs.onFileEventByPath(message.path!, 'delete'); + break; + } + + case 'rename': { + logger.debug(`File renamed: ${message.oldPath} -> ${message.path}`); + await vfs.onFileEventByPath(message.path!, 'rename', message.oldPath); + break; + } + + // Legacy events from meta-sort (for backward compatibility) case 'batch': { + if (!message.payload) break; const batch: BatchUpdatePayload = JSON.parse(message.payload); logger.debug(`Processing batch update: ${batch.changes.length} changes`); @@ -62,6 +93,7 @@ async function handleStreamMessage(message: StreamMessage, vfs: VirtualFileSyste } case 'reset': { + if (!message.payload) break; const reset: ResetPayload = JSON.parse(message.payload); logger.info(`Processing reset event: ${reset.action}`); await vfs.resetAndReload(); @@ -69,6 +101,7 @@ async function handleStreamMessage(message: StreamMessage, vfs: VirtualFileSyste } case 'plugin:complete': { + if (!message.payload) break; const event: PluginCompletePayload = JSON.parse(message.payload); // Only process plugins that provide VFS-relevant metadata diff --git a/packages/meta-fuse-core/src/kv/RedisClient.ts b/packages/meta-fuse-core/src/kv/RedisClient.ts index d28ecad..93c995d 100644 --- a/packages/meta-fuse-core/src/kv/RedisClient.ts +++ b/packages/meta-fuse-core/src/kv/RedisClient.ts @@ -17,11 +17,19 @@ const logger = new Logger({ name: 'RedisClient' }); /** * Stream message from Redis Streams + * File events from meta-core: add, change, delete, rename + * Legacy events from meta-sort: batch, reset, plugin:complete */ export interface StreamMessage { id: string; - type: 'batch' | 'reset' | 'plugin:complete'; - payload: string; + type: 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete'; + // Direct file event fields (from meta-core) + path?: string; + size?: string; + partialHash?: string; + oldPath?: string; + // Legacy payload field (for backward compatibility) + payload?: string; timestamp: string; } @@ -477,7 +485,7 @@ export class RedisClient implements Partial { * Initialize stream consumer group * Creates the consumer group at position 0 to read all historical events * - * @param stream - Stream name (e.g., 'meta-sort:events') + * @param stream - Stream name (e.g., 'file:events') * @param group - Consumer group name (e.g., 'meta-fuse-vfs') */ async initStreamConsumer(stream: string, group: string): Promise { @@ -668,20 +676,39 @@ export class RedisClient implements Partial { * Parse a stream entry into a StreamMessage */ private parseStreamEntry(id: string, fields: string[]): StreamMessage | null { - // Fields come as flat array: ['type', 'batch', 'payload', '...', 'timestamp', '123'] + // 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 || !fieldMap.payload) { - logger.warn(`Invalid stream entry ${id}: missing type or payload`); + if (!fieldMap.type) { + logger.warn(`Invalid stream entry ${id}: missing type`); + return null; + } + + // For direct file events (add/change/delete/rename), path is required + // For legacy events (batch/reset/plugin:complete), payload is required + const directEventTypes = ['add', 'change', 'delete', 'rename']; + const isDirectEvent = directEventTypes.includes(fieldMap.type); + + if (isDirectEvent && !fieldMap.path) { + logger.warn(`Invalid stream entry ${id}: missing path for ${fieldMap.type} event`); + return null; + } + + if (!isDirectEvent && !fieldMap.payload) { + logger.warn(`Invalid stream entry ${id}: missing payload 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, payload: fieldMap.payload, timestamp: fieldMap.timestamp || '0', }; diff --git a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts index f78dd34..87fc894 100644 --- a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts +++ b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts @@ -264,7 +264,74 @@ export class VirtualFileSystem { } /** - * Handle incremental file update from pub/sub + * Handle file event by path (from meta-core) + * Used for direct file events where we have the path but not the hashId + * + * @param path - File path relative to FILES_VOLUME + * @param action - Event type: add, change, delete, rename + * @param oldPath - For rename events, the old path + */ + async onFileEventByPath(path: string, action: 'add' | 'change' | 'delete' | 'rename', oldPath?: string): Promise { + // Construct source path by prepending FILES_VOLUME + const sourcePath = path.startsWith('/') ? `${this.config.filesPath}${path}` : `${this.config.filesPath}/${path}`; + + if (action === 'delete') { + // Find and remove file by source path + const virtualPath = this.sourcePathToVirtualPath.get(sourcePath); + if (virtualPath) { + this.removeFile(virtualPath); + logger.debug(`File deleted: ${virtualPath}`); + } + return; + } + + if (action === 'rename' && oldPath) { + // Remove old file first + const oldSourcePath = oldPath.startsWith('/') ? `${this.config.filesPath}${oldPath}` : `${this.config.filesPath}/${oldPath}`; + const oldVirtualPath = this.sourcePathToVirtualPath.get(oldSourcePath); + if (oldVirtualPath) { + this.removeFile(oldVirtualPath); + } + // Fall through to add the renamed file + } + + // For add/change/rename: Look up metadata in Redis by path + // We need to find the hashId for this path, then get the full metadata + const files = await this.redisClient.getAllFiles(); + const metadata = files.get(sourcePath); + + if (!metadata) { + // File exists on disk but not yet processed (metadata not in Redis) + // This is expected - meta-core detects files before meta-sort processes them + logger.debug(`File not yet in Redis: ${sourcePath}`); + return; + } + + // Compute virtual path for this file + const virtualPath = this.computeVirtualPath(metadata, sourcePath); + + if (!virtualPath) { + logger.debug(`Could not compute virtual path for: ${sourcePath}`); + return; + } + + // If updating, remove old entry first + if (action === 'change') { + const oldVirtualPath = this.sourcePathToVirtualPath.get(sourcePath); + if (oldVirtualPath && oldVirtualPath !== virtualPath) { + this.removeFile(oldVirtualPath); + } + } + + // Add/update file + await this.addFile(virtualPath, metadata); + this.sourcePathToVirtualPath.set(sourcePath, virtualPath); + + logger.debug(`File ${action}: ${virtualPath}`); + } + + /** + * Handle incremental file update from pub/sub (legacy, hashId-based) * Adds or updates a single file without full rebuild */ async onFileUpdate(hashId: string, action: 'add' | 'update' | 'remove'): Promise {