diff --git a/docker/nginx.conf b/docker/nginx.conf index e962ec1..89acaef 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -35,31 +35,6 @@ http { listen 80; server_name _; - # Welcome page at root - location = / { - root /app/welcome; - try_files /welcome.html =404; - } - - # Redirect /stats to /stats/ - location = /stats { - return 301 /stats/; - } - - # VFS Statistics dashboard (React SPA) - location /stats/ { - alias /app/ui/; - index index.html; - try_files $uri $uri/ /stats/index.html; - } - - # Static assets for stats dashboard - location /assets/ { - alias /app/ui/assets/; - expires 1y; - add_header Cache-Control "public, immutable"; - } - # API routes location /api/ { proxy_pass http://api; @@ -97,5 +72,19 @@ http { proxy_send_timeout 300; proxy_read_timeout 300; } + + # Static assets + location /assets/ { + alias /app/ui/assets/; + expires 1y; + add_header Cache-Control "public, immutable"; + } + + # React SPA at root (catch-all) + location / { + alias /app/ui/; + index index.html; + try_files $uri $uri/ /index.html; + } } } diff --git a/docker/welcome.html b/docker/welcome.html deleted file mode 100644 index 0174602..0000000 --- a/docker/welcome.html +++ /dev/null @@ -1,518 +0,0 @@ - - - - - - meta-fuse - Virtual Filesystem - - - - -
- - - -
-

meta-fuse

-

Virtual Filesystem Service

-
-
-
- Checking... -
-
-
- -
-

What is meta-fuse?

-

- meta-fuse is a virtual filesystem that presents your media files - in an organized structure based on metadata. It reads file information from the - KV leader (via Redis) and creates a virtual view of your library without duplicating any files. -

- -
- -
- -
📁
-

WebDAV Files

-

Browse and access your organized media library. Mount as a network drive on any device.

- /webdav - - - -
📊
-

VFS Statistics

-

View file counts, directory structure, connection status, and refresh metadata.

- /stats - -
- -
-
-

⚙️ Setup & Configuration

- ▼ -
-
-
-
-

🗂️ FUSE Mount (Docker)

-

Mount the virtual filesystem directly from the container. Add this volume to your docker-compose.yml to access the organized files on your host:

-
- volumes:
  - /path/on/host:/mnt/virtual:rw,shared -
-

The virtual filesystem is available inside the container at /mnt/virtual

-
- -
-

🌐 WebDAV Network Drive

-

Mount the virtual filesystem as a network drive on any device:

-
- Windows: - net use Z: http://<host>/webdav -
-
- macOS: - Finder → Go → Connect to Server → http://<host>/webdav -
-
- Linux: - sudo mount -t davfs http://<host>/webdav /mnt/vfs -
-
-
-
-
- - -
- - - - diff --git a/packages/meta-fuse-core/src/api/APIServer.ts b/packages/meta-fuse-core/src/api/APIServer.ts index 14aa6a2..76d582d 100644 --- a/packages/meta-fuse-core/src/api/APIServer.ts +++ b/packages/meta-fuse-core/src/api/APIServer.ts @@ -267,12 +267,17 @@ export class APIServer { url: string; api: string; status: string; - capabilities: string[]; - version: string; + role?: string; } const services: ServiceResponse[] = []; - let leaderInfo: { host: string; api: string; http: string } | null = null; + let leaderInfo: { + hostname: string; + baseUrl: string; + apiUrl: string; + redisUrl: string; + webdavUrl: string; + } | null = null; try { const serviceDiscovery = this.kvManager?.getServiceDiscovery(); @@ -280,17 +285,12 @@ export class APIServer { const allServices = await serviceDiscovery.discoverAllServices(); for (const svc of allServices) { - // Build dashboard URL from API URL - const apiUrl = svc.api || ''; - const dashboardPath = svc.endpoints?.dashboard || '/'; - services.push({ name: svc.name || 'Unknown', - url: apiUrl + dashboardPath, - api: apiUrl, + url: svc.baseUrl || '', + api: svc.baseUrl || '', status: svc.status || 'unknown', - capabilities: svc.capabilities || [], - version: svc.version || '', + role: svc.role, }); } } diff --git a/packages/meta-fuse-core/src/index.ts b/packages/meta-fuse-core/src/index.ts index 8b85a40..f491aa1 100644 --- a/packages/meta-fuse-core/src/index.ts +++ b/packages/meta-fuse-core/src/index.ts @@ -3,38 +3,35 @@ * * Standalone FUSE API server that reads metadata from Redis, * computes virtual paths, and exposes organized folder structure. + * + * Uses Redis Streams for reliable event delivery from meta-sort. */ import 'dotenv/config'; import { Logger } from 'tslog'; import { KVManager } from './kv/KVManager.js'; -import { RedisClient } from './kv/RedisClient.js'; +import { RedisClient, StreamMessage } from './kv/RedisClient.js'; import { VirtualFileSystem } from './vfs/VirtualFileSystem.js'; import { APIServer } from './api/APIServer.js'; const logger = new Logger({ name: 'meta-fuse' }); -// Pub/sub channel for batched updates from meta-sort -const UPDATE_CHANNEL = 'meta-sort:file:batch'; - -// Pub/sub channel for scan reset notification from meta-sort -const RESET_CHANNEL = 'meta-sort:scan:reset'; - -// Pub/sub channel for plugin completion events from meta-sort -const PLUGIN_COMPLETE_CHANNEL = 'meta-sort:plugin:complete'; +// Redis Streams for reliable event delivery from meta-sort +const EVENTS_STREAM = 'meta-sort:events'; +const CONSUMER_GROUP = 'meta-fuse-vfs'; // Plugins that provide metadata needed for VFS virtual paths // When these complete, we should update the VFS entry for that file const VFS_RELEVANT_PLUGINS = new Set(['filename-parser', 'tmdb', 'jellyfin-nfo']); -interface PluginCompleteMessage { +interface PluginCompletePayload { fileHash: string; pluginId: string; filePath: string; timestamp: number; } -interface BatchUpdateMessage { +interface BatchUpdatePayload { timestamp: number; changes: Array<{ action: 'add' | 'update' | 'remove'; @@ -42,6 +39,57 @@ interface BatchUpdateMessage { }>; } +interface ResetPayload { + timestamp: number; + action: 'reset'; +} + +/** + * Handle a stream message from Redis Streams + * Routes messages by type to appropriate handlers + */ +async function handleStreamMessage(message: StreamMessage, vfs: VirtualFileSystem): Promise { + try { + switch (message.type) { + case 'batch': { + 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': { + const reset: ResetPayload = JSON.parse(message.payload); + logger.info(`Processing reset event: ${reset.action}`); + + vfs.reset(); + logger.info('VFS reset complete, awaiting new file updates'); + break; + } + + case 'plugin:complete': { + 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}`); + } + } catch (error: any) { + logger.error(`Error processing stream message ${message.id}:`, error.message); + throw error; // Re-throw to prevent ACK + } +} + async function main(): Promise { logger.info('Starting meta-fuse...'); logger.info(`Node.js ${process.version}`); @@ -54,19 +102,11 @@ async function main(): Promise { redisPrefix: process.env.REDIS_PREFIX ?? '', apiPort: parseInt(process.env.API_PORT ?? '3000', 10), apiHost: process.env.API_HOST ?? '0.0.0.0', - vfsRefreshInterval: parseInt(process.env.VFS_REFRESH_INTERVAL ?? '30000', 10), baseUrl: process.env.BASE_URL, serviceVersion: process.env.SERVICE_VERSION ?? '1.0.0', - // WebDAV URL for meta-core file access (e.g., http://meta-core/webdav) - metaCoreWebdavUrl: process.env.META_CORE_WEBDAV_URL, }; logger.info(`Config: META_CORE_PATH=${config.metaCorePath}, FILES_VOLUME=${config.filesPath}`); - if (config.metaCoreWebdavUrl) { - logger.info(`WebDAV: Using meta-core WebDAV at ${config.metaCoreWebdavUrl}`); - } else { - logger.info(`WebDAV: Not configured, FUSE driver will use local filesystem`); - } // Initialize KV Manager (handles leader discovery) const kvManager = new KVManager({ @@ -75,6 +115,7 @@ async function main(): Promise { serviceName: 'meta-fuse', apiPort: config.apiPort, baseUrl: config.baseUrl, + redisPrefix: config.redisPrefix, }); let redisClient: RedisClient | null = null; @@ -86,6 +127,10 @@ async function main(): Promise { logger.info(`Received ${signal}, shutting down...`); try { + // Stop stream consumer first + if (redisClient) { + redisClient.stopStreamConsumer(); + } if (apiServer) await apiServer.stop(); if (vfs) await vfs.stop(); await kvManager.stop(); @@ -113,69 +158,58 @@ async function main(): Promise { return; } - // Initialize VFS + // Get WebDAV URL from leader info (discovered via service discovery) + // Use hostname to construct internal URL since webdavUrl uses external baseUrl + const leaderInfo = kvManager.getLeaderInfo(); + let webdavUrl: string | null = null; + if (leaderInfo?.hostname) { + // Construct internal WebDAV URL using leader hostname (nginx on port 80) + webdavUrl = `http://${leaderInfo.hostname}/webdav`; + logger.info(`WebDAV: Using ${webdavUrl} (from service discovery, leader: ${leaderInfo.hostname})`); + } else { + logger.info('WebDAV: Not available from leader, FUSE driver will use local filesystem'); + } + + // Initialize VFS (no periodic refresh - uses Redis Streams) vfs = new VirtualFileSystem(redisClient, { fileMode: parseInt(process.env.FUSE_FILE_MODE ?? '644', 8), directoryMode: parseInt(process.env.FUSE_DIR_MODE ?? '755', 8), uid: parseInt(process.env.PUID ?? '1000', 10), gid: parseInt(process.env.PGID ?? '1000', 10), - refreshInterval: config.vfsRefreshInterval, configDir: process.env.CONFIG_DIR ?? '/meta-fuse/config', - webdavBaseUrl: config.metaCoreWebdavUrl, + webdavBaseUrl: webdavUrl, filesPath: config.filesPath, }); - // Start VFS + // Start VFS (initial load from Redis) logger.info('Starting VirtualFileSystem...'); await vfs.start(); - // Subscribe to batch updates from meta-sort - logger.info(`Subscribing to ${UPDATE_CHANNEL}...`); - await redisClient.subscribe(UPDATE_CHANNEL, async (message: string) => { - try { - const batch: BatchUpdateMessage = JSON.parse(message); - logger.debug(`Received batch update: ${batch.changes.length} changes`); + // Initialize Redis Streams consumer + logger.info(`Initializing stream consumer for ${EVENTS_STREAM}...`); + await redisClient.initStreamConsumer(EVENTS_STREAM, CONSUMER_GROUP); - // Process each change incrementally - for (const change of batch.changes) { - await vfs!.onFileUpdate(change.hashId, change.action); - } - } catch (error: any) { - logger.error(`Error processing batch update: ${error.message}`); + // Process any pending entries from crashed consumers + await redisClient.processPendingEntries( + EVENTS_STREAM, + CONSUMER_GROUP, + 30000, // 30 second idle threshold + async (message: StreamMessage) => { + await handleStreamMessage(message, vfs!); } - }); + ); - // Subscribe to scan reset events from meta-sort - logger.info(`Subscribing to ${RESET_CHANNEL}...`); - await redisClient.subscribe(RESET_CHANNEL, async (message: string) => { - try { - const event = JSON.parse(message); - logger.info(`Received scan reset event: ${event.action}`); - - // Reset VFS for fresh scan - will be rebuilt incrementally - vfs!.reset(); - - logger.info('VFS reset complete, awaiting new file updates from meta-sort'); - } catch (error: any) { - logger.error(`Error processing reset event: ${error.message}`); - } - }); - - // Subscribe to plugin completion events from meta-sort - // When relevant plugins complete (filename-parser, tmdb, etc.), update VFS - logger.info(`Subscribing to ${PLUGIN_COMPLETE_CHANNEL}...`); - await redisClient.subscribe(PLUGIN_COMPLETE_CHANNEL, async (message: string) => { - try { - const event: PluginCompleteMessage = JSON.parse(message); - - // 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'); - } - } catch (error: any) { - logger.error(`Error processing plugin complete event: ${error.message}`); - } + // Start stream consumer in background + logger.info('Starting stream consumer...'); + redisClient.startStreamConsumer( + EVENTS_STREAM, + CONSUMER_GROUP, + async (message: StreamMessage) => { + await handleStreamMessage(message, vfs!); + }, + 5000 // 5 second block timeout + ).catch(error => { + logger.error('Stream consumer error:', error); }); // Initialize API server (with kvManager for service discovery) diff --git a/packages/meta-fuse-core/src/kv/IKVClient.ts b/packages/meta-fuse-core/src/kv/IKVClient.ts index db697ae..9a7944b 100644 --- a/packages/meta-fuse-core/src/kv/IKVClient.ts +++ b/packages/meta-fuse-core/src/kv/IKVClient.ts @@ -103,4 +103,7 @@ export interface ServiceInfo { /** Last heartbeat timestamp (ISO string) */ lastHeartbeat: string; + + /** Role for meta-core instances: 'leader', 'follower', or undefined for other services */ + role?: string; } diff --git a/packages/meta-fuse-core/src/kv/KVManager.ts b/packages/meta-fuse-core/src/kv/KVManager.ts index 1c9f6b6..2fae729 100644 --- a/packages/meta-fuse-core/src/kv/KVManager.ts +++ b/packages/meta-fuse-core/src/kv/KVManager.ts @@ -80,15 +80,14 @@ export class KVManager { logger.info(`Starting KVManager for ${this.config.serviceName}...`); - // Initialize service discovery (meta-core handles actual registration) - // We keep this for service discovery (reading other services) + // Initialize and start service discovery for registration and heartbeat const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`; this.serviceDiscovery = new ServiceDiscovery({ metaCorePath: this.config.metaCorePath, serviceName: this.config.serviceName, apiUrl }); - // Note: We don't start ServiceDiscovery - meta-core handles registration + await this.serviceDiscovery.start(); // If direct Redis URL provided, skip leader discovery if (this.config.redisUrl) { diff --git a/packages/meta-fuse-core/src/kv/LeaderClient.ts b/packages/meta-fuse-core/src/kv/LeaderClient.ts index 86bfc5a..50c8b1b 100644 --- a/packages/meta-fuse-core/src/kv/LeaderClient.ts +++ b/packages/meta-fuse-core/src/kv/LeaderClient.ts @@ -42,18 +42,87 @@ export class LeaderClient { private watcher: FSWatcher | null = null; private onChangeCallbacks: (() => void)[] = []; + // URL caching + private cachedUrls: URLsResponse | null = null; + private urlsCacheTime: number = 0; + private readonly urlsCacheTTL: number = 5000; // 5 seconds + constructor(config: LeaderClientConfig) { this.config = config; this.infoFilePath = `${config.metaCorePath}/locks/kv-leader.info`; } /** - * Read leader info from the info file + * Read API URL from file (plain text format) + */ + private async getApiUrlFromFile(): Promise { + try { + const content = await fs.readFile(this.infoFilePath, 'utf-8'); + return content.trim() || null; + } catch (error: any) { + logger.error(`Failed to read API URL from file: ${error.message}`); + return null; + } + } + + /** + * Fetch URLs from meta-core /urls API with caching + */ + private async fetchUrls(apiUrl: string): Promise { + // Check cache + const now = Date.now(); + if (this.cachedUrls && (now - this.urlsCacheTime) < this.urlsCacheTTL) { + return this.cachedUrls; + } + + try { + const response = await fetch(`${apiUrl}/urls`, { + method: 'GET', + headers: { 'Accept': 'application/json' }, + signal: AbortSignal.timeout(5000) + }); + + if (!response.ok) { + logger.error(`Failed to fetch URLs: ${response.status} ${response.statusText}`); + return null; + } + + this.cachedUrls = await response.json() as URLsResponse; + this.urlsCacheTime = now; + return this.cachedUrls; + } catch (error: any) { + logger.error(`Error calling /urls API: ${error.message}`); + return null; + } + } + + /** + * Read leader info from file and /urls API */ async getLeaderInfo(): Promise { try { - const content = await fs.readFile(this.infoFilePath, 'utf-8'); - this.leaderInfo = JSON.parse(content) as LeaderLockInfo; + // Read API URL from file (plain text) + const apiUrl = await this.getApiUrlFromFile(); + if (!apiUrl) { + return null; + } + + // Fetch full info from /urls API + const urls = await this.fetchUrls(apiUrl); + if (!urls) { + return null; + } + + // Convert URLsResponse to LeaderLockInfo + this.leaderInfo = { + hostname: urls.hostname, + baseUrl: urls.baseUrl, + apiUrl: urls.apiUrl, + redisUrl: urls.redisUrl, + webdavUrl: urls.webdavUrl, + timestamp: Date.now(), + pid: 0 // Unknown for remote leader + }; return this.leaderInfo; } catch (error: any) { logger.error(`Failed to read leader info: ${error.message}`); @@ -91,12 +160,11 @@ export class LeaderClient { */ async getUrls(): Promise { // First try using configured metaCoreUrl - let apiUrl = this.config.metaCoreUrl; + let apiUrl: string | null = this.config.metaCoreUrl ?? null; - // Fall back to reading from leader info + // Fall back to reading from file if (!apiUrl) { - const info = await this.getLeaderInfo(); - apiUrl = info?.apiUrl; + apiUrl = await this.getApiUrlFromFile(); } if (!apiUrl) { @@ -104,23 +172,7 @@ export class LeaderClient { return null; } - try { - const response = await fetch(`${apiUrl}/urls`, { - method: 'GET', - headers: { 'Accept': 'application/json' }, - signal: AbortSignal.timeout(5000) - }); - - if (!response.ok) { - logger.error(`Failed to get URLs: ${response.status} ${response.statusText}`); - return null; - } - - return await response.json() as URLsResponse; - } catch (error: any) { - logger.error(`Error calling /urls API: ${error.message}`); - return null; - } + return this.fetchUrls(apiUrl); } /** @@ -157,7 +209,11 @@ export class LeaderClient { try { this.watcher = watch(lockDir, (eventType, filename) => { if (filename === 'kv-leader.info') { - logger.debug(`Lock file ${eventType}, checking for leader...`); + logger.debug(`Lock file ${eventType}, invalidating cache...`); + // Invalidate cache to force fresh API call + this.cachedUrls = null; + this.urlsCacheTime = 0; + this.getLeaderInfo().then(() => { this.notifyChange(); }).catch((err) => logger.error(`Error reloading leader info: ${err.message}`)); diff --git a/packages/meta-fuse-core/src/kv/RedisClient.ts b/packages/meta-fuse-core/src/kv/RedisClient.ts index 2ab7d04..d28ecad 100644 --- a/packages/meta-fuse-core/src/kv/RedisClient.ts +++ b/packages/meta-fuse-core/src/kv/RedisClient.ts @@ -10,10 +10,26 @@ import { Redis } from 'ioredis'; import { Logger } from 'tslog'; +import * as os from 'os'; import type { IKVClient, KeyValuePair } from './IKVClient.js'; const logger = new Logger({ name: 'RedisClient' }); +/** + * Stream message from Redis Streams + */ +export interface StreamMessage { + id: string; + type: 'batch' | 'reset' | 'plugin:complete'; + payload: string; + timestamp: string; +} + +/** + * Stream consumer callback + */ +export type StreamMessageHandler = (message: StreamMessage) => Promise; + /** * Full metadata structure including fields needed for virtual path computation */ @@ -72,6 +88,11 @@ export class RedisClient implements Partial { private reconnectTimer: NodeJS.Timeout | null = null; private subscriptions: Map void> = new Map(); + // Stream consumer state + private streamConsumerRunning = false; + private streamConsumerAbort: AbortController | null = null; + private consumerName: string; + constructor(config: RedisClientConfig = {}) { this.config = { url: config.url ?? process.env.REDIS_URL ?? 'redis://127.0.0.1:6379', @@ -79,6 +100,9 @@ export class RedisClient implements Partial { filesVolume: config.filesVolume ?? process.env.FILES_VOLUME ?? '/files', reconnectInterval: config.reconnectInterval ?? 5000, }; + + // Unique consumer name for this instance + this.consumerName = `${os.hostname()}-${process.pid}`; } /** @@ -163,6 +187,9 @@ export class RedisClient implements Partial { * Disconnect from Redis */ async disconnect(): Promise { + // Stop stream consumer first + this.stopStreamConsumer(); + if (this.reconnectTimer) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; @@ -442,6 +469,224 @@ export class RedisClient implements Partial { } } + // ======================================================================== + // 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., 'meta-sort:events') + * @param group - Consumer group name (e.g., 'meta-fuse-vfs') + */ + async initStreamConsumer(stream: string, group: string): Promise { + if (!this.client) { + throw new Error('Redis not connected'); + } + + try { + // Create consumer group at position 0 (read all historical events) + // MKSTREAM creates the stream if it doesn't exist + await this.client.xgroup('CREATE', stream, group, '0', 'MKSTREAM'); + logger.info(`Created consumer group '${group}' for stream '${stream}'`); + } catch (error: any) { + // BUSYGROUP means group already exists - that's fine + if (error.message?.includes('BUSYGROUP')) { + logger.debug(`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 { + if (!this.client) { + return 0; + } + + let processed = 0; + let cursor = '0-0'; + + try { + while (true) { + // XAUTOCLAIM claims idle entries and returns them + const result = await this.client.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.client.xack(stream, group, id); + processed++; + } + } catch (error: any) { + logger.error(`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) { + logger.info(`Processed ${processed} pending stream entries`); + } + } catch (error: any) { + logger.error('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.client) { + throw new Error('Redis not connected'); + } + + if (this.streamConsumerRunning) { + logger.warn('Stream consumer already running'); + return; + } + + this.streamConsumerRunning = true; + this.streamConsumerAbort = new AbortController(); + + logger.info(`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) + // Use call() to bypass TypeScript's strict argument checking for xreadgroup + const result = await (this.client.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.client!.xack(stream, group, id); + } + } catch (error: any) { + logger.error(`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; + } + logger.error('Stream consumer error:', error.message); + // Brief pause before retrying + await new Promise(resolve => setTimeout(resolve, 1000)); + } + } + + logger.info('Stream consumer stopped'); + } + + /** + * Stop the stream consumer loop + */ + stopStreamConsumer(): void { + if (!this.streamConsumerRunning) { + return; + } + + logger.info('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', 'batch', 'payload', '...', '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`); + return null; + } + + return { + id, + type: fieldMap.type as StreamMessage['type'], + payload: fieldMap.payload, + timestamp: fieldMap.timestamp || '0', + }; + } + /** * Get stats about Redis data */ diff --git a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts b/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts index 7013717..c8f28a5 100644 --- a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts +++ b/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts @@ -7,8 +7,8 @@ * Services can discover each other by reading these JSON files. * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale. * - * Note: meta-core (Go sidecar) handles actual service registration. - * This class is primarily used for discovering other services. + * Each service must call start() to register itself and begin heartbeat updates. + * meta-core runs a cleanup process that removes services with stale heartbeats. */ import { promises as fs } from 'fs'; @@ -79,9 +79,8 @@ export class ServiceDiscovery { } /** - * Register this service - * Note: In the new architecture, meta-core handles registration. - * This is kept for backwards compatibility and testing. + * Register this service by writing service info to the shared filesystem. + * meta-core's cleanup process monitors these files and removes stale entries. */ async register(): Promise { await this.ensureServicesDir(); @@ -292,8 +291,7 @@ export class ServiceDiscovery { // ======================================================================== /** - * Full startup sequence - * Note: In the new architecture, meta-core handles registration. + * Full startup sequence: register service and start heartbeat loop. */ async start(): Promise { if (this.isStarted) return; diff --git a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts index 4dfdaad..e232b49 100644 --- a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts +++ b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts @@ -53,7 +53,6 @@ export interface VFSConfig { directoryMode?: number; uid?: number; gid?: number; - refreshInterval?: number; configDir?: string; // Directory for config files (renaming rules, etc.) webdavBaseUrl?: string | null; // meta-sort WebDAV URL for file access (e.g., http://meta-sort-dev/webdav) filesPath?: string; // Local files path prefix to strip when building WebDAV URLs @@ -69,7 +68,6 @@ export class VirtualFileSystem { private conditionEvaluator: ConditionEvaluator; private rulesConfig: RenamingConfig | null = null; private config: Required; - private refreshTimer: NodeJS.Timeout | null = null; private lastRefresh: Date | null = null; private cachedStats = { @@ -93,7 +91,6 @@ export class VirtualFileSystem { directoryMode: config.directoryMode ?? 0o755, uid: config.uid ?? 1000, gid: config.gid ?? 1000, - refreshInterval: config.refreshInterval ?? 30000, // 30 seconds configDir, webdavBaseUrl: config.webdavBaseUrl ?? null, filesPath: config.filesPath ?? '/files', @@ -120,86 +117,45 @@ export class VirtualFileSystem { /** * Start the VFS (connect to Redis and load data) + * Initial bootstrap only - updates come via Redis Streams */ async start(): Promise { logger.info('Starting VirtualFileSystem'); - // Initial load + // Initial load from Redis (bootstrap) await this.refresh(); - // Start periodic refresh - this.refreshTimer = setInterval(async () => { - try { - await this.refresh(); - } catch (error) { - logger.error('Failed to refresh VFS:', error); - } - }, this.config.refreshInterval); - - logger.info('VirtualFileSystem started'); + logger.info('VirtualFileSystem started (updates via Redis Streams)'); } /** * Stop the VFS */ async stop(): Promise { - if (this.refreshTimer) { - clearInterval(this.refreshTimer); - this.refreshTimer = null; - } logger.info('VirtualFileSystem stopped'); } /** * Reset the VFS to empty state * Called when meta-sort triggers a fresh scan - * The VFS will be rebuilt incrementally as files are processed via pub/sub - * - * IMPORTANT: This stops the periodic refresh timer to prevent repopulating - * from Redis (which still has old data). The VFS will only update via - * incremental pub/sub events until the next full refresh is triggered. + * The VFS will be rebuilt incrementally as files are processed via Redis Streams */ reset(): void { logger.info('Resetting VFS for fresh scan...'); - // Stop periodic refresh to prevent repopulating from stale Redis data - if (this.refreshTimer) { - clearInterval(this.refreshTimer); - this.refreshTimer = null; - logger.info('Periodic refresh timer stopped - VFS will update via pub/sub only'); - } - this.initRoot(); this.lastRefresh = null; logger.info('VFS reset complete - awaiting new file updates from meta-sort'); } - /** - * Resume periodic refresh from Redis - * Called after a scan is complete to re-enable auto-refresh - */ - resumePeriodicRefresh(): void { - if (this.refreshTimer) { - logger.debug('Periodic refresh already running'); - return; - } - - logger.info('Resuming periodic refresh from Redis'); - this.refreshTimer = setInterval(async () => { - try { - await this.refresh(); - } catch (error) { - logger.error('Failed to refresh VFS:', error); - } - }, this.config.refreshInterval); - } - /** * Refresh the VFS from Redis + * Called once on startup for initial bootstrap. + * Subsequent updates come via Redis Streams (onFileUpdate). * Uses template-based rules if configured, otherwise falls back to MetaDataToFolderStruct */ async refresh(): Promise { - logger.debug('Refreshing VFS from Redis'); + logger.debug('Loading VFS from Redis (bootstrap)'); try { // Load rules configuration diff --git a/packages/meta-fuse-ui/src/App.tsx b/packages/meta-fuse-ui/src/App.tsx index 1aca166..a899819 100644 --- a/packages/meta-fuse-ui/src/App.tsx +++ b/packages/meta-fuse-ui/src/App.tsx @@ -1,5 +1,6 @@ import { useState, useEffect, useCallback } from 'react'; import { RulesEditor } from './components/RulesEditor'; +import ServiceNav from './components/ServiceNav'; interface Stats { fileCount: number; @@ -37,6 +38,7 @@ function App() { const [error, setError] = useState(null); const [refreshing, setRefreshing] = useState(false); const [showRules, setShowRules] = useState(false); + const [showSetup, setShowSetup] = useState(false); const fetchData = useCallback(async () => { try { @@ -97,6 +99,7 @@ function App() { return (
+

meta-fuse

@@ -110,6 +113,32 @@ function App() { {error &&
{error}
} +
+

What is meta-fuse?

+

+ meta-fuse is a virtual filesystem that presents your media files + in an organized structure based on metadata. It reads file information from the + KV leader (via Redis) and creates a virtual view of your library without duplicating any files. +

+
    +
  • Access files organized by title, year, series, etc.
    • +
  • Mount via FUSE (native) or WebDAV (network drive)
    • +
  • No file duplication - virtual paths point to real files
    • +
  • Automatically updates when new files are processed
    • +
+
+ +
+ + 📁 +
+

WebDAV Files

+

Browse and access your organized media library. Mount as a network drive on any device.

+ /webdav +
+ +
+

VFS Control

@@ -222,6 +251,47 @@ function App() { )}
+ +
+
setShowSetup(!showSetup)} + > +

Setup & Configuration

+ ▼ +
+ {showSetup && ( +
+
+

FUSE Mount (Docker)

+

Mount the virtual filesystem directly from the container. Add this volume to your docker-compose.yml to access the organized files on your host:

+
+ volumes:
  - /path/on/host:/mnt/virtual:rw,shared +
+

The virtual filesystem is available inside the container at /mnt/virtual

+
+ +
+

WebDAV Network Drive

+

Mount the virtual filesystem as a network drive on any device:

+
+ Windows: net use Z: http://<host>/webdav +
+
+ macOS: Finder → Go → Connect to Server → http://<host>/webdav +
+
+ Linux: sudo mount -t davfs http://<host>/webdav /mnt/vfs +
+
+
+ )} +
+ +
+

Part of the MetaMesh project

+

Discovers KV leader via shared volume and connects to Redis automatically

+
); } diff --git a/packages/meta-fuse-ui/src/components/ServiceNav.tsx b/packages/meta-fuse-ui/src/components/ServiceNav.tsx new file mode 100644 index 0000000..9cfefaf --- /dev/null +++ b/packages/meta-fuse-ui/src/components/ServiceNav.tsx @@ -0,0 +1,170 @@ +import { useState, useEffect } from 'react'; + +interface ServiceInfo { + name: string; + url: string; + api: string; + status: string; + capabilities: string[]; + version: string; + role?: string; // "leader", "follower", or undefined (for non-meta-core services) +} + +interface ServicesResponse { + services: ServiceInfo[]; + current: string; +} + +const serviceIcons: Record = { + 'meta-sort': '📁', + 'meta-fuse': '🗂️', + 'meta-stremio': '🎬', + 'meta-orbit': '🌐', + 'meta-core': '⚙️', + 'default': '📦' +}; + +function formatServiceName(name: string): string { + return name.split('-').map(word => + word.charAt(0).toUpperCase() + word.slice(1) + ).join(' '); +} + +function ServiceNav() { + const [services, setServices] = useState([]); + const [loading, setLoading] = useState(true); + const currentService = 'meta-fuse'; + + useEffect(() => { + const fetchServices = async () => { + try { + const response = await fetch('/api/services'); + if (response.ok) { + const data: ServicesResponse = await response.json(); + setServices(data.services || []); + } + } catch (error) { + console.error('Failed to fetch services:', error); + } finally { + setLoading(false); + } + }; + + fetchServices(); + const interval = setInterval(fetchServices, 30000); + return () => clearInterval(interval); + }, []); + + if (loading) return null; + + // Filter out follower meta-core instances (only show leader) + const filteredServices = services.filter( + s => s.name !== 'meta-core' || s.role === 'leader' + ); + const sortedServices = [...filteredServices].sort((a, b) => a.name.localeCompare(b.name)); + + if (sortedServices.length === 0) return null; + + return ( + + ); +} + +export default ServiceNav; diff --git a/packages/meta-fuse-ui/src/index.css b/packages/meta-fuse-ui/src/index.css index 56d7dea..0d613ad 100644 --- a/packages/meta-fuse-ui/src/index.css +++ b/packages/meta-fuse-ui/src/index.css @@ -826,3 +826,217 @@ a:hover { margin-bottom: 1rem; font-size: 0.875rem; } + +/* Description Section */ +.description { + background: var(--bg-secondary); + border-radius: 12px; + padding: 1.5rem; + margin-bottom: 2rem; + border: 1px solid var(--border-color); +} + +.description h2 { + color: var(--accent-primary); + font-size: 1.25rem; + margin-bottom: 1rem; +} + +.description p { + color: var(--text-secondary); + line-height: 1.8; + margin-bottom: 1rem; +} + +.description p strong { + color: var(--text-primary); +} + +.description ul { + margin-left: 1.5rem; + line-height: 2; + color: var(--text-secondary); +} + +/* Quick Links Section */ +.quick-links { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); + gap: 1.5rem; + margin-bottom: 2rem; +} + +.quick-link-card { + display: flex; + align-items: flex-start; + gap: 1rem; + background: var(--bg-secondary); + border: 1px solid var(--border-color); + border-radius: 12px; + padding: 1.5rem; + text-decoration: none; + color: inherit; + transition: all 0.2s; +} + +.quick-link-card:hover { + border-color: var(--accent-primary); + transform: translateY(-2px); + box-shadow: 0 4px 20px rgba(78, 205, 196, 0.15); +} + +.quick-link-icon { + font-size: 2rem; + flex-shrink: 0; +} + +.quick-link-content h3 { + color: var(--text-primary); + font-size: 1.125rem; + margin-bottom: 0.5rem; +} + +.quick-link-content p { + color: var(--text-secondary); + font-size: 0.875rem; + line-height: 1.5; + margin-bottom: 0.75rem; +} + +.quick-link-url { + font-family: monospace; + font-size: 0.875rem; + color: var(--accent-primary); + background: rgba(78, 205, 196, 0.1); + padding: 0.25rem 0.5rem; + border-radius: 4px; +} + +/* Setup Section */ +.setup-section { + margin-top: 2rem; + margin-bottom: 2rem; +} + +.setup-header { + display: flex; + align-items: center; + justify-content: space-between; + cursor: pointer; + padding: 1rem 1.25rem; + background: var(--bg-secondary); + border: 1px solid var(--border-color); + border-radius: 12px; + transition: all 0.2s; + user-select: none; +} + +.setup-header:hover { + background: var(--bg-tertiary); + border-color: rgba(78, 205, 196, 0.3); +} + +.setup-header h3 { + color: var(--text-secondary); + font-size: 1rem; + font-weight: 500; + margin: 0; +} + +.toggle-icon { + font-size: 1rem; + color: var(--text-secondary); + transition: transform 0.2s; +} + +.toggle-icon.open { + transform: rotate(180deg); +} + +.setup-content { + margin-top: 1rem; + display: flex; + flex-direction: column; + gap: 1rem; +} + +.info-box { + background: var(--bg-secondary); + border: 1px solid var(--border-color); + border-radius: 12px; + padding: 1.25rem; +} + +.info-box h4 { + color: var(--accent-primary); + font-size: 0.95rem; + margin-bottom: 0.75rem; +} + +.info-box p { + color: var(--text-secondary); + font-size: 0.875rem; + margin-bottom: 1rem; + line-height: 1.6; +} + +.info-box .info-note { + margin-top: 0.75rem; + margin-bottom: 0; +} + +.info-box code { + background: var(--bg-primary); + padding: 0.125rem 0.375rem; + border-radius: 4px; + font-family: monospace; + color: var(--accent-primary); +} + +.command-box { + background: var(--bg-primary); + border-radius: 8px; + padding: 0.75rem 1rem; + margin-bottom: 0.5rem; + font-family: monospace; + font-size: 0.875rem; + overflow-x: auto; +} + +.command-box:last-child { + margin-bottom: 0; +} + +.command-box strong { + color: var(--accent-primary); + display: inline-block; + min-width: 80px; +} + +.command-box code { + color: var(--text-primary); + background: none; + padding: 0; +} + +/* Footer */ +.app-footer { + text-align: center; + padding-top: 2rem; + margin-top: 2rem; + border-top: 1px solid var(--border-color); + color: var(--text-secondary); + font-size: 0.875rem; +} + +.app-footer p { + margin-bottom: 0.5rem; +} + +.app-footer p:last-child { + margin-bottom: 0; +} + +.app-footer strong { + color: var(--accent-primary); +}