From ff62c9c76eacffc3354927083d214c63e8b4219a Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 7 Mar 2026 17:34:58 +0100 Subject: [PATCH] fix queue and stats --- README.md | 1 - packages/meta-sort-core/src/api/FuseAPI.ts | 173 ----- .../meta-sort-core/src/api/FuseAPIServer.ts | 345 ---------- packages/meta-sort-core/src/api/README.md | 299 --------- .../src/api/UnifiedAPIServer.ts | 301 +-------- .../src/api/VirtualFileSystem.ts | 630 ------------------ packages/meta-sort-core/src/api/api.test.ts | 98 +-- packages/meta-sort-core/src/api/index.ts | 5 +- .../ContainerPluginScheduler.ts | 15 +- packages/meta-sort-core/src/index.ts | 11 +- .../logic/UnifiedProcessingStateManager.ts | 5 +- .../src/logic/WatchedFileProcessor.ts | 171 +---- .../fileProcessor/FileProcessorPiscina.ts | 75 ++- .../src/logic/pipeline/PipelineConfig.ts | 6 - .../src/logic/pipeline/StreamingPipeline.ts | 23 +- .../src/metrics/PerformanceMetrics.ts | 48 -- .../src/types/ExtendedInterfaces.ts | 3 - .../meta-sort-core/src/webdav/WebdavClient.ts | 36 +- packages/meta-sort-ui/src/pages/Monitor.tsx | 87 --- packages/meta-sort-ui/src/pages/Plugins.tsx | 20 - 20 files changed, 150 insertions(+), 2202 deletions(-) delete mode 100644 packages/meta-sort-core/src/api/FuseAPI.ts delete mode 100644 packages/meta-sort-core/src/api/FuseAPIServer.ts delete mode 100644 packages/meta-sort-core/src/api/README.md delete mode 100644 packages/meta-sort-core/src/api/VirtualFileSystem.ts diff --git a/README.md b/README.md index e53a98f..d65c407 100644 --- a/README.md +++ b/README.md @@ -822,7 +822,6 @@ gdrive: /files/gdrive rclone vfs-cache-mode=full 0 0 "totalLightProcessing": 16, "totalHashProcessing": 8, "totalDone": 45000, - "totalFilesInVFS": 45150, "throughput": { "filesPerSecond": 34, "avgLightProcessingMs": 470, diff --git a/packages/meta-sort-core/src/api/FuseAPI.ts b/packages/meta-sort-core/src/api/FuseAPI.ts deleted file mode 100644 index a999cb5..0000000 --- a/packages/meta-sort-core/src/api/FuseAPI.ts +++ /dev/null @@ -1,173 +0,0 @@ -/** - * FUSE API for meta-mesh - * - * This API provides a virtual filesystem interface that can be consumed by a FUSE driver. - * It exposes the computed folder structure without creating actual hardlinks. - */ - -import { HashMeta } from '@metazla/meta-interface'; -import { Stats } from 'fs'; - -/** - * Represents a file or directory in the virtual filesystem - */ -export interface VirtualNode { - /** Virtual path in the output filesystem */ - path: string; - /** Node type */ - type: 'file' | 'directory'; - /** File name */ - name: string; - /** Parent directory path */ - parent: string; -} - -/** - * Represents a regular file in the virtual filesystem - */ -export interface VirtualFile extends VirtualNode { - type: 'file'; - /** Source file path on the real filesystem (null for virtual files like .meta) */ - sourcePath: string | null; - /** File size in bytes */ - size: number; - /** Last modified time */ - mtime: Date; - /** Creation time */ - ctime: Date; - /** File permissions (octal) */ - mode: number; - /** Associated metadata */ - metadata?: HashMeta; -} - -/** - * Represents a directory in the virtual filesystem - */ -export interface VirtualDirectory extends VirtualNode { - type: 'directory'; - /** Child nodes (file/directory names) */ - children: string[]; - /** Directory permissions (octal) */ - mode: number; - /** Last modified time */ - mtime: Date; -} - -/** - * File attributes returned by getattr() - compatible with FUSE stat structure - */ -export interface FileAttributes { - /** File size in bytes */ - size: number; - /** File mode (permissions + type) */ - mode: number; - /** Last modified time (Unix timestamp in seconds) */ - mtime: number; - /** Last access time (Unix timestamp in seconds) */ - atime: number; - /** Creation/change time (Unix timestamp in seconds) */ - ctime: number; - /** Number of hard links */ - nlink: number; - /** User ID */ - uid: number; - /** Group ID */ - gid: number; -} - -/** - * Read result for file content - */ -export interface ReadResult { - /** For regular files: source path to read from. For virtual files: null */ - sourcePath: string | null; - /** For virtual files: generated content. For regular files: null */ - content: Buffer | null; - /** Total file size */ - size: number; -} - -/** - * FUSE API operations - */ -export interface IFuseAPI { - /** - * List directory contents - * @param path Virtual directory path (e.g., "/Anime/Naruto/Season 1") - * @returns Array of file/directory names in the directory, or null if not found - */ - readdir(path: string): string[] | null; - - /** - * Get file or directory attributes - * @param path Virtual file or directory path - * @returns File attributes, or null if not found - */ - getattr(path: string): FileAttributes | null; - - /** - * Check if a path exists - * @param path Virtual path - * @returns true if path exists, false otherwise - */ - exists(path: string): boolean; - - /** - * Get read information for a file - * @param path Virtual file path - * @returns Read result with source path or content, or null if not found - */ - read(path: string): ReadResult | null; - - /** - * Get metadata for a file - * @param path Virtual file path - * @returns HashMeta if available, null otherwise - */ - getMetadata(path: string): HashMeta | null; - - /** - * Get the complete virtual filesystem tree - * @returns Root directory node - */ - getTree(): VirtualDirectory; - - /** - * Refresh the virtual filesystem from current state - * Called after processing completes to update the VFS - */ - refresh(): Promise; -} - -/** - * Events emitted by the FUSE API - */ -export interface FuseAPIEvents { - /** Emitted when the virtual filesystem is updated */ - 'vfs-updated': () => void; - /** Emitted when a file is added */ - 'file-added': (path: string) => void; - /** Emitted when a file is updated (e.g., metadata changed) */ - 'file-updated': (path: string) => void; - /** Emitted when a file is removed */ - 'file-removed': (path: string) => void; - /** Emitted when a directory is added */ - 'directory-added': (path: string) => void; -} - -/** - * Configuration for FUSE API - */ -export interface FuseAPIConfig { - /** Base path for the virtual filesystem (default: "/") */ - basePath?: string; - /** Default file permissions (default: 0o644) */ - fileMode?: number; - /** Default directory permissions (default: 0o755) */ - directoryMode?: number; - /** User ID for files (default: process.getuid()) */ - uid?: number; - /** Group ID for files (default: process.getgid()) */ - gid?: number; -} diff --git a/packages/meta-sort-core/src/api/FuseAPIServer.ts b/packages/meta-sort-core/src/api/FuseAPIServer.ts deleted file mode 100644 index f4130bf..0000000 --- a/packages/meta-sort-core/src/api/FuseAPIServer.ts +++ /dev/null @@ -1,345 +0,0 @@ -/** - * HTTP REST API Server for FUSE operations - * - * Exposes the VirtualFileSystem over HTTP for consumption by a FUSE driver. - */ - -import * as http from 'http'; -import { VirtualFileSystem } from './VirtualFileSystem.js'; -import { FileAttributes, ReadResult } from './FuseAPI.js'; - -export interface FuseAPIServerConfig { - /** Port to listen on (default: 3000) */ - port?: number; - /** Host to bind to (default: 'localhost') */ - host?: string; - /** Enable CORS (default: false) */ - enableCors?: boolean; -} - -export class FuseAPIServer { - private server: http.Server | null = null; - private vfs: VirtualFileSystem; - private config: Required; - - constructor(vfs: VirtualFileSystem, config: FuseAPIServerConfig = {}) { - this.vfs = vfs; - this.config = { - port: config.port ?? 3000, - host: config.host ?? 'localhost', - enableCors: config.enableCors ?? false - }; - } - - /** - * Start the HTTP server - */ - public async start(): Promise { - return new Promise((resolve, reject) => { - this.server = http.createServer(this.handleRequest.bind(this)); - - this.server.on('error', reject); - - this.server.listen(this.config.port, this.config.host, () => { - console.log(`FUSE API Server listening on http://${this.config.host}:${this.config.port}`); - resolve(); - }); - }); - } - - /** - * Stop the HTTP server - */ - public async stop(): Promise { - if (!this.server) { - return; - } - - return new Promise((resolve, reject) => { - this.server!.close((err) => { - if (err) { - reject(err); - } else { - console.log('FUSE API Server stopped'); - this.server = null; - resolve(); - } - }); - }); - } - - /** - * Handle incoming HTTP requests - */ - private async handleRequest(req: http.IncomingMessage, res: http.ServerResponse): Promise { - // CORS headers - if (this.config.enableCors) { - res.setHeader('Access-Control-Allow-Origin', '*'); - res.setHeader('Access-Control-Allow-Methods', 'GET, POST, OPTIONS'); - res.setHeader('Access-Control-Allow-Headers', 'Content-Type'); - - if (req.method === 'OPTIONS') { - res.writeHead(204); - res.end(); - return; - } - } - - try { - const url = new URL(req.url!, `http://${req.headers.host}`); - const pathname = url.pathname; - const method = req.method!; - - // Route the request - if (method === 'GET' && pathname === '/api/fuse/health') { - this.handleHealth(res); - } else if (method === 'GET' && pathname === '/api/fuse/stats') { - this.handleStats(res); - } else if (method === 'POST' && pathname === '/api/fuse/readdir') { - await this.handleReaddir(req, res); - } else if (method === 'POST' && pathname === '/api/fuse/getattr') { - await this.handleGetattr(req, res); - } else if (method === 'POST' && pathname === '/api/fuse/exists') { - await this.handleExists(req, res); - } else if (method === 'POST' && pathname === '/api/fuse/read') { - await this.handleRead(req, res); - } else if (method === 'POST' && pathname === '/api/fuse/metadata') { - await this.handleMetadata(req, res); - } else if (method === 'GET' && pathname === '/api/fuse/tree') { - this.handleTree(res); - } else if (method === 'GET' && pathname === '/api/fuse/files') { - this.handleFiles(res); - } else if (method === 'GET' && pathname === '/api/fuse/directories') { - this.handleDirectories(res); - } else if (method === 'POST' && pathname === '/api/fuse/refresh') { - await this.handleRefresh(req, res); - } else { - this.sendError(res, 404, 'Not Found'); - } - } catch (error) { - console.error('Error handling request:', error); - this.sendError(res, 500, 'Internal Server Error', error); - } - } - - /** - * Parse JSON body from request - */ - private async parseBody(req: http.IncomingMessage): Promise { - return new Promise((resolve, reject) => { - let body = ''; - req.on('data', (chunk) => { - body += chunk.toString(); - }); - req.on('end', () => { - try { - resolve(JSON.parse(body)); - } catch (error) { - reject(new Error('Invalid JSON')); - } - }); - req.on('error', reject); - }); - } - - /** - * Send JSON response - */ - private sendJson(res: http.ServerResponse, data: any, statusCode: number = 200): void { - res.writeHead(statusCode, { 'Content-Type': 'application/json' }); - res.end(JSON.stringify(data)); - } - - /** - * Send error response - */ - private sendError(res: http.ServerResponse, statusCode: number, message: string, error?: any): void { - this.sendJson(res, { - error: message, - details: error?.message ?? error - }, statusCode); - } - - // API Handlers - - /** - * GET /api/health - * Health check endpoint - */ - private handleHealth(res: http.ServerResponse): void { - this.sendJson(res, { - status: 'ok', - timestamp: new Date().toISOString() - }); - } - - /** - * GET /api/stats - * Get VFS statistics - */ - private handleStats(res: http.ServerResponse): void { - const stats = this.vfs.getStats(); - this.sendJson(res, stats); - } - - /** - * POST /api/readdir - * List directory contents - * Body: { path: string } - * Response: { entries: string[] } | { error: string } - */ - private async handleReaddir(req: http.IncomingMessage, res: http.ServerResponse): Promise { - const { path } = await this.parseBody(req); - - if (typeof path !== 'string') { - this.sendError(res, 400, 'Missing or invalid "path" parameter'); - return; - } - - const entries = this.vfs.readdir(path); - - if (entries === null) { - this.sendError(res, 404, 'Directory not found'); - return; - } - - this.sendJson(res, { entries }); - } - - /** - * POST /api/getattr - * Get file attributes - * Body: { path: string } - * Response: FileAttributes | { error: string } - */ - private async handleGetattr(req: http.IncomingMessage, res: http.ServerResponse): Promise { - const { path } = await this.parseBody(req); - - if (typeof path !== 'string') { - this.sendError(res, 400, 'Missing or invalid "path" parameter'); - return; - } - - const attrs = this.vfs.getattr(path); - - if (attrs === null) { - this.sendError(res, 404, 'Path not found'); - return; - } - - this.sendJson(res, attrs); - } - - /** - * POST /api/exists - * Check if path exists - * Body: { path: string } - * Response: { exists: boolean } - */ - private async handleExists(req: http.IncomingMessage, res: http.ServerResponse): Promise { - const { path } = await this.parseBody(req); - - if (typeof path !== 'string') { - this.sendError(res, 400, 'Missing or invalid "path" parameter'); - return; - } - - const exists = this.vfs.exists(path); - this.sendJson(res, { exists }); - } - - /** - * POST /api/read - * Read file - * Body: { path: string } - * Response: ReadResult | { error: string } - */ - private async handleRead(req: http.IncomingMessage, res: http.ServerResponse): Promise { - const { path } = await this.parseBody(req); - - if (typeof path !== 'string') { - this.sendError(res, 400, 'Missing or invalid "path" parameter'); - return; - } - - const result = this.vfs.read(path); - - if (result === null) { - this.sendError(res, 404, 'File not found'); - return; - } - - // Convert Buffer to base64 for JSON serialization - const response: any = { - sourcePath: result.sourcePath, - size: result.size - }; - - if (result.content !== null) { - response.content = result.content.toString('base64'); - response.contentEncoding = 'base64'; - } - - this.sendJson(res, response); - } - - /** - * POST /api/metadata - * Get file metadata - * Body: { path: string } - * Response: HashMeta | { error: string } - */ - private async handleMetadata(req: http.IncomingMessage, res: http.ServerResponse): Promise { - const { path } = await this.parseBody(req); - - if (typeof path !== 'string') { - this.sendError(res, 400, 'Missing or invalid "path" parameter'); - return; - } - - const metadata = this.vfs.getMetadata(path); - - if (metadata === null) { - this.sendError(res, 404, 'Metadata not found'); - return; - } - - this.sendJson(res, metadata); - } - - /** - * GET /api/tree - * Get complete VFS tree - */ - private handleTree(res: http.ServerResponse): void { - const tree = this.vfs.getTree(); - this.sendJson(res, tree); - } - - /** - * GET /api/files - * Get all file paths - */ - private handleFiles(res: http.ServerResponse): void { - const files = this.vfs.getAllFiles(); - this.sendJson(res, { files }); - } - - /** - * GET /api/directories - * Get all directory paths - */ - private handleDirectories(res: http.ServerResponse): void { - const directories = this.vfs.getAllDirectories(); - this.sendJson(res, { directories }); - } - - /** - * POST /api/refresh - * Refresh the VFS (no-op, VFS is updated by processing pipeline) - */ - private async handleRefresh(req: http.IncomingMessage, res: http.ServerResponse): Promise { - await this.vfs.refresh(); - this.sendJson(res, { status: 'ok' }); - } -} diff --git a/packages/meta-sort-core/src/api/README.md b/packages/meta-sort-core/src/api/README.md deleted file mode 100644 index 0b1bb66..0000000 --- a/packages/meta-sort-core/src/api/README.md +++ /dev/null @@ -1,299 +0,0 @@ -# FUSE API for Meta-Mesh - -This module provides a FUSE-compatible HTTP API that runs alongside meta-mesh's hardlink system. The API exposes a virtual filesystem that can be consumed by a future FUSE driver. - -## Overview - -The FUSE API runs **always-on alongside hardlinks**, providing: - -- **Dual output**: Both hardlinks (for local access) and virtual filesystem (for FUSE driver) -- **Dynamic updates**: VirtualFileSystem rebuilds on every processing cycle -- **Virtual .meta files**: Metadata files are generated on-the-fly without disk I/O -- **HTTP REST API**: Can be consumed by local or remote FUSE drivers -- **Zero configuration**: Always enabled, no opt-in required - -## Architecture - -### Components - -1. **VirtualFileSystem** - In-memory representation of the output directory structure -2. **FuseAPIServer** - HTTP REST API server exposing FUSE operations -3. **FuseAPI** - TypeScript interfaces defining the API contract - -### Data Flow - -``` -Input Files → File Processing → Metadata Extraction → Duplicate Removal - ↓ -Virtual Folder Structure Computation (diff generation) - ↓ - ├─→ VirtualFileSystem Build (in-memory) → FUSE API Server → Future FUSE Driver - │ - └─→ Hardlink Creation (on disk) → Direct File Access -``` - -Both outputs are generated in parallel from the same computed folder structure. - -## Configuration - -The FUSE API is **always enabled** and runs alongside hardlinks. You can optionally configure the server port and host: - -```bash -# FUSE API Server configuration (optional) -FUSE_API_PORT=3000 # Default: 3000 -FUSE_API_HOST=localhost # Default: localhost -``` - -The API server starts automatically when meta-mesh starts. - -## API Endpoints - -The FUSE API Server exposes the following HTTP REST endpoints: - -### Health & Info - -- **GET /api/health** - Health check - ```json - { "status": "ok", "timestamp": "2025-01-15T10:30:00.000Z" } - ``` - -- **GET /api/stats** - VFS statistics - ```json - { - "fileCount": 1250, - "directoryCount": 45, - "totalSize": 52428800, - "metaFileCount": 1250 - } - ``` - -### Directory Operations - -- **POST /api/readdir** - List directory contents - ```json - // Request - { "path": "/Anime/Naruto/Season 1" } - - // Response - { "entries": ["Naruto S01E01.mkv", "Naruto S01E01.mkv.meta", ...] } - ``` - -### File Operations - -- **POST /api/getattr** - Get file attributes - ```json - // Request - { "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" } - - // Response - { - "size": 524288000, - "mode": 33188, // 0o100644 (regular file, rw-r--r--) - "mtime": 1705315200, - "atime": 1705315200, - "ctime": 1705315200, - "nlink": 1, - "uid": 1000, - "gid": 1000 - } - ``` - -- **POST /api/exists** - Check if path exists - ```json - // Request - { "path": "/Anime/Naruto" } - - // Response - { "exists": true } - ``` - -- **POST /api/read** - Read file - ```json - // Request - { "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" } - - // Response (regular file) - { - "sourcePath": "/data/input/naruto_01.mkv", - "size": 524288000 - } - - // Response (virtual .meta file) - { - "sourcePath": null, - "content": "..base64-encoded content..", - "contentEncoding": "base64", - "size": 1024 - } - ``` - -- **POST /api/metadata** - Get file metadata - ```json - // Request - { "path": "/Anime/Naruto/Season 1/Naruto S01E01.mkv" } - - // Response - { - "title": "Naruto", - "season": "01", - "episode": "01", - "anime": true, - "sha256": "abc123...", - // ... full HashMeta object - } - ``` - -### Tree Operations - -- **GET /api/tree** - Get complete VFS tree structure -- **GET /api/files** - Get all file paths -- **GET /api/directories** - Get all directory paths -- **POST /api/refresh** - Refresh VFS (no-op, VFS updates automatically) - -## Usage Example - -### Starting meta-mesh with FUSE API - -The FUSE API starts automatically with meta-mesh: - -```bash -# Run meta-mesh (FUSE API starts automatically on port 3000) -pnpm run start:mesh - -# Or customize the port -export FUSE_API_PORT=8080 -pnpm run start:mesh -``` - -You'll see output like: -``` -FUSE API Server started on http://localhost:3000 -VirtualFileSystem build took 15ms -VirtualFileSystem: 1250 files (1250 .meta), 45 directories, 50.00 MB -``` - -### Consuming the API - -```typescript -import { VirtualFileSystem, FuseAPIServer } from '@metazla/meta-mesh/api'; - -// Create VFS instance -const vfs = new VirtualFileSystem({ - fileMode: 0o644, - directoryMode: 0o755, - uid: process.getuid(), - gid: process.getgid() -}); - -// Build from computed structure -vfs.buildFromComputed(sourceToDestMap, metadataMap); - -// Start API server -const server = new FuseAPIServer(vfs, { - port: 3000, - host: 'localhost' -}); - -await server.start(); -``` - -### FUSE Driver Integration - -A FUSE driver would make HTTP requests to these endpoints: - -```typescript -// Example pseudo-code for FUSE driver -async function readdir(path: string): Promise { - const response = await fetch('http://localhost:3000/api/readdir', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ path }) - }); - - const { entries } = await response.json(); - return entries; -} - -async function read(path: string): Promise { - const response = await fetch('http://localhost:3000/api/read', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ path }) - }); - - const result = await response.json(); - - if (result.sourcePath) { - // Regular file - read from source - return fs.readFileSync(result.sourcePath); - } else { - // Virtual file - use provided content - return Buffer.from(result.content, 'base64'); - } -} -``` - -## Virtual .meta Files - -`.meta` files are entirely virtual and generated on-the-fly: - -- **No disk I/O**: Content is serialized from in-memory metadata -- **Always up-to-date**: Reflects current metadata state -- **YAML format**: Uses `MetaMeshFormat.serialize()` for consistency -- **No source path**: `sourcePath` is `null`, `content` contains the YAML - -## Events - -The VirtualFileSystem emits events for monitoring: - -```typescript -vfs.on('vfs-updated', () => { - console.log('Virtual filesystem updated'); -}); - -vfs.on('file-added', (path: string) => { - console.log(`File added: ${path}`); -}); - -vfs.on('file-removed', (path: string) => { - console.log(`File removed: ${path}`); -}); - -vfs.on('directory-added', (path: string) => { - console.log(`Directory added: ${path}`); -}); -``` - -## Performance - -The VirtualFileSystem is highly performant: - -- **In-memory**: All operations are O(1) or O(n) in-memory -- **No disk I/O**: Except for reading source files when FUSE driver requests content -- **Lazy .meta generation**: Content is only serialized when requested -- **Fast updates**: Rebuilding the VFS takes milliseconds - -## Dual Output: Hardlinks + FUSE API - -Meta-mesh now provides **both** output methods simultaneously: - -| Feature | Hardlinks | FUSE API | Best For | -|---------|-----------|----------|----------| -| Disk usage | 1 inode per file | 0 (virtual) | Hardlinks: local use | -| Update speed | Slow (file I/O) | Fast (in-memory) | FUSE: frequent updates | -| .meta files | Written to disk | Virtual (on-demand) | FUSE: dynamic metadata | -| Remote access | No | Yes (HTTP) | FUSE: network drives | -| Compatibility | Any filesystem | Requires FUSE driver | Hardlinks: broad compat | -| Access method | Direct file access | HTTP + FUSE driver | Hardlinks: native apps | - -**Use hardlinks** for local applications like Jellyfin, Plex, Kodi that read files directly. - -**Use FUSE API** for future FUSE drivers that want dynamic, virtual filesystem access. - -## Future Enhancements - -- **WebSocket support**: Real-time updates for FUSE drivers -- **Authentication**: API key or JWT-based auth -- **Compression**: Gzip/Brotli for API responses -- **Caching**: ETag/Last-Modified headers -- **Batch operations**: Multi-file operations in single request diff --git a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts index 242547b..870a36d 100644 --- a/packages/meta-sort-core/src/api/UnifiedAPIServer.ts +++ b/packages/meta-sort-core/src/api/UnifiedAPIServer.ts @@ -7,7 +7,6 @@ import Fastify, { FastifyInstance } from 'fastify'; import fastifyCors from '@fastify/cors'; -import { VirtualFileSystem } from './VirtualFileSystem.js'; import { performanceMetrics } from '../metrics/PerformanceMetrics.js'; import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js'; import type { IKVClient } from '../kv/IKVClient.js'; @@ -32,7 +31,6 @@ export interface UnifiedAPIServerConfig { export class UnifiedAPIServer { private app: FastifyInstance; - private vfs: VirtualFileSystem; private unifiedStateManager: UnifiedProcessingStateManager | null = null; private kvClient: IKVClient | null = null; private kvManager: KVManager | null = null; @@ -51,7 +49,6 @@ export class UnifiedAPIServer { private static TOTAL_SIZE_CACHE_TTL_MS = 30000; // 30 seconds constructor( - vfs: VirtualFileSystem, config: UnifiedAPIServerConfig = {}, unifiedStateManager?: UnifiedProcessingStateManager, kvClient?: IKVClient, @@ -62,7 +59,6 @@ export class UnifiedAPIServer { getPluginManager?: () => PluginManager | null, getTaskScheduler?: () => TaskScheduler | null ) { - this.vfs = vfs; this.unifiedStateManager = unifiedStateManager || null; this.kvClient = kvClient || null; this.kvManager = kvManager || null; @@ -155,7 +151,6 @@ export class UnifiedAPIServer { status: 'ok', timestamp: new Date().toISOString(), services: { - fuse: 'ok', metrics: 'ok', unifiedProcessing: this.unifiedStateManager ? 'ok' : 'unavailable' } @@ -176,9 +171,6 @@ export class UnifiedAPIServer { }; }); - // FUSE API routes (under /api/fuse) - this.setupFuseRoutes(); - // Metrics API routes (under /api/metrics) this.setupMetricsRoutes(); @@ -288,191 +280,6 @@ export class UnifiedAPIServer { // - POST /api/mounts/:id/unmount - Request unmount // - GET /api/mounts/rclone/remotes - List rclone remotes - /** - * Setup FUSE API routes - */ - private setupFuseRoutes(): void { - // Health check - this.app.get('/api/fuse/health', async (request, reply) => { - return { - status: 'ok', - timestamp: new Date().toISOString() - }; - }); - - // VFS statistics - this.app.get('/api/fuse/stats', async (request, reply) => { - return this.vfs.getStats(); - }); - - // Database statistics (KV) - this.app.get('/api/fuse/db-stats', async (request, reply) => { - if (!this.kvClient) { - return reply.status(503).send({ - error: 'KV client not available' - }); - } - - try { - // Count files in database with /file/ prefix - const fileCount = await this.kvClient.countKeysWithPrefix('/file/'); - return { - fileCount, - source: 'KV' - }; - } catch (error: any) { - return reply.status(500).send({ - error: 'Failed to retrieve database stats', - details: error.message - }); - } - }); - - // List directory contents - this.app.post<{ Body: { path: string } }>('/api/fuse/readdir', async (request, reply) => { - const { path } = request.body; - - if (typeof path !== 'string') { - return reply.status(400).send({ - error: 'Missing or invalid "path" parameter' - }); - } - - const entries = this.vfs.readdir(path); - - if (entries === null) { - return reply.status(404).send({ error: 'Directory not found' }); - } - - return { entries }; - }); - - // Get file attributes - this.app.post<{ Body: { path: string } }>('/api/fuse/getattr', async (request, reply) => { - const { path } = request.body; - - if (typeof path !== 'string') { - return reply.status(400).send({ - error: 'Missing or invalid "path" parameter' - }); - } - - const attrs = this.vfs.getattr(path); - - if (attrs === null) { - return reply.status(404).send({ error: 'Path not found' }); - } - - return attrs; - }); - - // Check if path exists - this.app.post<{ Body: { path: string } }>('/api/fuse/exists', async (request, reply) => { - const { path } = request.body; - - if (typeof path !== 'string') { - return reply.status(400).send({ - error: 'Missing or invalid "path" parameter' - }); - } - - const exists = this.vfs.exists(path); - return { exists }; - }); - - // Read file - this.app.post<{ Body: { path: string } }>('/api/fuse/read', async (request, reply) => { - const { path } = request.body; - - if (typeof path !== 'string') { - return reply.status(400).send({ - error: 'Missing or invalid "path" parameter' - }); - } - - const result = this.vfs.read(path); - - if (result === null) { - return reply.status(404).send({ error: 'File not found' }); - } - - // Convert Buffer to base64 for JSON serialization - const response: any = { - sourcePath: result.sourcePath, - size: result.size - }; - - if (result.content !== null) { - response.content = result.content.toString('base64'); - response.contentEncoding = 'base64'; - } - - return response; - }); - - // Get file metadata - this.app.post<{ Body: { path: string } }>('/api/fuse/metadata', async (request, reply) => { - const { path } = request.body; - - if (typeof path !== 'string') { - return reply.status(400).send({ - error: 'Missing or invalid "path" parameter' - }); - } - - const metadata = this.vfs.getMetadata(path); - - if (metadata === null) { - return reply.status(404).send({ error: 'Metadata not found' }); - } - - return metadata; - }); - - // Get complete VFS tree - this.app.get('/api/fuse/tree', async (request, reply) => { - return this.vfs.getTree(); - }); - - // Get all file paths - this.app.get('/api/fuse/files', async (request, reply) => { - const files = this.vfs.getAllFiles(); - return { files }; - }); - - // Get all source file paths (for Stremio addon validation) - this.app.get('/api/fuse/source-paths', async (request, reply) => { - const files = this.vfs.getAllFiles(); - const sourcePaths: string[] = []; - - for (const file of files) { - // Skip virtual metadata files (.meta, .nfo, .xml) - if (file.endsWith('.meta') || file.endsWith('.nfo') || file.endsWith('.xml')) { - continue; - } - - const result = this.vfs.read(file); - if (result && result.sourcePath) { - sourcePaths.push(result.sourcePath); - } - } - - return { sourcePaths, count: sourcePaths.length }; - }); - - // Get all directory paths - this.app.get('/api/fuse/directories', async (request, reply) => { - const directories = this.vfs.getAllDirectories(); - return { directories }; - }); - - // Refresh VFS - this.app.post('/api/fuse/refresh', async (request, reply) => { - await this.vfs.refresh(); - return { status: 'ok' }; - }); - } - /** * Setup Metrics API routes */ @@ -573,7 +380,7 @@ export class UnifiedAPIServer { // Get snapshot from state manager // Note: VFS-related features have been moved to meta-fuse // totalDone now uses state manager's count of fully processed files - const snapshot = this.unifiedStateManager!.getSnapshot(0, this.fastQueueConcurrency, this.backgroundQueueConcurrency) as ExtendedProcessingSnapshot; + const snapshot = this.unifiedStateManager!.getSnapshot(this.fastQueueConcurrency, this.backgroundQueueConcurrency) as ExtendedProcessingSnapshot; // Add queue status if available if (this.getQueueStatus) { @@ -681,59 +488,6 @@ export class UnifiedAPIServer { return { status: 'ok', message: `${queued} files queued for retry` }; }); - // POST /api/processing/stop - Stop all processing - // Pauses pipeline queues and closes container plugin gate - // Running tasks will complete, but no new tasks will start - this.app.post('/api/processing/stop', async (request, reply) => { - // Pause the streaming pipeline (stops new files from being processed) - if (this.streamingPipeline) { - this.streamingPipeline.pause(); - } - - // Close the container plugin gate (stops new plugin tasks) - if (this.containerPluginScheduler) { - this.containerPluginScheduler.setGate(false); - } - - const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null; - const pipelinePaused = this.streamingPipeline?.isPaused() ?? false; - - console.log('[API] Processing stopped - pipeline paused, gate closed'); - - return { - status: 'ok', - message: 'Processing stopped - pipeline paused, gate closed', - gateStatus, - pipelinePaused - }; - }); - - // POST /api/processing/start - Resume all processing - // Resumes pipeline queues and opens container plugin gate - this.app.post('/api/processing/start', async (request, reply) => { - // Resume the streaming pipeline - if (this.streamingPipeline) { - this.streamingPipeline.resume(); - } - - // Open the container plugin gate - if (this.containerPluginScheduler) { - this.containerPluginScheduler.setGate(true); - } - - const gateStatus = this.containerPluginScheduler?.getGateStatus() ?? null; - const pipelinePaused = this.streamingPipeline?.isPaused() ?? false; - - console.log('[API] Processing started - pipeline resumed, gate opened'); - - return { - status: 'ok', - message: 'Processing started - pipeline resumed, gate opened', - gateStatus, - pipelinePaused - }; - }); - // POST /api/processing/wait-empty - Wait for queues to drain // Optional query param: timeout (ms, default 60000) this.app.post<{ Querystring: { timeout?: string } }>('/api/processing/wait-empty', async (request, reply) => { @@ -809,56 +563,12 @@ export class UnifiedAPIServer { } /** - * Setup Metadata Clear API routes - * NOTE: /api/scan/trigger has been removed - use meta-core's endpoint instead + * Setup Scan-related API routes + * NOTE: /api/scan/trigger and /api/metadata/clear have been removed + * Use meta-core's endpoints instead */ private setupScanRoutes(): void { - // Clear all metadata - // NOTE: To trigger a rescan after clearing, call meta-core's /api/scan/trigger endpoint - this.app.post('/api/metadata/clear', async (request, reply) => { - if (!this.kvClient) { - return reply.status(503).send({ - status: 'error', - message: 'KV client not available' - }); - } - - try { - console.log('Clearing all metadata via API'); - - // Get all file keys and delete them using flat key deletion - const hashIds = await this.kvClient.getAllHashIds(); - let deletedCount = 0; - - for (const hashId of hashIds) { - try { - // Use deleteMetadataFlat to delete all flat keys for this file - await this.kvClient.deleteMetadataFlat(hashId); - deletedCount++; - } catch (err) { - console.warn(`Failed to delete metadata for ${hashId}:`, err); - } - } - - // Note: deleteMetadataFlat already removes from file:__index__ - // No need to delete the index separately - - console.log(`Cleared ${deletedCount} file metadata entries`); - - return { - status: 'ok', - message: `Cleared ${deletedCount} files. To rescan, call meta-core's /api/scan/trigger endpoint.`, - deletedCount - }; - } catch (error: any) { - console.error('Error clearing metadata:', error); - return reply.status(500).send({ - status: 'error', - message: 'Failed to clear metadata', - details: error.message - }); - } - }); + // No routes - all scan-related functionality moved to meta-core } /** @@ -1785,7 +1495,6 @@ export class UnifiedAPIServer { host: this.config.host }); console.log(`Unified API Server listening on http://${this.config.host}:${this.config.port}`); - console.log(` - FUSE API: http://${this.config.host}:${this.config.port}/api/fuse/*`); console.log(` - Metrics API: http://${this.config.host}:${this.config.port}/api/metrics`); if (this.kvClient) { console.log(` - Metadata API: http://${this.config.host}:${this.config.port}/api/metadata/* (GET/PUT/DELETE/POST)`); diff --git a/packages/meta-sort-core/src/api/VirtualFileSystem.ts b/packages/meta-sort-core/src/api/VirtualFileSystem.ts deleted file mode 100644 index d779075..0000000 --- a/packages/meta-sort-core/src/api/VirtualFileSystem.ts +++ /dev/null @@ -1,630 +0,0 @@ -/** - * VirtualFileSystem - In-memory representation of the virtual output filesystem - * - * This class maintains the virtual directory structure and provides FUSE-compatible operations. - */ - -import { EventEmitter } from 'events'; -import * as path from 'path'; -import * as fs from 'fs'; -import { HashMeta } from '@metazla/meta-interface'; -import type { ExtendedHashMeta } from '../types/ExtendedInterfaces.js'; -import { - IFuseAPI, - VirtualDirectory, - VirtualFile, - FileAttributes, - ReadResult, - FuseAPIConfig, - FuseAPIEvents -} from './FuseAPI.js'; -import { MetaMeshFormat } from '../logic/MetaMeshFormat.js'; -import { MetadataFormatRegistry, IMetadataFormatHandler } from '../logic/MetadataFormatRegistry.js'; -import { config } from '../config/EnvConfig.js'; - -interface VFSNode { - type: 'file' | 'directory'; - name: string; - parent: string | null; - // For files - sourcePath?: string | null; - size?: number; - mtime?: Date; - ctime?: Date; - metadata?: HashMeta; - // For virtual metadata files - formatHandler?: IMetadataFormatHandler; - // For directories - children?: Set; -} - -export class VirtualFileSystem extends EventEmitter implements IFuseAPI { - private nodes: Map = new Map(); - private config: Required; - private metaFormat: MetaMeshFormat = new MetaMeshFormat(); - private formatRegistry: MetadataFormatRegistry = new MetadataFormatRegistry(); - private activeFormatHandlers: IMetadataFormatHandler[] = []; - - // Cached stats - updated incrementally on each modification - private cachedStats = { - fileCount: 0, - directoryCount: 0, - totalSize: 0, - metaFileCount: 0, - metadataFileCount: 0 - }; - - constructor(fuseConfig: FuseAPIConfig = {}) { - super(); - this.config = { - basePath: fuseConfig.basePath ?? '/', - fileMode: fuseConfig.fileMode ?? 0o644, - directoryMode: fuseConfig.directoryMode ?? 0o755, - uid: fuseConfig.uid ?? process.getuid(), - gid: fuseConfig.gid ?? process.getgid() - }; - - // Initialize active format handlers from configuration - this.activeFormatHandlers = this.formatRegistry.getHandlersForFormats(config.METADATA_FORMATS); - - if (this.activeFormatHandlers.length > 0) { - console.log(`VirtualFileSystem: Active metadata formats: ${this.activeFormatHandlers.map(h => h.formatType).join(', ')}`); - } else { - console.log('VirtualFileSystem: No metadata formats configured'); - } - - // Initialize root directory - this.nodes.set('/', { - type: 'directory', - name: '', - parent: null, - children: new Set() - }); - - // Initialize cached stats with root directory - this.cachedStats.directoryCount = 1; - } - - /** - * Build the virtual filesystem from a computed folder structure - * @param computed Map of source paths to destination paths - * @param metadata Map of source paths to HashMeta - * @param basePath Base path to strip from destination paths (e.g., OUTPUT_FOLDER) - */ - public buildFromComputed( - computed: Map, - metadata: Map, - basePath?: string - ): void { - // Clear existing structure (except root) - this.nodes.clear(); - this.nodes.set('/', { - type: 'directory', - name: '', - parent: null, - children: new Set() - }); - - // Reset cached stats (root directory will be counted) - this.cachedStats = { - fileCount: 0, - directoryCount: 1, // Root directory - totalSize: 0, - metaFileCount: 0, - metadataFileCount: 0 - }; - - // Normalize basePath if provided - const normalizedBasePath = basePath ? path.normalize(basePath) : null; - - // Add all files and their parent directories - for (const [sourcePath, destPath] of computed.entries()) { - // Strip basePath from destPath to create relative virtual path - let virtualPath = destPath; - if (normalizedBasePath && destPath.startsWith(normalizedBasePath)) { - virtualPath = destPath.slice(normalizedBasePath.length); - // Ensure path starts with / - if (!virtualPath.startsWith('/')) { - virtualPath = '/' + virtualPath; - } - } - this.addFile(virtualPath, sourcePath, metadata.get(sourcePath)); - } - - this.emit('vfs-updated'); - } - - /** - * Add a file to the virtual filesystem, creating parent directories as needed - */ - public addFile(virtualPath: string, sourcePath: string, metadata?: HashMeta): void { - // Normalize path - virtualPath = this.normalizePath(virtualPath); - - // Create parent directories - const dirname = path.dirname(virtualPath); - this.ensureDirectory(dirname); - - // Try to use metadata for file stats (avoid disk I/O during fast load) - let size: number; - let mtime: Date; - let ctime: Date; - - const extMeta = metadata as ExtendedHashMeta | undefined; - if (extMeta && extMeta.sizeByte) { - // Use cached metadata (fast path - no disk I/O) - size = parseInt(String(extMeta.sizeByte), 10); - // Use current time as placeholder for mtime/ctime (will be corrected on next scan) - mtime = new Date(); - ctime = new Date(); - } else { - // Fallback: stat the file (slow path - disk I/O) - try { - const stats = fs.statSync(sourcePath); - size = stats.size; - mtime = stats.mtime; - ctime = stats.ctime; - } catch (error) { - console.warn(`Cannot stat source file: ${sourcePath}`, error); - return; - } - } - - // Add the file node - const filename = path.basename(virtualPath); - this.nodes.set(virtualPath, { - type: 'file', - name: filename, - parent: dirname, - 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 cached stats - this.cachedStats.fileCount++; - this.cachedStats.totalSize += size; - - // Also add corresponding metadata files based on configured formats - this.addMetadataFiles(virtualPath, metadata, mtime); - - this.emit('file-added', virtualPath); - } - - /** - * Add virtual metadata files for all configured formats - */ - private addMetadataFiles(mediaPath: string, metadata: HashMeta | undefined, mtime: Date): void { - if (!metadata) return; - - // Create a metadata file for each configured format - for (const handler of this.activeFormatHandlers) { - const metaPath = `${mediaPath}${handler.extension}`; - const dirname = path.dirname(metaPath); - const filename = path.basename(metaPath); - - // Pre-generate metadata content to calculate size - const metaContent = handler.serialize(metadata); - const metaBuffer = Buffer.from(metaContent, 'utf-8'); - - // Add the virtual metadata file node - this.nodes.set(metaPath, { - type: 'file', - name: filename, - parent: dirname, - sourcePath: null, // Virtual file, no source - size: metaBuffer.length, - mtime: mtime, - ctime: mtime, - metadata, - formatHandler: handler - }); - - // Add to parent's children - const parentNode = this.nodes.get(dirname); - if (parentNode && parentNode.type === 'directory') { - parentNode.children!.add(filename); - } - - // Update cached stats - this.cachedStats.fileCount++; - this.cachedStats.totalSize += metaBuffer.length; - this.cachedStats.metadataFileCount++; - // Legacy: Count .meta files separately for backward compatibility - if (handler.extension === '.meta') { - this.cachedStats.metaFileCount++; - } - } - } - - /** - * Ensure a directory exists in the VFS, creating parent directories as needed - */ - private ensureDirectory(dirPath: string): void { - dirPath = this.normalizePath(dirPath); - - if (dirPath === '/' || this.nodes.has(dirPath)) { - return; - } - - // Recursively create parent - const parentPath = path.dirname(dirPath); - this.ensureDirectory(parentPath); - - // Create this directory - const dirName = path.basename(dirPath); - this.nodes.set(dirPath, { - type: 'directory', - name: dirName, - parent: parentPath, - children: new Set() - }); - - // Add to parent's children - const parentNode = this.nodes.get(parentPath); - if (parentNode && parentNode.type === 'directory') { - parentNode.children!.add(dirName); - } - - // Update cached stats - this.cachedStats.directoryCount++; - - this.emit('directory-added', dirPath); - } - - /** - * Normalize path to use forward slashes and ensure leading slash - */ - private normalizePath(p: string): string { - // Convert to forward slashes - p = p.split(path.sep).join('/'); - - // Ensure leading slash - if (!p.startsWith('/')) { - p = '/' + p; - } - - // Remove trailing slash (except for root) - if (p !== '/' && p.endsWith('/')) { - p = p.slice(0, -1); - } - - return p; - } - - // IFuseAPI implementation - - public readdir(dirPath: string): string[] | null { - dirPath = this.normalizePath(dirPath); - const node = this.nodes.get(dirPath); - - if (!node || node.type !== 'directory') { - return null; - } - - return Array.from(node.children!); - } - - public getattr(filepath: string): FileAttributes | null { - filepath = this.normalizePath(filepath); - const node = this.nodes.get(filepath); - - if (!node) { - return null; - } - - if (node.type === 'directory') { - return { - size: 4096, // Standard directory size - mode: 0o040000 | this.config.directoryMode, // S_IFDIR | permissions - mtime: Date.now() / 1000, - atime: Date.now() / 1000, - ctime: Date.now() / 1000, - nlink: 2, // . and .. - uid: this.config.uid, - gid: this.config.gid - }; - } else { - return { - size: node.size!, - mode: 0o100000 | this.config.fileMode, // S_IFREG | permissions - mtime: node.mtime!.getTime() / 1000, - atime: node.mtime!.getTime() / 1000, - ctime: node.ctime!.getTime() / 1000, - nlink: 1, - uid: this.config.uid, - gid: this.config.gid - }; - } - } - - public exists(filepath: string): boolean { - filepath = this.normalizePath(filepath); - return this.nodes.has(filepath); - } - - public read(filepath: string): ReadResult | null { - filepath = this.normalizePath(filepath); - const node = this.nodes.get(filepath); - - if (!node || node.type !== 'file') { - return null; - } - - // Virtual metadata file (any format) - if (node.sourcePath === null && node.metadata && node.formatHandler) { - const metaContent = node.formatHandler.serialize(node.metadata); - const metaBuffer = Buffer.from(metaContent, 'utf-8'); - - return { - sourcePath: null, - content: metaBuffer, - size: metaBuffer.length - }; - } - - // Regular file - return source path for reading - return { - sourcePath: node.sourcePath!, - content: null, - size: node.size! - }; - } - - public getMetadata(filepath: string): HashMeta | null { - filepath = this.normalizePath(filepath); - const node = this.nodes.get(filepath); - - if (!node || node.type !== 'file') { - return null; - } - - return node.metadata ?? null; - } - - public getTree(): VirtualDirectory { - const rootNode = this.nodes.get('/')!; - - return { - path: '/', - type: 'directory', - name: '', - parent: '', - children: Array.from(rootNode.children!), - mode: this.config.directoryMode, - mtime: new Date() - }; - } - - public async refresh(): Promise { - // Refresh is triggered by external calls after processing - // The buildFromComputed method handles the actual refresh - this.emit('vfs-updated'); - } - - /** - * Get all file paths in the VFS - */ - public getAllFiles(): string[] { - const files: string[] = []; - for (const [filepath, node] of this.nodes.entries()) { - if (node.type === 'file') { - files.push(filepath); - } - } - return files; - } - - /** - * Get all directory paths in the VFS - */ - public getAllDirectories(): string[] { - const dirs: string[] = []; - for (const [dirpath, node] of this.nodes.entries()) { - if (node.type === 'directory') { - dirs.push(dirpath); - } - } - return dirs; - } - - /** - * Get statistics about the VFS (returns cached values, updated incrementally) - */ - public getStats(): { - fileCount: number; - directoryCount: number; - totalSize: number; - metaFileCount: number; - metadataFileCount: number; - } { - // Return cached stats - no iteration needed! - return { ...this.cachedStats }; - } - - /** - * Count files in VFS that have completed hash computation - * A file is considered "done" if hashStatus === 'complete' (all hashes computed without errors) - */ - public countFilesWithHash(): number { - let count = 0; - for (const [filepath, node] of this.nodes.entries()) { - // Only count actual media files (not directories, not metadata files) - if (node.type === 'file' && node.sourcePath && node.metadata) { - // Check if file has completed processing (both light and hash phases) - const extMeta = node.metadata as ExtendedHashMeta; - if (extMeta.processingStatus === 'complete') { - count++; - } - } - } - return count; - } - - /** - * Update metadata for an existing file in the VFS (e.g., when hash computation completes) - * This also regenerates the metadata files with updated content - */ - public updateFileMetadata(virtualPath: string, metadata: HashMeta): void { - virtualPath = this.normalizePath(virtualPath); - const node = this.nodes.get(virtualPath); - - if (!node || node.type !== 'directory') { - // Update file metadata - if (node) { - node.metadata = metadata; - - // Regenerate metadata files with updated content - if (node.mtime) { - // Get old metadata file sizes before regenerating - const oldSizes = new Map(); - for (const handler of this.activeFormatHandlers) { - const metaPath = `${virtualPath}${handler.extension}`; - const oldMetaNode = this.nodes.get(metaPath); - if (oldMetaNode) { - oldSizes.set(metaPath, oldMetaNode.size ?? 0); - } - } - - // Regenerate metadata files (this will overwrite the nodes) - this.addMetadataFilesUpdate(virtualPath, metadata, node.mtime, oldSizes); - } - - this.emit('file-updated', virtualPath); - } - } - } - - /** - * Add/update virtual metadata files for all configured formats (used during updates) - * This version adjusts cached stats based on size differences - */ - private addMetadataFilesUpdate( - mediaPath: string, - metadata: HashMeta, - mtime: Date, - oldSizes: Map - ): void { - // Create a metadata file for each configured format - for (const handler of this.activeFormatHandlers) { - const metaPath = `${mediaPath}${handler.extension}`; - const dirname = path.dirname(metaPath); - const filename = path.basename(metaPath); - - // Pre-generate metadata content to calculate size - const metaContent = handler.serialize(metadata); - const metaBuffer = Buffer.from(metaContent, 'utf-8'); - const newSize = metaBuffer.length; - - const oldSize = oldSizes.get(metaPath); - const isNew = oldSize === undefined; - - // Update the virtual metadata file node - this.nodes.set(metaPath, { - type: 'file', - name: filename, - parent: dirname, - sourcePath: null, // Virtual file, no source - size: newSize, - mtime: mtime, - ctime: mtime, - metadata, - formatHandler: handler - }); - - // Add to parent's children (no-op if already exists) - const parentNode = this.nodes.get(dirname); - if (parentNode && parentNode.type === 'directory') { - parentNode.children!.add(filename); - } - - // Update cached stats - if (isNew) { - // New metadata file - this.cachedStats.fileCount++; - this.cachedStats.totalSize += newSize; - this.cachedStats.metadataFileCount++; - if (handler.extension === '.meta') { - this.cachedStats.metaFileCount++; - } - } else { - // Update existing - adjust size difference - this.cachedStats.totalSize += (newSize - oldSize); - } - } - } - - /** - * Remove a file and its associated metadata files from the VFS - */ - public removeFile(virtualPath: string): void { - virtualPath = this.normalizePath(virtualPath); - const node = this.nodes.get(virtualPath); - - if (!node || node.type !== 'file') { - return; - } - - // Update cached stats for main file - this.cachedStats.fileCount--; - this.cachedStats.totalSize -= node.size ?? 0; - - // Remove the main file - this.nodes.delete(virtualPath); - - // Remove from parent's children - const parentNode = this.nodes.get(node.parent!); - if (parentNode && parentNode.type === 'directory') { - parentNode.children!.delete(node.name); - } - - // Remove associated metadata files (.nfo, .meta, etc.) - for (const handler of this.activeFormatHandlers) { - const metaPath = `${virtualPath}${handler.extension}`; - const metaNode = this.nodes.get(metaPath); - - if (metaNode) { - // Update cached stats for metadata file - this.cachedStats.fileCount--; - this.cachedStats.totalSize -= metaNode.size ?? 0; - this.cachedStats.metadataFileCount--; - if (handler.extension === '.meta') { - this.cachedStats.metaFileCount--; - } - - this.nodes.delete(metaPath); - - // Remove from parent's children - const metaParent = this.nodes.get(metaNode.parent!); - if (metaParent && metaParent.type === 'directory') { - metaParent.children!.delete(metaNode.name); - } - } - } - - this.emit('file-removed', virtualPath); - } - - /** - * Check if a file exists in the VFS and return its metadata - */ - public getFileMetadata(virtualPath: string): HashMeta | undefined { - virtualPath = this.normalizePath(virtualPath); - const node = this.nodes.get(virtualPath); - return node?.metadata; - } -} - -// Type-safe event emitter -export interface VirtualFileSystem { - on(event: K, listener: FuseAPIEvents[K]): this; - emit(event: K, ...args: Parameters): boolean; -} diff --git a/packages/meta-sort-core/src/api/api.test.ts b/packages/meta-sort-core/src/api/api.test.ts index aea3e09..91e7a9a 100644 --- a/packages/meta-sort-core/src/api/api.test.ts +++ b/packages/meta-sort-core/src/api/api.test.ts @@ -7,7 +7,6 @@ import { expect } from 'chai'; import { UnifiedAPIServer } from './UnifiedAPIServer.js'; -import { VirtualFileSystem } from './VirtualFileSystem.js'; import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js'; import type { IKVClient } from '../kv/IKVClient.js'; import type { FastifyInstance } from 'fastify'; @@ -240,12 +239,8 @@ describe('UnifiedAPIServer', function() { mockStateManager.addDone('/files/test/Test Movie.mp4'); mockStateManager.addDiscovered('/files/test/pending.mp4'); - // Create VFS (no params needed - uses defaults) - const vfs = new VirtualFileSystem(); - // Create server server = new UnifiedAPIServer( - vfs, { port: 3000, host: 'localhost', enableCors: true }, mockStateManager as any, mockKV as any, @@ -276,7 +271,6 @@ describe('UnifiedAPIServer', function() { expect(response.statusCode).to.equal(200); const body = JSON.parse(response.payload); expect(body.status).to.equal('ok'); - expect(body.services).to.have.property('fuse'); expect(body.services).to.have.property('metrics'); }); @@ -416,97 +410,7 @@ describe('UnifiedAPIServer', function() { // NOTE: /api/scan/trigger has been removed from meta-sort (Architecture V3) // Scan is now handled by meta-core. Tests for scan trigger are in core.bats. - // ========================================================================= - // FUSE API (Virtual Filesystem) - // ========================================================================= - - describe('FUSE API (/api/fuse/*)', function() { - it('GET /api/fuse/health returns ok', async function() { - const response = await app.inject({ - method: 'GET', - url: '/api/fuse/health' - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body.status).to.equal('ok'); - }); - - it('GET /api/fuse/stats returns VFS statistics', async function() { - const response = await app.inject({ - method: 'GET', - url: '/api/fuse/stats' - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body).to.have.property('fileCount'); - expect(body).to.have.property('directoryCount'); - }); - - it('POST /api/fuse/readdir returns directory contents', async function() { - const response = await app.inject({ - method: 'POST', - url: '/api/fuse/readdir', - payload: { path: '/' } - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body).to.have.property('entries'); - expect(Array.isArray(body.entries)).to.be.true; - }); - - it('POST /api/fuse/exists checks path existence', async function() { - const response = await app.inject({ - method: 'POST', - url: '/api/fuse/exists', - payload: { path: '/' } - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body).to.have.property('exists'); - expect(typeof body.exists).to.equal('boolean'); - }); - - it('GET /api/fuse/tree returns full VFS tree', async function() { - const response = await app.inject({ - method: 'GET', - url: '/api/fuse/tree' - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - // API returns tree structure directly (path, type, children) - expect(body).to.have.property('path'); - expect(body).to.have.property('type'); - }); - - it('GET /api/fuse/files returns all file paths', async function() { - const response = await app.inject({ - method: 'GET', - url: '/api/fuse/files' - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body).to.have.property('files'); - expect(Array.isArray(body.files)).to.be.true; - }); - - it('GET /api/fuse/directories returns all directory paths', async function() { - const response = await app.inject({ - method: 'GET', - url: '/api/fuse/directories' - }); - - expect(response.statusCode).to.equal(200); - const body = JSON.parse(response.payload); - expect(body).to.have.property('directories'); - expect(Array.isArray(body.directories)).to.be.true; - }); - }); + // NOTE: FUSE API (/api/fuse/*) has been moved to meta-fuse service. // ========================================================================= // Plugins API (Plugins Page) diff --git a/packages/meta-sort-core/src/api/index.ts b/packages/meta-sort-core/src/api/index.ts index a27cd20..59d32ab 100644 --- a/packages/meta-sort-core/src/api/index.ts +++ b/packages/meta-sort-core/src/api/index.ts @@ -1,11 +1,8 @@ /** * API exports * - * This module exports all API components including FUSE API, Metrics API, and Unified API Server. + * This module exports all API components including Metrics API and Unified API Server. */ -export { IFuseAPI, VirtualNode, VirtualFile, VirtualDirectory, FileAttributes, ReadResult, FuseAPIConfig, FuseAPIEvents } from './FuseAPI.js'; -export { VirtualFileSystem } from './VirtualFileSystem.js'; -export { FuseAPIServer, FuseAPIServerConfig } from './FuseAPIServer.js'; export { MetricsAPI, MetricsAPIConfig } from './MetricsAPI.js'; export { UnifiedAPIServer, UnifiedAPIServerConfig } from './UnifiedAPIServer.js'; diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 9f1c702..f7554ff 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -489,6 +489,9 @@ export class ContainerPluginScheduler extends EventEmitter { /** * Check if all tasks for a file are complete + * + * Note: Multiple files may share the same fileHash (e.g., duplicate files). + * When this happens, we need to emit file:complete for EACH unique filePath. */ private checkFileCompletion(fileHash: string): void { const taskIds = this.fileTasks.get(fileHash); @@ -497,7 +500,7 @@ export class ContainerPluginScheduler extends EventEmitter { } let allComplete = true; - let filePath: string | undefined; + const uniqueFilePaths = new Set(); for (const taskId of taskIds) { const task = this.pendingTasks.get(taskId); @@ -505,7 +508,8 @@ export class ContainerPluginScheduler extends EventEmitter { continue; } - filePath = task.filePath; + // Collect all unique filePaths that share this hash + uniqueFilePaths.add(task.filePath); if ( task.status !== 'completed' && @@ -517,7 +521,7 @@ export class ContainerPluginScheduler extends EventEmitter { } } - if (allComplete && filePath) { + if (allComplete && uniqueFilePaths.size > 0) { // Clean up task tracking for (const taskId of taskIds) { this.pendingTasks.delete(taskId); @@ -533,7 +537,10 @@ export class ContainerPluginScheduler extends EventEmitter { } } - this.emit('file:complete', { fileHash, filePath }); + // Emit file:complete for EACH unique filePath (handles duplicate files with same hash) + for (const filePath of uniqueFilePaths) { + this.emit('file:complete', { fileHash, filePath }); + } } } diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index b31e94c..a000c96 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -65,15 +65,13 @@ const pipeline = new StreamingPipeline({ fileProcessor: fileProcessor.fileProcessor, stateManager: fileProcessor.unifiedStateManager, kvClient: fileProcessor.getKVClient(), - virtualFileSystem: fileProcessor.getVirtualFileSystem(), metaDataToFolderStruct: fileProcessor.metaDataToFolderStruct }); // Connect pipeline to fileProcessor for queue status monitoring fileProcessor.setPipeline(pipeline); -// VFS and state manager are available immediately -const vfs = fileProcessor.getVirtualFileSystem(); +// State manager is available immediately const unifiedStateManager = fileProcessor.getUnifiedStateManager(); // API Server will be initialized after KV is ready (deferred to async init) @@ -160,7 +158,7 @@ process.on('SIGINT', async () => { // Start Unified API Server AFTER KV is ready (so kvClient is available) const kvClient = fileProcessor.getKVClient(); - apiServer = new UnifiedAPIServer(vfs, { + apiServer = new UnifiedAPIServer({ port: config.FUSE_API_PORT, host: config.FUSE_API_HOST, enableCors: true, @@ -248,11 +246,6 @@ process.on('SIGINT', async () => { apiServer.setStreamingPipeline(pipeline); console.log('[Startup] StreamingPipeline connected to API server'); - // First, rebuild VFS from KV to restore state from previous runs - console.log('[Startup] Rebuilding VirtualFileSystem from KV...'); - await fileProcessor.rebuildVFSFromKV(); - console.log('[Startup] VFS rebuild completed'); - // Log cache performance summary const metrics = performanceMetrics.getMetrics(); if (metrics.cacheHits.midhash256 || metrics.cacheMisses.midhash256) { diff --git a/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts b/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts index fd55b97..630104b 100644 --- a/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts +++ b/packages/meta-sort-core/src/logic/UnifiedProcessingStateManager.ts @@ -51,7 +51,6 @@ export interface UnifiedProcessingSnapshot { totalLightProcessing: number; totalHashProcessing: number; totalDone: number; - totalFilesInVFS: number; // Total number of files accessible in the Virtual File System /** Files that completed validation but are waiting for fast queue */ awaitingFastQueue: number; @@ -217,11 +216,10 @@ export class UnifiedProcessingStateManager { /** * Get a snapshot of the current processing state - * @param vfsFileCount - Optional total number of files in the Virtual File System * @param fastQueueConcurrency - Number of fast queue workers * @param backgroundQueueConcurrency - Number of background queue workers */ - getSnapshot(vfsFileCount: number = 0, fastQueueConcurrency?: number, backgroundQueueConcurrency?: number): UnifiedProcessingSnapshot { + getSnapshot(fastQueueConcurrency?: number, backgroundQueueConcurrency?: number): UnifiedProcessingSnapshot { // Limit returned arrays to prevent slow API responses with large queues const MAX_ITEMS_PER_STATE = 100; @@ -269,7 +267,6 @@ export class UnifiedProcessingStateManager { totalLightProcessing: this.lightProcessing.size, totalHashProcessing: this.hashProcessing.size, totalDone: this.totalProcessedCount, // Total count of all processed files (not limited to history) - totalFilesInVFS: vfsFileCount, // Total number of files accessible in the Virtual File System // Files that completed validation but are waiting for fast queue (set by API layer from queue status) awaitingFastQueue: 0, diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index eefa1fa..2444e1f 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -8,11 +8,10 @@ import {FileProcessorPiscina} from "./fileProcessor/FileProcessorPiscina.js"; import {renamingRule} from "../config/RenamingRule.js"; import {MetaMeshFormat} from "./MetaMeshFormat.js"; import {JellyfinAPI} from "../jellyfin/JellyfinAPI.js"; -import {VirtualFileSystem} from "../api/VirtualFileSystem.js"; import {performanceMetrics} from "../metrics/PerformanceMetrics.js"; import {UnifiedProcessingStateManager} from "./UnifiedProcessingStateManager.js"; import type {IKVClient} from "../kv/IKVClient.js"; -import { type ExtendedFileAnalyzer, type ExtendedHashMeta } from "../types/ExtendedInterfaces.js"; +import { type ExtendedFileAnalyzer } from "../types/ExtendedInterfaces.js"; import * as webdav from '../webdav/WebdavClient.js'; import PQueue from "p-queue"; import os from 'os'; @@ -25,7 +24,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { jellyfinAPI = new JellyfinAPI(); metaMeshFormat = new MetaMeshFormat(); metaDataToFolderStruct = new MetaDataToFolderStruct(renamingRule); - virtualFileSystem: VirtualFileSystem; unifiedStateManager = new UnifiedProcessingStateManager(); fileProcessor:FileAnalyzerInterface = new FileProcessorPiscina(); fileType = new FileType(); @@ -41,12 +39,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { private pipeline: StreamingPipeline | null = null; constructor() { - // Always initialize VirtualFileSystem for FUSE API - this.virtualFileSystem = new VirtualFileSystem({ - fileMode: 0o644, - directoryMode: config.CHMOD_FOR_NEW_FOLDER || 0o755 - }); - // Pass the unified state manager to the fileProcessor if (this.fileProcessor instanceof FileProcessorPiscina) { this.fileProcessor.setUnifiedStateManager(this.unifiedStateManager); @@ -62,10 +54,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { console.log(` - TaskScheduler handles plugin execution (fast/background queues)`); } - getVirtualFileSystem(): VirtualFileSystem { - return this.virtualFileSystem; - } - getUnifiedStateManager(): UnifiedProcessingStateManager { return this.unifiedStateManager; } @@ -193,7 +181,7 @@ export class WatchedFileProcessor implements FileProcessorInterface { fastQueueConcurrency: config.FAST_QUEUE_CONCURRENCY, }); - // Set up event listeners for VFS updates + // Set up event listeners for state manager updates this.setupTaskSchedulerEvents(); console.log(`[WatchedFileProcessor] TaskScheduler initialized with:`); @@ -201,30 +189,17 @@ export class WatchedFileProcessor implements FileProcessorInterface { } /** - * Set up TaskScheduler event listeners for VFS integration + * Set up TaskScheduler event listeners * Note: Event publishing to meta-fuse is handled by meta-core (file:events stream) */ private setupTaskSchedulerEvents(): void { if (!this.taskScheduler) return; - // When a task completes, check if we should update VFS - this.taskScheduler.on('task:complete', (event: any) => { - const { task } = event; - - // Update VFS when filename-parser completes (has basic metadata) - if (task.pluginId === 'filename-parser') { - this.updateVFSForFile(task.fileHash, task.filePath); - } - }); - // When all tasks for a file complete this.taskScheduler.on('file:complete', (event: any) => { const { fileHash, filePath } = event; console.log(`[TaskScheduler] All plugins complete for ${filePath}`); - // Final VFS update with complete metadata - this.updateVFSForFile(fileHash, filePath); - // Update unified state manager if (this.unifiedStateManager) { this.unifiedStateManager.completeHashProcessing(filePath, fileHash); @@ -232,21 +207,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { }); } - /** - * Update VFS for a file with current metadata - */ - private updateVFSForFile(fileHash: string, filePath: string): void { - const metadata = this.fileProcessor.getDatabase().get(filePath); - if (metadata && metadata.cid_midhash256) { - try { - const virtualPath = this.metaDataToFolderStruct.renamingRule(metadata as any, ''); - this.virtualFileSystem.addFile(virtualPath, filePath, metadata); - } catch (e) { - // Renaming rule may fail for incomplete metadata - } - } - } - markDiscovered(filePath: string): void { // STATE 1: Mark as discovered when file is found (before queue) this.unifiedStateManager.addDiscovered(filePath); @@ -274,10 +234,32 @@ export class WatchedFileProcessor implements FileProcessorInterface { // STEP 1: Compute midhash256 (file identifier) // Use WebDAV for file stats and hash computation const fileStats = await webdav.stat(filePath); - if (!fileStats || !fileStats.exists) { - console.error(`[WatchedFileProcessor] File not found via WebDAV: ${filePath}`); + if (!fileStats.exists) { + // Generate specific error message based on error type + const errorType = fileStats.error || 'unknown'; + let errorMsg: string; + switch (errorType) { + case 'timeout': + errorMsg = 'WebDAV request timeout'; + break; + case 'not_configured': + errorMsg = 'WebDAV client not configured'; + break; + case 'not_found': + errorMsg = 'File not found'; + break; + case 'network_error': + errorMsg = 'WebDAV network error'; + break; + case 'http_error': + errorMsg = `WebDAV HTTP error (${fileStats.statusCode})`; + break; + default: + errorMsg = 'WebDAV error'; + } + console.error(`[WatchedFileProcessor] ${errorMsg}: ${filePath}`); if (this.unifiedStateManager) { - this.unifiedStateManager.completeLightProcessing(filePath, '', undefined, 'File not found'); + this.unifiedStateManager.completeLightProcessing(filePath, '', undefined, errorMsg); } return; } @@ -370,7 +352,7 @@ export class WatchedFileProcessor implements FileProcessorInterface { async finalize(): Promise { try { - console.log(`Virtual filesystem update start`); + console.log(`Finalize processing start`); const start = performance.now(); if (this.fileProcessor.getDatabase().size === 0) { console.log(`No files to process`); @@ -378,29 +360,21 @@ export class WatchedFileProcessor implements FileProcessorInterface { } // NOTE: Duplicate detection has been moved to meta-dup service + // NOTE: VFS building has been moved to meta-fuse service - let startTime = performance.now(); + const startTime = performance.now(); console.log(`Generating virtual structure for ${this.fileProcessor.getDatabase().size} files`); - const virtualStructure = this.metaDataToFolderStruct.generateVirtualStructure(this.fileProcessor.getDatabase()); + this.metaDataToFolderStruct.generateVirtualStructure(this.fileProcessor.getDatabase()); const virtualStructureTime = Math.ceil(performance.now() - startTime); console.log(`Virtual structure generation took ${virtualStructureTime}ms`); performanceMetrics.recordVirtualStructureGeneration(virtualStructureTime); - // Build VirtualFileSystem for FUSE/WebDAV API - startTime = performance.now(); - this.virtualFileSystem.buildFromComputed(virtualStructure, this.fileProcessor.getDatabase()); - const vfsBuildTime = Math.ceil(performance.now() - startTime); - console.log(`VirtualFileSystem build took ${vfsBuildTime}ms`); - performanceMetrics.recordVFSUpdate(vfsBuildTime); - const stats = this.virtualFileSystem.getStats(); - console.log(`VirtualFileSystem: ${stats.fileCount} files (${stats.metaFileCount} .meta), ${stats.directoryCount} directories, ${(stats.totalSize / 1024 / 1024).toFixed(2)} MB`); - if(this.jellyfinAPI.jellyApiAvailable()) { await this.jellyfinAPI.refreshLibrary(); console.log(`Jellyfin refreshed`); } - console.log(`Virtual filesystem update took ${Math.ceil(performance.now() - start)}ms`); + console.log(`Finalize processing took ${Math.ceil(performance.now() - start)}ms`); } catch (error) { console.error(error); } @@ -472,12 +446,12 @@ export class WatchedFileProcessor implements FileProcessorInterface { const elapsed = Math.ceil(performance.now() - startTime); console.log(`[Cleanup] Removed ${removedCount} stale entries in ${elapsed}ms`); - // If we removed files, rebuild VFS to update structure + // If we removed files, finalize processing and notify Stremio if (removedCount > 0) { - console.log('[Cleanup] Rebuilding VFS after cleanup...'); + console.log('[Cleanup] Finalizing after cleanup...'); await this.finalize(); - // Notify Stremio addon to remove videos (incremental updates like FUSE) + // Notify Stremio addon to remove videos if (removedHashIds.length > 0) { console.log(`[Cleanup] Notifying Stremio to remove ${removedHashIds.length} videos...`); for (const hashId of removedHashIds) { @@ -490,78 +464,7 @@ export class WatchedFileProcessor implements FileProcessorInterface { } /** - * Rebuild VirtualFileSystem from KV data - * This is called on startup to restore the VFS from persistent storage - * - * Fast-load strategy: Load all metadata from KV WITHOUT disk validation. - * Discovery will later validate file existence and clean up stale entries. - * - * Design Philosophy: - * - KV = distributed metadata archive (permanent record) - * - In-memory DB = current VFS state (only present files) - * - Discovery = source of truth (validates and cleans) - */ - async rebuildVFSFromKV(): Promise { - const kvClient = this.getKVClient(); - if (!kvClient) { - console.warn('[VFS Rebuild] KV client not available, skipping VFS rebuild'); - return; - } - - try { - console.log('[VFS Rebuild] Starting fast-load from KV...'); - const startTime = performance.now(); - - // Get all hash IDs from KV - const hashIds = await kvClient.getAllHashIds(); - console.log(`[VFS Rebuild] Found ${hashIds.length} files in KV`); - - if (hashIds.length === 0) { - console.log('[VFS Rebuild] No files in KV, VFS remains empty'); - return; - } - - // Batch fetch all metadata in parallel (no disk validation) - console.log('[VFS Rebuild] Fetching metadata in parallel...'); - const metadataList = await Promise.all( - hashIds.map(async (hashId) => { - try { - return await kvClient.getMetadataFlat(hashId); - } catch (error) { - console.error(`[VFS Rebuild] Failed to load metadata for ${hashId}:`, error); - return null; - } - }) - ); - - // Populate in-memory database without disk validation - let loadedCount = 0; - for (const metadata of metadataList) { - if (metadata && metadata.filePath) { - // Mark as unverified - discovery will validate later - (metadata as ExtendedHashMeta)._lastVerified = 0; - this.fileProcessor.getDatabase().set(metadata.filePath, metadata); - loadedCount++; - } - } - - console.log(`[VFS Rebuild] Loaded ${loadedCount}/${hashIds.length} files into memory (unverified)`); - - // Build VFS immediately - may contain stale entries until discovery validates - if (this.fileProcessor.getDatabase().size > 0) { - await this.finalize(); - console.log(`[VFS Rebuild] VFS rebuilt in ${Math.ceil(performance.now() - startTime)}ms`); - console.log('[VFS Rebuild] Discovery will validate file existence in background'); - } else { - console.log('[VFS Rebuild] No files loaded, VFS remains empty'); - } - } catch (error) { - console.error('[VFS Rebuild] Error rebuilding VFS from KV:', error); - } - } - - /** - * Notify Stremio addon to remove video (incremental update like FUSE) + * Notify Stremio addon to remove video */ private async notifyStremioRemove(hashId: string): Promise { try { diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts index 97d36af..5b62dd1 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts @@ -161,8 +161,30 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ } const stats = await webdav.stat(filePath); - if (!stats || !stats.exists) { - throw new Error(`File not found via WebDAV: ${filePath}`); + if (!stats.exists) { + // Generate specific error message based on error type + const errorType = stats.error || 'unknown'; + let errorMsg: string; + switch (errorType) { + case 'timeout': + errorMsg = `WebDAV request timeout for: ${filePath}`; + break; + case 'not_configured': + errorMsg = `WebDAV client not configured, cannot access: ${filePath}`; + break; + case 'not_found': + errorMsg = `File not found via WebDAV: ${filePath}`; + break; + case 'network_error': + errorMsg = `WebDAV network error for: ${filePath}`; + break; + case 'http_error': + errorMsg = `WebDAV HTTP error (${stats.statusCode}) for: ${filePath}`; + break; + default: + errorMsg = `WebDAV error for: ${filePath}`; + } + throw new Error(errorMsg); } // STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files) @@ -348,23 +370,46 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ this.unifiedStateManager.startLightProcessing(filePath); } - const stats = await webdav.stat(filePath); - if (!stats || !stats.exists) { - throw new Error(`File not found via WebDAV: ${filePath}`); - } - // STEP 1: Use midhash256 from meta-core if provided, otherwise check cache/compute locally let midHash256: string | undefined; if (providedMidHash256) { - // midhash256 provided by meta-core - use directly (no file I/O needed) + // midhash256 provided by meta-core - use directly (no WebDAV I/O needed!) + // meta-core already verified the file exists when it computed the hash midHash256 = providedMidHash256; performanceMetrics.recordCacheHit('midhash256'); - - // Store in local cache for future lookups - this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 }); + // Note: We skip cache storage here since we don't have file stats (size/mtime) + // This is fine because meta-core will provide the hash again on rescan } else { - // Fallback: Check cache or compute locally (legacy path) + // Fallback: Need to access file via WebDAV (legacy path without meta-core events) + const stats = await webdav.stat(filePath); + if (!stats.exists) { + // Generate specific error message based on error type + const errorType = stats.error || 'unknown'; + let errorMsg: string; + switch (errorType) { + case 'timeout': + errorMsg = `WebDAV request timeout for: ${filePath}`; + break; + case 'not_configured': + errorMsg = `WebDAV client not configured, cannot access: ${filePath}`; + break; + case 'not_found': + errorMsg = `File not found via WebDAV: ${filePath}`; + break; + case 'network_error': + errorMsg = `WebDAV network error for: ${filePath}`; + break; + case 'http_error': + errorMsg = `WebDAV HTTP error (${stats.statusCode}) for: ${filePath}`; + break; + default: + errorMsg = `WebDAV error for: ${filePath}`; + } + throw new Error(errorMsg); + } + + // Check cache or compute locally const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); if (indexLine && indexLine.cid_midhash256) { @@ -475,10 +520,8 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ this.unifiedStateManager.startHashProcessing(filePath); } - const stats = await webdav.stat(filePath); - if (!stats || !stats.exists) { - throw new Error(`File not found via WebDAV: ${filePath}`); - } + // Note: We skip WebDAV stat here - file existence was verified during light phase + // If file was deleted, the container plugin will fail when accessing it const metadata = this.database.get(filePath); if (!metadata) { throw new Error('Metadata not found - light processing may have failed'); diff --git a/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts b/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts index 7d3b94d..a36c3db 100644 --- a/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts +++ b/packages/meta-sort-core/src/logic/pipeline/PipelineConfig.ts @@ -1,7 +1,6 @@ import {FileAnalyzerInterface} from "../fileProcessor/FileAnalyzerInterface.js"; import {UnifiedProcessingStateManager} from "../UnifiedProcessingStateManager.js"; import type {IKVClient} from "../../kv/IKVClient.js"; -import {VirtualFileSystem} from "../../api/VirtualFileSystem.js"; import {MetaDataToFolderStruct} from "../MetaDataToFolderStruct.js"; import type {ContainerPluginScheduler} from "../../container-plugins/ContainerPluginScheduler.js"; @@ -53,11 +52,6 @@ export interface PipelineConfig { */ kvClient?: IKVClient; - /** - * Virtual filesystem (for instant file visibility) - */ - virtualFileSystem: VirtualFileSystem; - /** * Metadata to folder structure converter (for virtual path generation) */ diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index dafb590..7b626cb 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -162,26 +162,13 @@ export class StreamingPipeline { await this.config.fileProcessor.processLightPhase(filePath, current, queueSize, midhash256); - // After light processing, add to VFS + // After light processing, dispatch to plugins and queue background tasks const metadata = this.config.fileProcessor.getDatabase().get(filePath); if (metadata && metadata.cid_midhash256) { - // Generate virtual path from metadata - const virtualPath = this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath); - - if (virtualPath) { - // Add to VFS immediately (file appears with permanent midhash256 ID) - this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata); - - // Notify Stremio addon to add video (incremental update like FUSE) - if (metadata.cid_midhash256) { - this.notifyStremioAdd(metadata.cid_midhash256).catch(err => { - // Silent fail - Stremio addon may not be running - }); - } - } else { - // File cannot be added to VFS (e.g., unsupported type, missing title) - console.log(`⚠ File skipped from VFS (no virtual path): ${filePath}`); - } + // Notify Stremio addon to add video + this.notifyStremioAdd(metadata.cid_midhash256).catch(err => { + // Silent fail - Stremio addon may not be running + }); // Dispatch tasks to container plugins (fire and forget) this.dispatchContainerPluginTasks(filePath, metadata.cid_midhash256, metadata); diff --git a/packages/meta-sort-core/src/metrics/PerformanceMetrics.ts b/packages/meta-sort-core/src/metrics/PerformanceMetrics.ts index 334f20f..5c735be 100644 --- a/packages/meta-sort-core/src/metrics/PerformanceMetrics.ts +++ b/packages/meta-sort-core/src/metrics/PerformanceMetrics.ts @@ -59,12 +59,6 @@ export interface PerformanceMetricsData { // Per-plugin timing metrics pluginProcessingTimes: Record; - // VFS update metrics - totalVFSUpdates: number; - averageVFSUpdateTime: number; - lastVFSUpdateTime: number; - recentVFSUpdateTimes: MetricEntry[]; - // Duplicate detection metrics averageDuplicateDetectionTime: number; lastDuplicateDetectionTime: number; @@ -121,12 +115,6 @@ export class PerformanceMetrics { private hashProcessingMin: number = Infinity; private hashProcessingMax: number = 0; - // VFS update tracking - running statistics - private recentVFSUpdates: MetricEntry[] = []; - private totalVFSUpdates: number = 0; - private vfsUpdateSum: number = 0; - private lastVFSUpdateTime: number = 0; - // Duplicate detection tracking - running statistics private totalDuplicatesFound: number = 0; private duplicateDetectionSum: number = 0; @@ -222,28 +210,6 @@ export class PerformanceMetrics { this.lastActivityTime = Date.now(); } - /** - * Record a VFS update event - */ - recordVFSUpdate(timeMs: number): void { - // Update running statistics - this.vfsUpdateSum += timeMs; - this.lastVFSUpdateTime = timeMs; - this.totalVFSUpdates++; - - // Keep recent list for display only - this.recentVFSUpdates.push({ - timestamp: Date.now(), - value: timeMs - }); - - if (this.recentVFSUpdates.length > this.maxRecentEntries) { - this.recentVFSUpdates.shift(); - } - - this.lastActivityTime = Date.now(); - } - /** * Record a duplicate detection event */ @@ -447,14 +413,6 @@ export class PerformanceMetrics { return acc; }, {} as Record), - // VFS update metrics - O(1) from running statistics - totalVFSUpdates: this.totalVFSUpdates, - averageVFSUpdateTime: this.totalVFSUpdates > 0 - ? this.vfsUpdateSum / this.totalVFSUpdates - : 0, - lastVFSUpdateTime: this.lastVFSUpdateTime, - recentVFSUpdateTimes: this.recentVFSUpdates.slice(-10), // Last 10 updates - // Duplicate detection metrics - O(1) from running statistics averageDuplicateDetectionTime: this.duplicateDetectionCount > 0 ? this.duplicateDetectionSum / this.duplicateDetectionCount @@ -541,12 +499,6 @@ export class PerformanceMetrics { this.hashProcessingMin = Infinity; this.hashProcessingMax = 0; - // Reset VFS update metrics - this.recentVFSUpdates = []; - this.totalVFSUpdates = 0; - this.vfsUpdateSum = 0; - this.lastVFSUpdateTime = 0; - // Reset duplicate detection metrics this.totalDuplicatesFound = 0; this.duplicateDetectionSum = 0; diff --git a/packages/meta-sort-core/src/types/ExtendedInterfaces.ts b/packages/meta-sort-core/src/types/ExtendedInterfaces.ts index cb3bd23..0a77ad8 100644 --- a/packages/meta-sort-core/src/types/ExtendedInterfaces.ts +++ b/packages/meta-sort-core/src/types/ExtendedInterfaces.ts @@ -91,9 +91,6 @@ export interface ExtendedHashMeta extends HashMeta { /** Source file path */ filePath?: string; - - /** Last verification timestamp (0 = unverified, used during VFS rebuild) */ - _lastVerified?: number; } // ============================================================================= diff --git a/packages/meta-sort-core/src/webdav/WebdavClient.ts b/packages/meta-sort-core/src/webdav/WebdavClient.ts index 234a13f..10a17cf 100644 --- a/packages/meta-sort-core/src/webdav/WebdavClient.ts +++ b/packages/meta-sort-core/src/webdav/WebdavClient.ts @@ -75,6 +75,11 @@ export function toWebdavUrl(filePath: string): string | null { return WEBDAV_URL + encodedPath; } +/** + * Error types for WebDAV operations + */ +export type WebdavErrorType = 'not_found' | 'timeout' | 'not_configured' | 'network_error' | 'http_error'; + /** * File stats returned from WebDAV HEAD request. */ @@ -82,19 +87,23 @@ export interface WebdavFileStats { size: number; mtime: Date; exists: boolean; + /** Error type if the request failed */ + error?: WebdavErrorType; + /** HTTP status code if applicable */ + statusCode?: number; } /** * Get file stats via HTTP HEAD request to WebDAV. * * @param filePath - File path - * @returns File stats or null on error + * @returns File stats with error information if failed */ -export async function stat(filePath: string): Promise { +export async function stat(filePath: string): Promise { const url = toWebdavUrl(filePath); if (!url) { console.error('[webdav-client] Not configured, cannot stat file'); - return null; + return { size: 0, mtime: new Date(0), exists: false, error: 'not_configured' }; } try { @@ -104,12 +113,12 @@ export async function stat(filePath: string): Promise { }); if (response.status === 404) { - return { size: 0, mtime: new Date(0), exists: false }; + return { size: 0, mtime: new Date(0), exists: false, error: 'not_found', statusCode: 404 }; } if (!response.ok) { console.error(`[webdav-client] HEAD error for ${url}: ${response.status}`); - return null; + return { size: 0, mtime: new Date(0), exists: false, error: 'http_error', statusCode: response.status }; } const contentLength = response.headers.get('Content-Length'); @@ -120,9 +129,20 @@ export async function stat(filePath: string): Promise { mtime: lastModified ? new Date(lastModified) : new Date(), exists: true }; - } catch (error) { + } catch (error: any) { + // Distinguish timeout errors from other network errors + const isTimeout = error?.name === 'TimeoutError' || + error?.code === 'ABORT_ERR' || + error?.message?.includes('timeout') || + error?.message?.includes('aborted'); + + if (isTimeout) { + console.error(`[webdav-client] HEAD timeout for ${url}`); + return { size: 0, mtime: new Date(0), exists: false, error: 'timeout' }; + } + console.error(`[webdav-client] HEAD error for ${url}:`, error); - return null; + return { size: 0, mtime: new Date(0), exists: false, error: 'network_error' }; } } @@ -134,7 +154,7 @@ export async function stat(filePath: string): Promise { */ export async function exists(filePath: string): Promise { const stats = await stat(filePath); - return stats?.exists ?? false; + return stats.exists; } /** diff --git a/packages/meta-sort-ui/src/pages/Monitor.tsx b/packages/meta-sort-ui/src/pages/Monitor.tsx index 85a0849..c0084fc 100644 --- a/packages/meta-sort-ui/src/pages/Monitor.tsx +++ b/packages/meta-sort-ui/src/pages/Monitor.tsx @@ -54,33 +54,6 @@ function Monitor() { return () => clearInterval(interval); }, [fetchData]); - const triggerScan = async () => { - try { - await fetch('/api/scan/trigger', { method: 'POST' }); - fetchData(); - } catch (err) { - console.error('Failed to trigger scan:', err); - } - }; - - const stopProcessing = async () => { - try { - await fetch('/api/processing/stop', { method: 'POST' }); - fetchData(); - } catch (err) { - console.error('Failed to stop processing:', err); - } - }; - - const startProcessing = async () => { - try { - await fetch('/api/processing/start', { method: 'POST' }); - fetchData(); - } catch (err) { - console.error('Failed to start processing:', err); - } - }; - const retryFile = async (filePath: string) => { try { await fetch('/api/processing/retry', { @@ -103,25 +76,6 @@ function Monitor() { } }; - const clearMetadata = async () => { - if (!window.confirm('Are you sure you want to clear all metadata? This will delete all file metadata and trigger a full rescan.')) { - return; - } - try { - const response = await fetch('/api/metadata/clear', { method: 'POST' }); - const data = await response.json(); - if (response.ok) { - alert(`Cleared ${data.deletedCount} files. Rescan triggered.`); - } else { - alert(`Failed to clear metadata: ${data.message}`); - } - fetchData(); - } catch (err) { - console.error('Failed to clear metadata:', err); - alert('Failed to clear metadata'); - } - }; - // Calculate progress for 6-stage pipeline (including failed) // Discovered = files waiting for fast queue (validated but not yet processing) const getProgress = () => { @@ -193,23 +147,6 @@ function Monitor() {

Processing Monitor

-
- {status?.computed?.gateOpen === false ? ( - - ) : ( - - )} - - -
{/* Pipeline Progress Bar - 6 Stage (including failed) */} @@ -574,30 +511,6 @@ function Monitor() { font-size: 1.8rem; } - .header-actions { - display: flex; - gap: 12px; - } - - .btn-success { - background: linear-gradient(135deg, #10b981, #059669); - color: white; - } - - .btn-success:hover { - background: linear-gradient(135deg, #059669, #047857); - } - - .btn-danger { - background: linear-gradient(135deg, #ef4444, #dc2626); - color: white; - } - - .btn-danger:hover { - background: linear-gradient(135deg, #dc2626, #b91c1c); - } - - /* Pipeline Progress Bar */ .pipeline-section { margin-bottom: 24px; diff --git a/packages/meta-sort-ui/src/pages/Plugins.tsx b/packages/meta-sort-ui/src/pages/Plugins.tsx index 05c3c07..20b3809 100644 --- a/packages/meta-sort-ui/src/pages/Plugins.tsx +++ b/packages/meta-sort-ui/src/pages/Plugins.tsx @@ -21,7 +21,6 @@ function Plugins() { const [configEditing, setConfigEditing] = useState | null>(null); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); - const [scanStatus, setScanStatus] = useState(null); // Container plugin management state const [showAddPluginModal, setShowAddPluginModal] = useState(false); @@ -115,18 +114,6 @@ function Plugins() { } }; - const triggerScan = async () => { - setScanStatus('Scanning...'); - try { - await fetch('/api/scan/trigger', { method: 'POST' }); - setScanStatus('Scan triggered'); - setTimeout(() => setScanStatus(null), 3000); - } catch (err) { - console.error('Failed to trigger scan:', err); - setScanStatus('Scan failed'); - } - }; - const rescanPlugins = async () => { setLoading(true); try { @@ -344,13 +331,6 @@ function Plugins() { > Rescan Plugins -