diff --git a/packages/meta-fuse-core/src/api/APIServer.ts b/packages/meta-fuse-core/src/api/APIServer.ts index 76d582d..f2492d9 100644 --- a/packages/meta-fuse-core/src/api/APIServer.ts +++ b/packages/meta-fuse-core/src/api/APIServer.ts @@ -85,10 +85,7 @@ export class APIServer { this.app.post('/api/fuse/rules/validate', this.handleValidateRule.bind(this)); this.app.get('/api/fuse/rules/variables', this.handleGetVariables.bind(this)); - // Service discovery (for inter-service navigation) - if (this.kvManager) { - this.app.get('/api/services', this.handleServices.bind(this)); - } + // Note: /api/services is handled by nginx proxy to meta-core (centralized service discovery) } /** @@ -162,7 +159,7 @@ export class APIServer { return; } - const attrs = this.vfs.getattr(path); + const attrs = await this.vfs.getattr(path); if (attrs === null) { reply.status(404).send({ error: 'Path not found' }); @@ -198,7 +195,7 @@ export class APIServer { return; } - const result = this.vfs.read(path); + const result = await this.vfs.read(path); if (result === null) { reply.status(404).send({ error: 'File not found' }); @@ -261,64 +258,6 @@ export class APIServer { reply.send({ status: 'ok' }); } - private async handleServices(_req: FastifyRequest, reply: FastifyReply): Promise { - interface ServiceResponse { - name: string; - url: string; - api: string; - status: string; - role?: string; - } - - const services: ServiceResponse[] = []; - let leaderInfo: { - hostname: string; - baseUrl: string; - apiUrl: string; - redisUrl: string; - webdavUrl: string; - } | null = null; - - try { - const serviceDiscovery = this.kvManager?.getServiceDiscovery(); - if (serviceDiscovery) { - const allServices = await serviceDiscovery.discoverAllServices(); - - for (const svc of allServices) { - services.push({ - name: svc.name || 'Unknown', - url: svc.baseUrl || '', - api: svc.baseUrl || '', - status: svc.status || 'unknown', - role: svc.role, - }); - } - } - } catch (error) { - logger.error('Error discovering services:', error); - } - - // Get leader info - if (this.kvManager) { - const info = this.kvManager.getLeaderInfo(); - if (info) { - leaderInfo = { - hostname: info.hostname, - baseUrl: info.baseUrl, - apiUrl: info.apiUrl, - redisUrl: info.redisUrl, - webdavUrl: info.webdavUrl, - }; - } - } - - reply.send({ - services, - current: 'meta-fuse', - leader: leaderInfo, - }); - } - // ============================================ // Renaming Rules API Handlers // ============================================ diff --git a/packages/meta-fuse-core/src/kv/KVManager.ts b/packages/meta-fuse-core/src/kv/KVManager.ts index 2fae729..58c0078 100644 --- a/packages/meta-fuse-core/src/kv/KVManager.ts +++ b/packages/meta-fuse-core/src/kv/KVManager.ts @@ -13,7 +13,7 @@ import { hostname } from 'os'; import { Logger } from 'tslog'; import { LeaderClient } from './LeaderClient.js'; -import { ServiceDiscovery } from './ServiceDiscovery.js'; +import { ServiceRegistration } from './ServiceRegistration.js'; import { RedisClient } from './RedisClient.js'; import type { LeaderLockInfo } from './IKVClient.js'; @@ -48,7 +48,7 @@ interface KVManagerConfig { export class KVManager { private config: Required; private leaderClient: LeaderClient | null = null; - private serviceDiscovery: ServiceDiscovery | null = null; + private serviceRegistration: ServiceRegistration | null = null; private redisClient: RedisClient | null = null; private isStarted = false; private isShuttingDown = false; @@ -80,14 +80,14 @@ export class KVManager { logger.info(`Starting KVManager for ${this.config.serviceName}...`); - // Initialize and start service discovery for registration and heartbeat + // Initialize and start service registration for heartbeat const apiUrl = this.config.baseUrl || `http://${hostname()}:${this.config.apiPort}`; - this.serviceDiscovery = new ServiceDiscovery({ + this.serviceRegistration = new ServiceRegistration({ metaCorePath: this.config.metaCorePath, serviceName: this.config.serviceName, apiUrl }); - await this.serviceDiscovery.start(); + await this.serviceRegistration.start(); // If direct Redis URL provided, skip leader discovery if (this.config.redisUrl) { @@ -162,10 +162,10 @@ export class KVManager { this.leaderClient = null; } - // Stop service discovery (if it was started) - if (this.serviceDiscovery) { - await this.serviceDiscovery.stop(); - this.serviceDiscovery = null; + // Stop service registration (if it was started) + if (this.serviceRegistration) { + await this.serviceRegistration.stop(); + this.serviceRegistration = null; } // Disconnect Redis @@ -341,10 +341,10 @@ export class KVManager { } /** - * Get service discovery (for inter-service navigation) + * Get service registration instance */ - getServiceDiscovery(): ServiceDiscovery | null { - return this.serviceDiscovery; + getServiceRegistration(): ServiceRegistration | null { + return this.serviceRegistration; } /** diff --git a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts b/packages/meta-fuse-core/src/kv/ServiceRegistration.ts similarity index 57% rename from packages/meta-fuse-core/src/kv/ServiceDiscovery.ts rename to packages/meta-fuse-core/src/kv/ServiceRegistration.ts index c8f28a5..2601e6f 100644 --- a/packages/meta-fuse-core/src/kv/ServiceDiscovery.ts +++ b/packages/meta-fuse-core/src/kv/ServiceRegistration.ts @@ -1,11 +1,11 @@ /** - * Service Discovery using shared filesystem + * Service Registration using shared filesystem * * Each service registers itself in /meta-core/services/{service-name}-{hostname}.json * with its base URL, status, and heartbeat timestamp. * - * Services can discover each other by reading these JSON files. - * Stale detection: if lastHeartbeat > 60 seconds ago, service is considered stale. + * Service discovery is centralized in meta-core - this class only handles registration. + * For service discovery, use the /api/services endpoint (proxied to meta-core via nginx). * * Each service must call start() to register itself and begin heartbeat updates. * meta-core runs a cleanup process that removes services with stale heartbeats. @@ -13,13 +13,12 @@ import { promises as fs } from 'fs'; import { hostname } from 'os'; -import { join } from 'path'; import { Logger } from 'tslog'; import type { ServiceInfo } from './IKVClient.js'; -const logger = new Logger({ name: 'ServiceDiscovery' }); +const logger = new Logger({ name: 'ServiceRegistration' }); -interface ServiceDiscoveryConfig { +interface ServiceRegistrationConfig { /** Path to META_CORE_VOLUME (e.g., /meta-core) */ metaCorePath: string; @@ -36,8 +35,8 @@ interface ServiceDiscoveryConfig { staleThreshold?: number; } -export class ServiceDiscovery { - private config: ServiceDiscoveryConfig; +export class ServiceRegistration { + private config: ServiceRegistrationConfig; private servicesDir: string; private serviceFilePath: string; private heartbeatTimer: NodeJS.Timeout | null = null; @@ -45,7 +44,7 @@ export class ServiceDiscovery { private isStarted = false; private currentHostname: string; - constructor(config: ServiceDiscoveryConfig) { + constructor(config: ServiceRegistrationConfig) { this.config = { heartbeatInterval: 30000, staleThreshold: 60000, @@ -167,125 +166,6 @@ export class ServiceDiscovery { } } - /** - * Check if a service info is stale - */ - private isStale(info: ServiceInfo): boolean { - const lastHeartbeat = new Date(info.lastHeartbeat).getTime(); - return Date.now() - lastHeartbeat > this.config.staleThreshold!; - } - - // ======================================================================== - // Discovery Methods - // ======================================================================== - - /** - * Discover a service by name - * Looks for files matching pattern: {name}-*.json (hostname-based naming) - */ - async discoverService(name: string): Promise { - try { - // First try exact match for backward compatibility - const exactPath = join(this.servicesDir, `${name}.json`); - try { - const content = await fs.readFile(exactPath, 'utf-8'); - const service = JSON.parse(content) as ServiceInfo; - - if (this.isStale(service)) { - service.status = 'stale'; - } - - return service.status === 'running' ? service : null; - } catch { - // Try hostname-based files - } - - // Search for hostname-based files: name-*.json - const files = await fs.readdir(this.servicesDir); - const matchingFiles = files.filter(f => - f.startsWith(`${name}-`) && f.endsWith('.json') - ); - - // Return first valid (non-stale) service - for (const file of matchingFiles) { - const filePath = join(this.servicesDir, file); - const content = await fs.readFile(filePath, 'utf-8'); - const service = JSON.parse(content) as ServiceInfo; - - if (this.isStale(service)) { - service.status = 'stale'; - continue; // Skip stale services - } - - if (service.status === 'running') { - return service; - } - } - - return null; - } catch (error: any) { - if (error.code === 'ENOENT') { - return null; // Service not registered - } - throw error; - } - } - - /** - * Discover all registered services - */ - async discoverAllServices(): Promise { - try { - const files = await fs.readdir(this.servicesDir); - const services: ServiceInfo[] = []; - - for (const file of files) { - if (!file.endsWith('.json')) continue; - - const filePath = join(this.servicesDir, file); - try { - const content = await fs.readFile(filePath, 'utf-8'); - const service = JSON.parse(content) as ServiceInfo; - - if (this.isStale(service)) { - service.status = 'stale'; - } - - services.push(service); - } catch (error) { - logger.error(`Failed to read ${file}:`, error); - } - } - - return services; - } catch (error: any) { - if (error.code === 'ENOENT') { - return []; // Services directory doesn't exist yet - } - throw error; - } - } - - /** - * Check if a service is healthy - */ - async isServiceHealthy(name: string): Promise { - const service = await this.discoverService(name); - if (!service) return false; - - // Try to ping the service's health endpoint - const url = `${service.baseUrl}/health`; - - try { - const response = await fetch(url, { - signal: AbortSignal.timeout(5000) - }); - return response.ok; - } catch { - return false; - } - } - // ======================================================================== // Lifecycle // ======================================================================== diff --git a/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts b/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts index c5162d8..4a5cb72 100644 --- a/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts +++ b/packages/meta-fuse-core/src/vfs/RulesPropertyExtractor.ts @@ -30,6 +30,7 @@ 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 + 'sizeByte', // Alternative name for size (used by meta-sort plugins) 'mtime', // Modification time 'ctime', // Creation time 'fileName', // Filename for fallback naming diff --git a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts index 2d47d05..19a8183 100644 --- a/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts +++ b/packages/meta-fuse-core/src/vfs/VirtualFileSystem.ts @@ -667,6 +667,38 @@ export class VirtualFileSystem implements VFSUpdateCallback { this.cachedStats.totalSize += size; } + /** + * Ensure file has a valid size, fetching via WebDAV if needed + * This is called lazily when getattr() or read() is invoked + */ + private async ensureFileSize(node: VFSNode): Promise { + // Only fetch if size is 0/missing and we have WebDAV configured + if (node.size && node.size > 0) { + return; + } + + if (!node.sourcePath || !this.config.webdavBaseUrl) { + return; + } + + logger.debug(`Fetching file size via WebDAV for: ${node.sourcePath}`); + const stats = await this.statViaWebDAV(node.sourcePath); + + if (stats && stats.size > 0) { + // Update the node with the fetched size + const oldSize = node.size ?? 0; + node.size = stats.size; + if (stats.mtime) { + node.mtime = stats.mtime; + } + // Update total size stats + this.cachedStats.totalSize = this.cachedStats.totalSize - oldSize + stats.size; + logger.debug(`WebDAV size fetch successful: ${node.sourcePath} = ${stats.size} bytes`); + } else { + logger.warn(`WebDAV size fetch failed for: ${node.sourcePath}`); + } + } + /** * Get file stats via WebDAV HEAD request */ @@ -841,8 +873,9 @@ export class VirtualFileSystem implements VFSUpdateCallback { /** * Get file attributes + * Lazily fetches file size via WebDAV if missing */ - getattr(filepath: string): FileAttributes | null { + async getattr(filepath: string): Promise { filepath = this.normalizePath(filepath); const node = this.nodes.get(filepath); @@ -862,6 +895,9 @@ export class VirtualFileSystem implements VFSUpdateCallback { gid: this.config.gid, }; } else { + // Lazily fetch size via WebDAV if missing + await this.ensureFileSize(node); + return { size: node.size!, mode: 0o100000 | this.config.fileMode, @@ -885,8 +921,9 @@ export class VirtualFileSystem implements VFSUpdateCallback { /** * Read file (returns source path and/or WebDAV URL for actual reading) + * Lazily fetches file size via WebDAV if missing */ - read(filepath: string): ReadResult | null { + async read(filepath: string): Promise { filepath = this.normalizePath(filepath); const node = this.nodes.get(filepath); @@ -894,6 +931,9 @@ export class VirtualFileSystem implements VFSUpdateCallback { return null; } + // Lazily fetch size via WebDAV if missing + await this.ensureFileSize(node); + // Compute WebDAV URL if configured let webdavUrl: string | null = null; if (this.config.webdavBaseUrl && node.sourcePath) { diff --git a/packages/meta-fuse-ui/src/components/ServiceNav.tsx b/packages/meta-fuse-ui/src/components/ServiceNav.tsx index 9cfefaf..8af754a 100644 --- a/packages/meta-fuse-ui/src/components/ServiceNav.tsx +++ b/packages/meta-fuse-ui/src/components/ServiceNav.tsx @@ -2,12 +2,10 @@ import { useState, useEffect } from 'react'; interface ServiceInfo { name: string; - url: string; - api: string; + hostname: string; + baseUrl: string; status: string; - capabilities: string[]; - version: string; - role?: string; // "leader", "follower", or undefined (for non-meta-core services) + lastHeartbeat: string; } interface ServicesResponse { @@ -19,8 +17,9 @@ const serviceIcons: Record = { 'meta-sort': '📁', 'meta-fuse': '🗂️', 'meta-stremio': '🎬', - 'meta-orbit': '🌐', 'meta-core': '⚙️', + 'meta-dup': '🔍', + 'meta-orbit': '🌐', 'default': '📦' }; @@ -57,11 +56,7 @@ function ServiceNav() { 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)); + const sortedServices = [...services].sort((a, b) => a.name.localeCompare(b.name)); if (sortedServices.length === 0) return null; @@ -75,12 +70,12 @@ function ServiceNav() { return ( { e.preventDefault(); if (!isActive) { - window.location.href = service.url; + window.location.href = service.baseUrl; } }} >