cleaning of the files event system

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
+// Redis Streams for reliable event delivery from meta-core
-const EVENTS_STREAM = 'meta-sort:events';
+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<void> {
     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

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';
+    type: 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete';
-    payload: string;
+    // 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<IKVClient> {
      * 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<void> {
@@ -668,20 +676,39 @@ export class RedisClient implements Partial<IKVClient> {
      * 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<string, string> = {};
         for (let i = 0; i < fields.length; i += 2) {
             fieldMap[fields[i]] = fields[i + 1];
         }
 
-        if (!fieldMap.type || !fieldMap.payload) {
+        if (!fieldMap.type) {
-            logger.warn(`Invalid stream entry ${id}: missing type or payload`);
+            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',
         };

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<void> {
+        // 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<void> {