diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index f4e3579..284f028 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -12,112 +12,35 @@ import { Logger } from 'tslog'; import { KVManager } from './kv/KVManager.js'; import { RedisClient, StreamMessage } from './kv/RedisClient.js'; import { VirtualFileSystem } from './vfs/VirtualFileSystem.js'; +import { StreamingStateBuilder } from './vfs/StreamingStateBuilder.js'; import { APIServer } from './api/APIServer.js'; const logger = new Logger({ name: 'meta-fuse' }); -// Redis Streams for reliable event delivery from meta-core -const EVENTS_STREAM = 'file:events'; -const CONSUMER_GROUP = 'meta-fuse-vfs'; +// Redis Streams - using new meta:events stream from meta-core flat key architecture +const EVENTS_STREAM = 'meta:events'; -// 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']); - -interface PluginCompletePayload { - fileHash: string; - pluginId: string; - filePath: string; - timestamp: number; -} - -interface BatchUpdatePayload { - timestamp: number; - changes: Array<{ - action: 'add' | 'update' | 'remove'; - hashId: string; - }>; -} - -interface ResetPayload { - timestamp: number; - action: 'reset'; -} +// Legacy stream name for backward compatibility during transition +const LEGACY_EVENTS_STREAM = 'file:events'; /** - * Handle a stream message from Redis Streams - * Routes messages by type to appropriate handlers + * Handle a stream message from meta:events stream * - * 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) + * New meta:events format (flat key architecture): + * - set/del: Property-level changes, processed by StreamingStateBuilder + * + * The StreamingStateBuilder filters events by VFS-relevant properties, + * fetches values, and notifies VFS when files are complete. */ -async function handleStreamMessage(message: StreamMessage, vfs: VirtualFileSystem): Promise { +async function handleStreamingEvent( + message: StreamMessage, + stateBuilder: StreamingStateBuilder +): 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`); - - for (const change of batch.changes) { - await vfs.onFileUpdate(change.hashId, change.action); - } - break; - } - - case 'reset': { - if (!message.payload) break; - const reset: ResetPayload = JSON.parse(message.payload); - logger.info(`Processing reset event: ${reset.action}`); - await vfs.resetAndReload(); - break; - } - - case 'plugin:complete': { - if (!message.payload) break; - const event: PluginCompletePayload = JSON.parse(message.payload); - - // Only process plugins that provide VFS-relevant metadata - if (VFS_RELEVANT_PLUGINS.has(event.pluginId)) { - logger.debug(`Plugin ${event.pluginId} completed for ${event.fileHash}, updating VFS`); - await vfs.onFileUpdate(event.fileHash, 'update'); - } - break; - } - - default: - logger.warn(`Unknown stream message type: ${message.type}`); - } + await stateBuilder.processEvent(message); } catch (error: any) { logger.error(`Error processing stream message ${message.id}:`, error.message); - throw error; // Re-throw to prevent ACK + throw error; } } @@ -151,6 +74,7 @@ async function main(): Promise { let redisClient: RedisClient | null = null; let vfs: VirtualFileSystem | null = null; + let stateBuilder: StreamingStateBuilder | null = null; let apiServer: APIServer | null = null; // Graceful shutdown @@ -178,7 +102,7 @@ async function main(): Promise { // Handle ready event from KV manager kvManager.onReady(async () => { - logger.info('KV Manager ready, initializing VFS...'); + logger.info('KV Manager ready, initializing VFS with streaming bootstrap...'); try { // Get Redis client from KV manager @@ -201,7 +125,7 @@ async function main(): Promise { logger.info('WebDAV: Not available from leader, FUSE driver will use local filesystem'); } - // Initialize VFS (no periodic refresh - uses Redis Streams) + // Initialize VFS in streaming mode (no HGETALL/SCAN bootstrap) vfs = new VirtualFileSystem(redisClient, { fileMode: parseInt(process.env.FUSE_FILE_MODE ?? '644', 8), directoryMode: parseInt(process.env.FUSE_DIR_MODE ?? '755', 8), @@ -212,31 +136,46 @@ async function main(): Promise { filesPath: config.filesPath, }); - // Start VFS (initial load from Redis) - logger.info('Starting VirtualFileSystem...'); + // Enable streaming mode - VFS will be populated via event callbacks + vfs.enableStreamingMode(); + + // Start VFS (just initializes, no data load in streaming mode) + logger.info('Starting VirtualFileSystem in streaming mode...'); await vfs.start(); - // Initialize Redis Streams consumer - logger.info(`Initializing stream consumer for ${EVENTS_STREAM}...`); - await redisClient.initStreamConsumer(EVENTS_STREAM, CONSUMER_GROUP); + // Create streaming state builder with VFS as callback + stateBuilder = new StreamingStateBuilder({ + redisClient, + rulesConfig: vfs.getRulesConfig(), + vfsCallback: vfs, + filesPath: config.filesPath, + }); - // Process any pending entries from crashed consumers - await redisClient.processPendingEntries( + // Bootstrap: Replay meta:events stream from position 0 + // This rebuilds state from all historical events + logger.info(`Replaying ${EVENTS_STREAM} stream from position 0 (streaming bootstrap)...`); + const bootstrapStart = Date.now(); + + const lastId = await redisClient.replayStream( EVENTS_STREAM, - CONSUMER_GROUP, - 30000, // 30 second idle threshold + '0', async (message: StreamMessage) => { - await handleStreamMessage(message, vfs!); - } + await handleStreamingEvent(message, stateBuilder!); + }, + 100 // Batch size ); - // Start stream consumer in background - logger.info('Starting stream consumer...'); - redisClient.startStreamConsumer( + const bootstrapTime = Date.now() - bootstrapStart; + const stats = vfs.getStats(); + logger.info(`Streaming bootstrap complete: ${stats.fileCount} files in ${bootstrapTime}ms`); + + // Start live stream consumer from where replay left off + logger.info(`Starting live stream consumer from ${lastId}...`); + redisClient.startSimpleStreamConsumer( EVENTS_STREAM, - CONSUMER_GROUP, + lastId, async (message: StreamMessage) => { - await handleStreamMessage(message, vfs!); + await handleStreamingEvent(message, stateBuilder!); }, 5000 // 5 second block timeout ).catch(error => { @@ -256,9 +195,13 @@ async function main(): Promise { logger.info('meta-fuse is ready!'); logger.info(`API: http://${config.apiHost}:${config.apiPort}`); - // Log initial stats - const stats = vfs.getStats(); - logger.info(`VFS: ${stats.fileCount} files, ${stats.directoryCount} directories`); + // Log final stats + const finalStats = vfs.getStats(); + logger.info(`VFS: ${finalStats.fileCount} files, ${finalStats.directoryCount} directories`); + + // Log streaming state builder stats + const builderStats = stateBuilder.getStats(); + logger.info(`State builder: ${builderStats.eventsProcessed} events, ${builderStats.propertiesFetched} properties fetched, ${builderStats.propertiesSkipped} skipped`); } catch (error) { logger.error('Failed to initialize after KV ready:', error); } diff --git a/packages/meta-fuse-core/src/kv/RedisClient.ts b/packages/meta-fuse-core/src/kv/RedisClient.ts index dd19010..b8106fc 100644 --- a/packages/meta-fuse-core/src/kv/RedisClient.ts +++ b/packages/meta-fuse-core/src/kv/RedisClient.ts @@ -17,22 +17,40 @@ 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 + * + * New meta:events format (from meta-core flat key architecture): + * - type: 'set' | 'del' - keyspace notification type + * - key: 'file:abc123/property' - the Redis key that changed + * - ts: timestamp + * + * Legacy file:events format (deprecated): + * - type: add, change, delete, rename, batch, reset, plugin:complete + * - path, size, midhash256, oldPath, payload fields */ export interface StreamMessage { id: string; - type: 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete'; - // Direct file event fields (from meta-core) + type: 'set' | 'del' | 'add' | 'change' | 'delete' | 'rename' | 'batch' | 'reset' | 'plugin:complete'; + // New meta:events fields (flat key architecture) + key?: string; // Redis key that changed: file:abc123/property + ts?: string; // Timestamp + // Legacy file:events fields (for backward compatibility) path?: string; size?: string; midhash256?: string; // midhash256 CID computed by meta-core oldPath?: string; - // Legacy payload field (for backward compatibility) payload?: string; timestamp: string; } +/** + * Parsed meta:events key + * Extracts hashId and property from keys like 'file:abc123/property' + */ +export interface ParsedMetaEventKey { + hashId: string; + property: string; +} + /** * Stream consumer callback */ @@ -224,8 +242,108 @@ export class RedisClient implements Partial { return this.isConnected; } + // ======================================================================== + // Flat Key Methods (for meta-core flat key architecture) + // ======================================================================== + + /** + * Get a single property value from Redis using flat key format + * Key format: file:{hashId}/{property} + * + * @param hashId - The file hash ID + * @param property - The property name (can include / for nested props like 'titles/eng') + * @returns The property value or null if not found + */ + async getProperty(hashId: string, property: string): Promise { + if (!this.client || !this.isConnected) { + return null; + } + + try { + const key = `${this.config.prefix}file:${hashId}/${property}`; + return await this.client.get(key); + } catch (error: any) { + logger.error(`Failed to get property ${property} for ${hashId}:`, error.message); + return null; + } + } + + /** + * Get multiple properties for a file using flat keys + * More efficient than multiple getProperty calls using MGET + * + * @param hashId - The file hash ID + * @param properties - Array of property names to fetch + * @returns Map of property name to value (null values excluded) + */ + async getProperties(hashId: string, properties: string[]): Promise> { + const result = new Map(); + + if (!this.client || !this.isConnected || properties.length === 0) { + return result; + } + + try { + const keys = properties.map(prop => `${this.config.prefix}file:${hashId}/${prop}`); + const values = await this.client.mget(keys); + + for (let i = 0; i < properties.length; i++) { + const value = values[i]; + if (value !== null) { + result.set(properties[i], value); + } + } + } catch (error: any) { + logger.error(`Failed to get properties for ${hashId}:`, error.message); + } + + return result; + } + + /** + * Parse a meta:events key to extract hashId and property + * Key format: file:{hashId}/{property} + * + * @param key - The Redis key from the event + * @returns Parsed key with hashId and property, or null if invalid format + */ + parseMetaEventKey(key: string): ParsedMetaEventKey | null { + // Remove prefix if present + let keyWithoutPrefix = key; + if (this.config.prefix && key.startsWith(this.config.prefix)) { + keyWithoutPrefix = key.slice(this.config.prefix.length); + } + + // Key format: file:{hashId}/{property} + // Property can contain slashes (e.g., 'titles/eng') + const match = keyWithoutPrefix.match(/^file:([^/]+)\/(.+)$/); + if (!match) { + return null; + } + + return { + hashId: match[1], + property: match[2], + }; + } + + /** + * Check if a key is a file metadata key (vs other keys like indexes) + * + * @param key - The Redis key to check + * @returns True if this is a file metadata key + */ + isFileMetadataKey(key: string): boolean { + return this.parseMetaEventKey(key) !== null; + } + + // ======================================================================== + // Legacy HGETALL-based Methods (deprecated, kept for compatibility) + // ======================================================================== + /** * Get all file metadata from Redis + * @deprecated Use streaming state builder with flat keys instead */ async getAllFiles(): Promise> { const files = new Map(); @@ -288,6 +406,7 @@ export class RedisClient implements Partial { /** * Get file metadata by hash ID + * @deprecated Use streaming state builder with flat keys instead */ async getFileByHashId(hashId: string): Promise { if (!this.client || !this.isConnected) { @@ -672,11 +791,183 @@ export class RedisClient implements Partial { this.streamConsumerAbort = null; } + /** + * Read stream entries from a specific position using XREAD (no consumer groups) + * Used for streaming bootstrap - replays all events from the beginning + * + * @param stream - Stream name + * @param fromId - Start position ('0' for beginning, '$' for end/new only) + * @param count - Maximum entries to read per call + * @param blockMs - Block timeout (0 for non-blocking) + * @returns Array of [id, message] tuples, empty if none + */ + async readStreamFrom( + stream: string, + fromId: string, + count: number = 100, + blockMs: number = 0 + ): Promise> { + if (!this.client || !this.isConnected) { + return []; + } + + try { + let result: Array<[string, Array<[string, string[]]>]> | null; + + if (blockMs > 0) { + result = await this.client.call( + 'XREAD', + 'BLOCK', blockMs, + 'COUNT', count, + 'STREAMS', stream, + fromId + ) as typeof result; + } else { + result = await this.client.call( + 'XREAD', + 'COUNT', count, + 'STREAMS', stream, + fromId + ) as typeof result; + } + + if (!result || result.length === 0) { + return []; + } + + const entries: Array<{ id: string; message: StreamMessage }> = []; + + for (const [_streamName, streamEntries] of result) { + for (const [id, fields] of streamEntries) { + const message = this.parseStreamEntry(id, fields); + if (message) { + entries.push({ id, message }); + } + } + } + + return entries; + } catch (error: any) { + logger.error('Error reading stream:', error.message); + return []; + } + } + + /** + * Read all stream entries from a position, calling handler for each + * Returns the last processed entry ID for resumption + * + * @param stream - Stream name + * @param fromId - Start position ('0' for beginning) + * @param onMessage - Handler for each message + * @param batchSize - Entries per read + * @returns Last processed entry ID, or fromId if no entries + */ + async replayStream( + stream: string, + fromId: string, + onMessage: StreamMessageHandler, + batchSize: number = 100 + ): Promise { + let lastId = fromId; + let totalProcessed = 0; + const startTime = Date.now(); + + logger.info(`Replaying stream ${stream} from position ${fromId}...`); + + while (true) { + const entries = await this.readStreamFrom(stream, lastId, batchSize, 0); + + if (entries.length === 0) { + break; + } + + for (const { id, message } of entries) { + try { + await onMessage(message); + lastId = id; + totalProcessed++; + } catch (error: any) { + logger.error(`Error processing stream entry ${id}:`, error.message); + // Continue with next entry + } + } + + // If we got fewer than batchSize entries, we've reached the end + if (entries.length < batchSize) { + break; + } + } + + const elapsed = Date.now() - startTime; + logger.info(`Stream replay complete: ${totalProcessed} entries in ${elapsed}ms, last ID: ${lastId}`); + + return lastId; + } + + /** + * Start a stream consumer that reads from a specific position using XREAD + * Does not use consumer groups - for simpler replay/live consumption + * + * @param stream - Stream name + * @param fromId - Start position ('0' for beginning, or last ID from replayStream) + * @param onMessage - Handler for each message + * @param blockMs - Block timeout for polling (default: 5000) + */ + async startSimpleStreamConsumer( + stream: string, + fromId: string, + onMessage: StreamMessageHandler, + blockMs: number = 5000 + ): Promise { + if (!this.client) { + throw new Error('Redis not connected'); + } + + if (this.streamConsumerRunning) { + logger.warn('Stream consumer already running'); + return; + } + + this.streamConsumerRunning = true; + this.streamConsumerAbort = new AbortController(); + + let lastId = fromId; + logger.info(`Starting simple stream consumer: ${stream} from ${fromId}`); + + while (this.streamConsumerRunning && !this.streamConsumerAbort.signal.aborted) { + try { + const entries = await this.readStreamFrom(stream, lastId, 10, blockMs); + + for (const { id, message } of entries) { + try { + await onMessage(message); + lastId = id; + } catch (error: any) { + logger.error(`Error processing stream entry ${id}:`, error.message); + } + } + } catch (error: any) { + if (this.streamConsumerAbort?.signal.aborted) { + break; + } + logger.error('Stream consumer error:', error.message); + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + + logger.info('Simple stream consumer stopped'); + } + /** * Parse a stream entry into a StreamMessage + * + * Supports two formats: + * 1. New meta:events format: { type: 'set'|'del', key: 'file:abc123/prop', ts: '...' } + * 2. Legacy file:events format: { type: 'add'|'change'|..., path: '...', payload: '...' } */ private parseStreamEntry(id: string, fields: string[]): StreamMessage | null { - // Fields come as flat array: ['type', 'add', 'path', '...', 'timestamp', '123'] + // Fields come as flat array: ['type', 'set', 'key', 'file:abc123/prop', 'ts', '123'] const fieldMap: Record = {}; for (let i = 0; i < fields.length; i += 2) { fieldMap[fields[i]] = fields[i + 1]; @@ -687,6 +978,24 @@ export class RedisClient implements Partial { return null; } + // New meta:events format (set/del events with key field) + const metaEventTypes = ['set', 'del']; + if (metaEventTypes.includes(fieldMap.type)) { + if (!fieldMap.key) { + logger.warn(`Invalid stream entry ${id}: missing key for ${fieldMap.type} event`); + return null; + } + + return { + id, + type: fieldMap.type as 'set' | 'del', + key: fieldMap.key, + ts: fieldMap.ts, + timestamp: fieldMap.ts || fieldMap.timestamp || '0', + }; + } + + // Legacy file:events format // 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']; @@ -697,7 +1006,8 @@ export class RedisClient implements Partial { return null; } - if (!isDirectEvent && !fieldMap.payload) { + const legacyEventTypes = ['batch', 'reset', 'plugin:complete']; + if (legacyEventTypes.includes(fieldMap.type) && !fieldMap.payload) { logger.warn(`Invalid stream entry ${id}: missing payload for ${fieldMap.type} event`); return null; } diff --git a/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts b/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts new file mode 100644 index 0000000..c5162d8 --- /dev/null +++ b/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts @@ -0,0 +1,163 @@ +/** + * Rules Property Extractor + * + * Extracts the set of metadata properties that are relevant for VFS + * virtual path computation. This is used to filter which Redis key + * change events should trigger VFS updates. + * + * Properties are extracted from: + * 1. Rule templates: {title|originalTitle}, {season:pad2}, etc. + * 2. Rule conditions: { field: 'fileType', value: 'video' }, etc. + */ + +import { Logger } from 'tslog'; +import { TemplateEngine } from './template/TemplateEngine.js'; +import type { + RenamingConfig, + RenamingRule, + Condition, + ConditionGroup, +} from './types/RenamingRuleTypes.js'; +import { isConditionGroup } from './types/RenamingRuleTypes.js'; + +const logger = new Logger({ name: 'RulesPropertyExtractor' }); + +/** + * Core properties that are always required for VFS operation + * These are needed regardless of what rules are configured + */ +const CORE_PROPERTIES = new Set([ + 'filePath', // Source file location (must exist for file to appear in VFS) + 'size', // File size attribute + 'fileSize', // Alternative name for size + 'mtime', // Modification time + 'ctime', // Creation time + 'fileName', // Filename for fallback naming + 'extension', // File extension +]); + +/** + * Extract all property names used in a ConditionGroup (recursively) + */ +function extractConditionFields(conditionOrGroup: Condition | ConditionGroup): string[] { + if (isConditionGroup(conditionOrGroup)) { + const fields: string[] = []; + for (const condition of conditionOrGroup.conditions) { + fields.push(...extractConditionFields(condition)); + } + return fields; + } + + // It's a Condition + return [conditionOrGroup.field]; +} + +/** + * Extract all property names from a single rule + */ +function extractRuleProperties(rule: RenamingRule, templateEngine: TemplateEngine): string[] { + const properties: string[] = []; + + // Extract from template + const templateVars = templateEngine.extractVariables(rule.template); + properties.push(...templateVars); + + // Extract from conditions + const conditionFields = extractConditionFields(rule.conditions); + properties.push(...conditionFields); + + return properties; +} + +/** + * Normalize a property path for comparison + * Handles nested paths like 'titles/eng' vs 'titles.eng' + */ +function normalizePropertyPath(path: string): string { + // Convert slash-separated to dot-separated + return path.replace(/\//g, '.'); +} + +/** + * Check if a property path matches any of the VFS-relevant properties + * Handles both exact matches and prefix matches for nested properties + */ +export function isVfsRelevantProperty(property: string, vfsProperties: Set): boolean { + const normalized = normalizePropertyPath(property); + + // Direct match + if (vfsProperties.has(normalized)) { + return true; + } + + // Check if this is a nested property of a tracked property + // e.g., if we track 'titles', then 'titles.eng' or 'titles/eng' should match + for (const vfsProp of vfsProperties) { + const normalizedVfsProp = normalizePropertyPath(vfsProp); + + // Check if the property starts with a tracked path + if (normalized.startsWith(normalizedVfsProp + '.')) { + return true; + } + + // Check if a tracked path starts with this property + // e.g., if we track 'titles.eng' and get update for 'titles' + if (normalizedVfsProp.startsWith(normalized + '.')) { + return true; + } + } + + return false; +} + +/** + * RulesPropertyExtractor class + * + * Analyzes renaming rules configuration and extracts the set of + * metadata properties that are needed for VFS virtual path computation. + */ +export class RulesPropertyExtractor { + private templateEngine: TemplateEngine; + + constructor() { + this.templateEngine = new TemplateEngine(); + } + + /** + * Extract all VFS-relevant properties from a RenamingConfig + * + * @param config The renaming configuration to analyze + * @returns Set of property names that affect VFS virtual paths + */ + extractVfsProperties(config: RenamingConfig): Set { + const properties = new Set(CORE_PROPERTIES); + + // Extract from all rules + for (const rule of config.rules) { + if (!rule.enabled) continue; + + const ruleProps = extractRuleProperties(rule, this.templateEngine); + for (const prop of ruleProps) { + properties.add(normalizePropertyPath(prop)); + } + } + + // Extract from default rule if present + if (config.defaultRule) { + const defaultProps = extractRuleProperties(config.defaultRule, this.templateEngine); + for (const prop of defaultProps) { + properties.add(normalizePropertyPath(prop)); + } + } + + logger.debug(`Extracted ${properties.size} VFS-relevant properties from rules config`); + return properties; + } + + /** + * Get the core properties that are always required + */ + getCoreProperties(): Set { + return new Set(CORE_PROPERTIES); + } +} diff --git a/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts b/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts new file mode 100644 index 0000000..2b45b40 --- /dev/null +++ b/packages/meta-fuse-core/src/vfs/StreamingStateBuilder.ts @@ -0,0 +1,338 @@ +/** + * Streaming State Builder + * + * Manages file metadata state by processing events from the meta:events stream. + * Each event triggers a GET for the changed property, building state incrementally. + * + * This replaces the batch HGETALL/SCAN approach with a streaming model: + * 1. On startup: Replay meta:events from position 0 to rebuild state + * 2. Live: Continue consuming new events for real-time updates + */ + +import { Logger } from 'tslog'; +import { RedisClient, StreamMessage, ParsedMetaEventKey } from '../kv/RedisClient.js'; +import { RulesPropertyExtractor, isVfsRelevantProperty } from './RulesPropertyExtractor.js'; +import type { RenamingConfig } from './types/RenamingRuleTypes.js'; + +const logger = new Logger({ name: 'StreamingStateBuilder' }); + +/** + * File metadata state - map of property name to value + */ +export type FileState = Map; + +/** + * All files state - map of hashId to FileState + */ +export type FilesState = Map; + +/** + * Callback for VFS updates + */ +export interface VFSUpdateCallback { + onPropertyChange(hashId: string, property: string, value: string): void; + onPropertyDelete(hashId: string, property: string): void; + onFileDelete(hashId: string): void; + onFileComplete(hashId: string, metadata: Record): void; +} + +/** + * Configuration for StreamingStateBuilder + */ +export interface StreamingStateBuilderConfig { + /** Redis client for fetching property values */ + redisClient: RedisClient; + /** Renaming rules config for extracting VFS-relevant properties */ + rulesConfig: RenamingConfig; + /** VFS update callback */ + vfsCallback?: VFSUpdateCallback; + /** FILES_VOLUME path prefix */ + filesPath?: string; +} + +/** + * StreamingStateBuilder class + * + * Builds and maintains file metadata state by processing stream events. + * Only fetches and stores VFS-relevant properties to minimize memory usage. + */ +export class StreamingStateBuilder { + private state: FilesState = new Map(); + private vfsRelevantProps: Set; + private redisClient: RedisClient; + private vfsCallback?: VFSUpdateCallback; + private filesPath: string; + private propertyExtractor: RulesPropertyExtractor; + + // Stats + private stats = { + eventsProcessed: 0, + propertiesFetched: 0, + propertiesSkipped: 0, + filesComplete: 0, + lastEventId: '0', + }; + + constructor(config: StreamingStateBuilderConfig) { + this.redisClient = config.redisClient; + this.vfsCallback = config.vfsCallback; + this.filesPath = config.filesPath ?? '/files'; + + // Extract VFS-relevant properties from rules + this.propertyExtractor = new RulesPropertyExtractor(); + this.vfsRelevantProps = this.propertyExtractor.extractVfsProperties(config.rulesConfig); + + logger.info(`StreamingStateBuilder initialized with ${this.vfsRelevantProps.size} VFS-relevant properties`); + logger.debug('VFS-relevant properties:', Array.from(this.vfsRelevantProps).join(', ')); + } + + /** + * Set the VFS update callback + */ + setVFSCallback(callback: VFSUpdateCallback): void { + this.vfsCallback = callback; + } + + /** + * Update rules config and re-extract VFS-relevant properties + */ + updateRulesConfig(rulesConfig: RenamingConfig): void { + this.vfsRelevantProps = this.propertyExtractor.extractVfsProperties(rulesConfig); + logger.info(`Updated VFS-relevant properties: ${this.vfsRelevantProps.size}`); + } + + /** + * Process a stream message event + * This is the main entry point for event processing + */ + async processEvent(message: StreamMessage): Promise { + this.stats.eventsProcessed++; + this.stats.lastEventId = message.id; + + // Handle new meta:events format (set/del with key field) + if ((message.type === 'set' || message.type === 'del') && message.key) { + await this.processMetaEvent(message); + return; + } + + // Ignore legacy event types - they're not supported in flat key architecture + // Legacy events used path/payload format which doesn't apply to flat keys + logger.debug(`Ignoring legacy event type: ${message.type}`); + } + + /** + * Process a meta:events format event (set/del with key) + */ + private async processMetaEvent(message: StreamMessage): Promise { + const parsed = this.redisClient.parseMetaEventKey(message.key!); + if (!parsed) { + // Not a file metadata key (could be index or other key) + return; + } + + const { hashId, property } = parsed; + + if (message.type === 'del') { + this.handlePropertyDelete(hashId, property); + return; + } + + // message.type === 'set' + await this.handlePropertySet(hashId, property); + } + + /** + * Handle a property set event + * Fetches the property value if VFS-relevant and updates state + */ + private async handlePropertySet(hashId: string, property: string): Promise { + // Check if this property is VFS-relevant + if (!isVfsRelevantProperty(property, this.vfsRelevantProps)) { + this.stats.propertiesSkipped++; + return; + } + + // Fetch the property value from Redis + const value = await this.redisClient.getProperty(hashId, property); + if (value === null) { + // Property was set then deleted before we could fetch it + // Or it's a different data type (shouldn't happen in flat key architecture) + return; + } + + this.stats.propertiesFetched++; + + // Check if file was complete BEFORE updating state + const wasComplete = this.isFileComplete(hashId); + + // Update internal state + this.updateProperty(hashId, property, value); + + // Notify VFS of changes + if (this.vfsCallback) { + this.vfsCallback.onPropertyChange(hashId, property, value); + + // If file just became complete (has filePath now), notify VFS + if (!wasComplete && this.isFileComplete(hashId)) { + this.stats.filesComplete++; + const metadata = this.getFileMetadata(hashId); + if (metadata) { + this.vfsCallback.onFileComplete(hashId, metadata); + } + } else if (wasComplete) { + // File was already complete, but properties changed - update VFS + // This handles cases where title, season, etc. arrive after filePath + const metadata = this.getFileMetadata(hashId); + if (metadata) { + this.vfsCallback.onFileComplete(hashId, metadata); + } + } + } + } + + /** + * Handle a property delete event + */ + private handlePropertyDelete(hashId: string, property: string): void { + const fileState = this.state.get(hashId); + if (!fileState) { + return; + } + + // Remove the property + fileState.delete(property); + + // If this was the filePath, the file is no longer in VFS + if (property === 'filePath') { + this.state.delete(hashId); + if (this.vfsCallback) { + this.vfsCallback.onFileDelete(hashId); + } + return; + } + + // If file has no more properties, remove it + if (fileState.size === 0) { + this.state.delete(hashId); + if (this.vfsCallback) { + this.vfsCallback.onFileDelete(hashId); + } + return; + } + + // Notify VFS of property deletion + if (this.vfsCallback) { + this.vfsCallback.onPropertyDelete(hashId, property); + } + } + + /** + * Update a property in the state + */ + private updateProperty(hashId: string, property: string, value: string): void { + let fileState = this.state.get(hashId); + if (!fileState) { + fileState = new Map(); + this.state.set(hashId, fileState); + } + fileState.set(property, value); + } + + /** + * Check if a file has filePath (required to appear in VFS) + */ + private isFileComplete(hashId: string): boolean { + const fileState = this.state.get(hashId); + return fileState?.has('filePath') ?? false; + } + + /** + * Get metadata for a single file + */ + getFileMetadata(hashId: string): Record | null { + const fileState = this.state.get(hashId); + if (!fileState) { + return null; + } + return Object.fromEntries(fileState); + } + + /** + * Get all files with their metadata + * Only returns files that are "complete" (have filePath) + */ + getAllFiles(): Map> { + const result = new Map>(); + + for (const [hashId, fileState] of this.state) { + if (fileState.has('filePath')) { + result.set(hashId, Object.fromEntries(fileState)); + } + } + + return result; + } + + /** + * Get file count (complete files only) + */ + getFileCount(): number { + let count = 0; + for (const fileState of this.state.values()) { + if (fileState.has('filePath')) { + count++; + } + } + return count; + } + + /** + * Get statistics + */ + getStats(): typeof this.stats & { fileCount: number; stateSize: number } { + return { + ...this.stats, + fileCount: this.getFileCount(), + stateSize: this.state.size, + }; + } + + /** + * Clear all state (for reset) + */ + clear(): void { + this.state.clear(); + this.stats = { + eventsProcessed: 0, + propertiesFetched: 0, + propertiesSkipped: 0, + filesComplete: 0, + lastEventId: '0', + }; + logger.info('State cleared'); + } + + /** + * Get the last processed event ID + */ + getLastEventId(): string { + return this.stats.lastEventId; + } + + /** + * Resolve source path from filePath property + * Prepends FILES_VOLUME if needed + */ + resolveSourcePath(filePath: string): string { + if (filePath.startsWith('/')) { + // Already absolute + if (filePath.startsWith(this.filesPath)) { + return filePath; + } + // Absolute but not in filesPath - prepend + return `${this.filesPath}${filePath}`; + } + // Relative path - prepend filesPath + return `${this.filesPath}/${filePath}`; + } +} diff --git a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts index 87fc894..8826d3c 100644 --- a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts +++ b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts @@ -15,6 +15,7 @@ import { TemplateEngine } from './template/TemplateEngine.js'; import { ConditionEvaluator } from './template/ConditionEvaluator.js'; import { ConfigStorage } from '../config/ConfigStorage.js'; import type { RenamingConfig, RenamingRule, PreviewItem, PreviewResponse } from './types/RenamingRuleTypes.js'; +import type { VFSUpdateCallback } from './StreamingStateBuilder.js'; const logger = new Logger({ name: 'VirtualFileSystem' }); @@ -58,9 +59,10 @@ export interface VFSConfig { filesPath?: string; // Local files path prefix to strip when building WebDAV URLs } -export class VirtualFileSystem { +export class VirtualFileSystem implements VFSUpdateCallback { private nodes: Map = new Map(); private sourcePathToVirtualPath: Map = new Map(); + private hashIdToVirtualPath: Map = new Map(); // Track hashId -> virtualPath for streaming updates private redisClient: RedisClient; private configStorage: ConfigStorage; private metaDataToFolderStruct: MetaDataToFolderStruct; @@ -69,6 +71,7 @@ export class VirtualFileSystem { private rulesConfig: RenamingConfig | null = null; private config: Required; private lastRefresh: Date | null = null; + private useStreamingMode: boolean = false; // Flag to indicate streaming mode private cachedStats = { fileCount: 0, @@ -106,6 +109,7 @@ export class VirtualFileSystem { private initRoot(): void { this.nodes.clear(); this.sourcePathToVirtualPath.clear(); + this.hashIdToVirtualPath.clear(); this.nodes.set('/', { type: 'directory', name: '', @@ -115,17 +119,125 @@ export class VirtualFileSystem { this.cachedStats = { fileCount: 0, directoryCount: 1, totalSize: 0 }; } + /** + * Enable streaming mode + * In streaming mode, VFS is populated via event callbacks instead of refresh() + */ + enableStreamingMode(): void { + this.useStreamingMode = true; + logger.info('VFS streaming mode enabled'); + } + + /** + * Convert flat properties map to FileMetadata + * Used when processing streaming events + * + * Note: During migration, not all metadata may be in flat key format yet. + * We extract what we can from filePath as a fallback. + */ + private flatPropertiesToMetadata(props: Record, hashId: string): FileMetadata { + // Get the original path (relative to FILES_VOLUME) + const originalPath = props.filePath ?? ''; + + // Resolve full source path by prepending FILES_VOLUME + let sourcePath = originalPath; + if (originalPath && !originalPath.startsWith('/')) { + sourcePath = `${this.config.filesPath}/${originalPath}`; + } else if (!originalPath.startsWith(this.config.filesPath) && originalPath.startsWith('/')) { + sourcePath = `${this.config.filesPath}${originalPath}`; + } + + // Extract filename from path as fallback + const pathFileName = sourcePath ? path.basename(sourcePath) : undefined; + const fileName = props.fileName ?? pathFileName; + + // Extract extension from filename + const extension = props.extension ?? (pathFileName ? path.extname(pathFileName).slice(1) : undefined); + + // Parse size + const size = parseInt(props.fileSize ?? props.size ?? props.sizeByte ?? '0', 10); + + // Parse timestamps + let mtime = 0; + if (props.mtime) { + const parsed = Date.parse(props.mtime); + mtime = isNaN(parsed) ? parseFloat(props.mtime) : parsed; + } + + // Parse titles - can be from nested paths like 'titles/eng' + let titles: { eng?: string; [key: string]: string | undefined } | undefined; + if (props['titles/eng']) { + titles = { eng: props['titles/eng'] }; + } + + // Parse season/episode + const season = props.season !== undefined ? parseInt(props.season, 10) : undefined; + const episode = props.episode !== undefined ? parseInt(props.episode, 10) : undefined; + + // Parse year + const movieYear = props.movieYear ? parseInt(props.movieYear, 10) : undefined; + const year = props.year ? parseInt(props.year, 10) : movieYear; + + // Infer fileType from extension if not provided + let fileType = props.fileType; + if (!fileType && extension) { + const videoExts = ['mkv', 'mp4', 'avi', 'mov', 'wmv', 'flv', 'webm', 'm4v']; + const subtitleExts = ['srt', 'ass', 'ssa', 'sub', 'idx', 'vtt']; + const torrentExts = ['torrent']; + + if (videoExts.includes(extension.toLowerCase())) { + fileType = 'video'; + } else if (subtitleExts.includes(extension.toLowerCase())) { + fileType = 'subtitle'; + } else if (torrentExts.includes(extension.toLowerCase())) { + fileType = 'torrent'; + } + } + + return { + sourcePath, + originalPath, + size, + mtime, + ctime: props.ctime ? parseFloat(props.ctime) : mtime, + hashId, + title: props.title, + titles, + originalTitle: props.originalTitle, + fileName, + season, + episode, + extra: props.extra === 'true', + movieYear, + year, + fileType, + extension, + version: props.version, + subtitleLanguage: props.subtitleLanguage, + }; + } + /** * Start the VFS (connect to Redis and load data) - * Initial bootstrap only - updates come via Redis Streams + * + * In legacy mode: Does initial load from Redis using HGETALL/SCAN + * In streaming mode: Just initializes, data comes via StreamingStateBuilder events */ async start(): Promise { logger.info('Starting VirtualFileSystem'); - // Initial load from Redis (bootstrap) - await this.refresh(); + // Load rules configuration + this.loadRulesConfig(); - logger.info('VirtualFileSystem started (updates via Redis Streams)'); + if (this.useStreamingMode) { + // In streaming mode, VFS is populated via event callbacks + // StreamingStateBuilder will call onFileComplete for each file + logger.info('VirtualFileSystem started in streaming mode (data via events)'); + } else { + // Legacy mode: Initial load from Redis (bootstrap) + await this.refresh(); + logger.info('VirtualFileSystem started in legacy mode (updates via Redis Streams)'); + } } /** @@ -903,4 +1015,151 @@ export class VirtualFileSystem { return null; } + + // ============================================ + // VFSUpdateCallback Implementation + // (for streaming state builder integration) + // ============================================ + + /** + * Handle a property change event from the streaming state builder + * This is called for each VFS-relevant property that changes + * + * In streaming mode, we don't immediately update the VFS on every property change. + * Instead, we wait for onFileComplete which has the full metadata. + */ + onPropertyChange(hashId: string, property: string, value: string): void { + // In streaming mode, individual property changes don't trigger VFS updates + // The VFS is updated when onFileComplete is called with full metadata + logger.debug(`Property changed for ${hashId}: ${property}=${value.substring(0, 50)}...`); + } + + /** + * Handle a property delete event from the streaming state builder + */ + onPropertyDelete(hashId: string, property: string): void { + logger.debug(`Property deleted for ${hashId}: ${property}`); + + // If filePath was deleted, the file should be removed from VFS + // This is handled by onFileDelete, so we just log here + } + + /** + * Handle file deletion from the streaming state builder + * Called when filePath is deleted or all properties are removed + */ + onFileDelete(hashId: string): void { + const virtualPath = this.hashIdToVirtualPath.get(hashId); + + if (virtualPath) { + this.removeFile(virtualPath); + this.hashIdToVirtualPath.delete(hashId); + logger.debug(`File deleted from VFS: ${hashId} -> ${virtualPath}`); + } + } + + /** + * Handle file complete event from the streaming state builder + * Called when a file has filePath and enough metadata to compute virtual path + * + * @param hashId - The file hash ID + * @param metadata - Flat properties map from the streaming state builder + */ + onFileComplete(hashId: string, metadata: Record): void { + // Load rules if not loaded + if (!this.rulesConfig) { + this.loadRulesConfig(); + } + + // Convert flat properties to FileMetadata + const fileMetadata = this.flatPropertiesToMetadata(metadata, hashId); + + if (!fileMetadata.sourcePath) { + logger.warn(`onFileComplete: No sourcePath for ${hashId}`); + return; + } + + // Compute virtual path + const virtualPath = this.computeVirtualPath(fileMetadata, fileMetadata.sourcePath); + + if (!virtualPath) { + logger.debug(`Could not compute virtual path for: ${fileMetadata.sourcePath}`); + return; + } + + // If file already exists at a different virtual path, remove old entry + const oldVirtualPath = this.hashIdToVirtualPath.get(hashId); + if (oldVirtualPath && oldVirtualPath !== virtualPath) { + this.removeFile(oldVirtualPath); + } + + // Add/update file in VFS + this.addFileSync(virtualPath, fileMetadata); + this.sourcePathToVirtualPath.set(fileMetadata.sourcePath, virtualPath); + this.hashIdToVirtualPath.set(hashId, virtualPath); + + logger.debug(`File complete: ${hashId} -> ${virtualPath}`); + this.lastRefresh = new Date(); + } + + /** + * Synchronous version of addFile for streaming updates + * Used by onFileComplete which is called synchronously from event processing + */ + private addFileSync(virtualPath: string, metadata: FileMetadata): void { + virtualPath = this.normalizePath(virtualPath); + + // Skip if file already exists at this path + if (this.nodes.has(virtualPath) && this.nodes.get(virtualPath)!.type === 'file') { + // Update existing file + const node = this.nodes.get(virtualPath)!; + const oldSize = node.size ?? 0; + this.cachedStats.totalSize -= oldSize; + this.cachedStats.fileCount--; + } + + // Create parent directories + const dirname = path.dirname(virtualPath); + this.ensureDirectory(dirname); + + // Get file stats from metadata + const size = metadata.size || 0; + const mtime = new Date(metadata.mtime || Date.now()); + const ctime = new Date(metadata.ctime || metadata.mtime || Date.now()); + + // Add file node + const filename = path.basename(virtualPath); + this.nodes.set(virtualPath, { + type: 'file', + name: filename, + parent: dirname, + sourcePath: metadata.sourcePath, + size, + mtime, + ctime, + metadata, + }); + + // Add to parent's children + const parentNode = this.nodes.get(dirname); + if (parentNode && parentNode.type === 'directory') { + parentNode.children!.add(filename); + } + + // Update stats + this.cachedStats.fileCount++; + this.cachedStats.totalSize += size; + } + + /** + * Update a file in VFS when its metadata changes + * Called when streaming events indicate a property change for an existing file + * + * @param hashId - The file hash ID + * @param metadata - Updated metadata properties + */ + updateFileFromStreaming(hashId: string, metadata: Record): void { + // Just delegate to onFileComplete which handles both add and update + this.onFileComplete(hashId, metadata); + } }