From 30d156494c369ee1ac91572727ce0729f153fb30 Mon Sep 17 00:00:00 2001 From: Pierre Date: Sat, 17 Jan 2026 19:59:53 +0100 Subject: [PATCH] add api test and bugfix --- packages/meta-sort-core/.mocharc.json | 5 + packages/meta-sort-core/package.json | 2 +- packages/meta-sort-core/src/api/api.test.ts | 1042 +++++++++++++++++ .../meta-sort-core/src/config/EnvConfig.ts | 12 +- .../ContainerPluginScheduler.ts | 30 +- packages/meta-sort-core/src/index.ts | 12 +- .../src/logic/WatchedFileProcessor.ts | 4 - .../src/logic/pipeline/StreamingPipeline.ts | 135 +-- .../src/plugin-engine/TaskScheduler.ts | 26 +- packages/meta-sort-core/tsconfig.test.json | 8 + 10 files changed, 1159 insertions(+), 117 deletions(-) create mode 100644 packages/meta-sort-core/.mocharc.json create mode 100644 packages/meta-sort-core/src/api/api.test.ts create mode 100644 packages/meta-sort-core/tsconfig.test.json diff --git a/packages/meta-sort-core/.mocharc.json b/packages/meta-sort-core/.mocharc.json new file mode 100644 index 0000000..22072d8 --- /dev/null +++ b/packages/meta-sort-core/.mocharc.json @@ -0,0 +1,5 @@ +{ + "spec": "dist/api/api.test.js", + "timeout": 30000, + "exit": true +} diff --git a/packages/meta-sort-core/package.json b/packages/meta-sort-core/package.json index 2ae8d1b..985f81e 100644 --- a/packages/meta-sort-core/package.json +++ b/packages/meta-sort-core/package.json @@ -9,7 +9,7 @@ "dev": "tsc --watch", "start": "node dist/index.js", "exec": "node dist/index.js", - "test": "mocha", + "test": "tsc -p tsconfig.test.json && mocha", "lint": "eslint src --ext ts" }, "license": "MIT", diff --git a/packages/meta-sort-core/src/api/api.test.ts b/packages/meta-sort-core/src/api/api.test.ts new file mode 100644 index 0000000..4666f83 --- /dev/null +++ b/packages/meta-sort-core/src/api/api.test.ts @@ -0,0 +1,1042 @@ +/** + * API Integration Tests for UnifiedAPIServer + * + * Tests the UI-related API endpoints using Fastify's inject() method. + * These tests run inside the meta-sort container with a real Redis instance. + */ + +import { expect } from 'chai'; +import { UnifiedAPIServer } from './UnifiedAPIServer.js'; +import { VirtualFileSystem } from './VirtualFileSystem.js'; +import { UnifiedProcessingStateManager } from '../logic/UnifiedProcessingStateManager.js'; +import { DuplicateResult } from '../logic/DuplicateFinder.js'; +import type { IKVClient } from '../kv/IKVClient.js'; +import type { FastifyInstance } from 'fastify'; + +// ============================================================================= +// Mock Implementations +// ============================================================================= + +/** + * Mock KV Client for testing + * Implements IKVClient interface with in-memory storage + */ +class MockKVClient implements IKVClient { + private data: Map> = new Map(); + private rawStore: Map = new Map(); + + // Basic operations + async set(key: string, value: any): Promise { + this.rawStore.set(key, JSON.stringify(value)); + } + + async get(key: string): Promise { + const val = this.rawStore.get(key); + return val ? JSON.parse(val) : null; + } + + async delete(key: string): Promise { + this.rawStore.delete(key); + } + + async countKeysWithPrefix(prefix: string): Promise { + let count = 0; + for (const key of this.rawStore.keys()) { + if (key.startsWith(prefix)) count++; + } + return count; + } + + async health(): Promise { + return true; + } + + // Nested key architecture methods + async setProperty(key: string, value: string): Promise { + this.rawStore.set(key, value); + } + + async getProperty(key: string): Promise { + return this.rawStore.get(key) ?? null; + } + + async getRange(prefix: string): Promise> { + const pairs: Array<{ key: string; value: string }> = []; + for (const [key, value] of this.rawStore.entries()) { + if (key.startsWith(prefix)) { + pairs.push({ key, value }); + } + } + return pairs; + } + + async setMultiple(pairs: Array<{ key: string; value: string }>): Promise { + for (const { key, value } of pairs) { + this.rawStore.set(key, value); + } + } + + async deleteRange(prefix: string): Promise { + let count = 0; + for (const key of this.rawStore.keys()) { + if (key.startsWith(prefix)) { + this.rawStore.delete(key); + count++; + } + } + return count; + } + + // High-level metadata operations + async setMetadataFlat(hashId: string, metadata: any, excludeFields?: string[]): Promise { + if (!this.data.has(hashId)) { + this.data.set(hashId, new Map()); + } + const entry = this.data.get(hashId)!; + for (const [key, value] of Object.entries(metadata)) { + if (!excludeFields?.includes(key)) { + entry.set(key, value); + } + } + } + + async getMetadataFlat(hashId: string): Promise { + const entry = this.data.get(hashId); + if (!entry) return null; + return Object.fromEntries(entry); + } + + async getMetadata(hashId: string, propertyPath: string): Promise { + const entry = this.data.get(hashId); + if (!entry) return null; + return entry.get(propertyPath) ?? null; + } + + async deleteMetadataFlat(hashId: string): Promise { + const entry = this.data.get(hashId); + const count = entry?.size ?? 0; + this.data.delete(hashId); + return count; + } + + async getAllHashIds(): Promise { + return Array.from(this.data.keys()); + } + + async close(): Promise { + // No-op for mock + } + + // Helper for tests to seed data + seedData(hashId: string, metadata: Record): void { + const entry = new Map(Object.entries(metadata)); + this.data.set(hashId, entry); + } + + clear(): void { + this.data.clear(); + this.rawStore.clear(); + } +} + +/** + * Mock Processing State Manager + */ +class MockProcessingStateManager { + private pending = new Set(); + private lightProcessing = new Set(); + private hashProcessing = new Set(); + private done = new Set(); + private failed = new Map(); + + getSnapshot() { + return { + pending: Array.from(this.pending), + lightProcessing: Array.from(this.lightProcessing), + hashProcessing: Array.from(this.hashProcessing), + done: Array.from(this.done) + }; + } + + getFailedFiles(): Array<{ path: string; error: string; timestamp: number }> { + return Array.from(this.failed.entries()).map(([path, error]) => ({ + path, + error, + timestamp: Date.now() + })); + } + + retryFile(filePath: string): boolean { + if (this.failed.has(filePath)) { + this.failed.delete(filePath); + this.pending.add(filePath); + return true; + } + return false; + } + + retryAllFailed(): number { + const count = this.failed.size; + for (const path of this.failed.keys()) { + this.pending.add(path); + } + this.failed.clear(); + return count; + } + + // Helpers for tests + addPending(path: string): void { + this.pending.add(path); + } + + addDone(path: string): void { + this.done.add(path); + } + + addFailed(path: string, error: string): void { + this.failed.set(path, error); + } +} + +/** + * Mock Duplicate Result Provider + */ +function createMockDuplicateResult(): DuplicateResult { + return { + hashDuplicates: [ + { + key: 'abc123', + files: ['/files/movie1.mp4', '/files/movie1-copy.mp4'] + } + ], + titleDuplicates: [ + { + key: 'Movie Title', + files: ['/files/movie1.mp4', '/files/movie2.mp4'] + } + ] + }; +} + +// ============================================================================= +// Test Suite +// ============================================================================= + +describe('UnifiedAPIServer', function() { + let server: UnifiedAPIServer; + let app: FastifyInstance; + let mockKV: MockKVClient; + let mockStateManager: MockProcessingStateManager; + let duplicateResult = createMockDuplicateResult(); + + before(async function() { + // Create mocks + mockKV = new MockKVClient(); + mockStateManager = new MockProcessingStateManager(); + + // Seed some test data + mockKV.seedData('testhash001', { + fileName: 'Test Movie.mp4', + filePath: '/files/test/Test Movie.mp4', + sizeByte: 1000000, + videoType: 'movie', + title: 'Test Movie' + }); + mockKV.seedData('testhash002', { + fileName: 'Test Episode S01E01.mp4', + filePath: '/files/test/Test Episode S01E01.mp4', + sizeByte: 500000, + videoType: 'episode', + title: 'Test Episode' + }); + + mockStateManager.addDone('/files/test/Test Movie.mp4'); + mockStateManager.addPending('/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, + () => duplicateResult, + mockKV as any, + async () => { /* scan trigger */ }, + 4, // backgroundQueueConcurrency + 16, // fastQueueConcurrency + () => ({ fast: { pending: 0, running: 0 }, background: { pending: 0, running: 0 } }) + ); + + app = server.getApp(); + await app.ready(); + }); + + after(async function() { + await app.close(); + }); + + // ========================================================================= + // Health Endpoints + // ========================================================================= + + describe('Health Endpoints', function() { + it('GET /health returns ok status', async function() { + const response = await app.inject({ + method: 'GET', + url: '/health' + }); + + 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'); + }); + + it('GET /api/health returns redis status', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/health' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('status'); + expect(body).to.have.property('redis'); + expect(body).to.have.property('uptime'); + }); + + it('GET /meta-health returns status', async function() { + const response = await app.inject({ + method: 'GET', + url: '/meta-health' + }); + + expect(response.statusCode).to.equal(200); + }); + }); + + // ========================================================================= + // Processing API (Monitor Page) + // ========================================================================= + + describe('Processing API (/api/processing/*)', function() { + it('GET /api/processing/status returns processing snapshot', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/processing/status' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('pending'); + expect(body).to.have.property('lightProcessing'); + expect(body).to.have.property('hashProcessing'); + expect(body).to.have.property('done'); + expect(Array.isArray(body.pending)).to.be.true; + expect(Array.isArray(body.done)).to.be.true; + }); + + it('GET /api/processing/queue returns queue items', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/processing/queue' + }); + + // May return 200 with items or 500 if TaskScheduler not available + expect([200, 500]).to.include(response.statusCode); + if (response.statusCode === 200) { + const body = JSON.parse(response.payload); + expect(body).to.have.property('items'); + expect(Array.isArray(body.items)).to.be.true; + } + }); + + it('GET /api/processing/failed returns failed files list', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/processing/failed' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('failedFiles'); + expect(body).to.have.property('totalFailed'); + expect(Array.isArray(body.failedFiles)).to.be.true; + }); + + it('POST /api/processing/retry returns 400 without filePath', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/processing/retry', + payload: {} + }); + + expect(response.statusCode).to.equal(400); + }); + + it('POST /api/processing/retry-all returns status', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/processing/retry-all' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('status', 'ok'); + expect(body).to.have.property('message'); + }); + }); + + // ========================================================================= + // Duplicates API (Duplicates Page) + // ========================================================================= + + describe('Duplicates API (/api/duplicates)', function() { + it('GET /api/duplicates returns duplicate analysis', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/duplicates' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('hashDuplicates'); + expect(body).to.have.property('titleDuplicates'); + expect(Array.isArray(body.hashDuplicates)).to.be.true; + expect(Array.isArray(body.titleDuplicates)).to.be.true; + }); + + it('POST /api/duplicates/refresh acknowledges request', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/duplicates/refresh' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('status'); + }); + }); + + // ========================================================================= + // Metadata API (File Browser) + // ========================================================================= + + describe('Metadata API (/api/metadata/*)', function() { + it('GET /api/metadata/hash-ids returns list of hash IDs', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/metadata/hash-ids' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('hashIds'); + expect(Array.isArray(body.hashIds)).to.be.true; + expect(body.hashIds).to.include('testhash001'); + }); + + it('GET /api/metadata/list returns paginated file list', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/metadata/list?limit=10&offset=0' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('files'); + expect(body).to.have.property('total'); + expect(Array.isArray(body.files)).to.be.true; + }); + + it('GET /api/metadata/:hashId returns file metadata', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/metadata/testhash001' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + // API returns metadata flat (not wrapped in { hashId, metadata }) + expect(body).to.have.property('fileName', 'Test Movie.mp4'); + }); + + it('GET /api/metadata/:hashId returns 404 for unknown hash', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/metadata/unknownhash' + }); + + expect(response.statusCode).to.equal(404); + }); + + it('GET /api/metadata/:hashId/property returns specific property', async function() { + // The API uses getProperty with key format /file/{hashId}/{property} + // We need to seed the property in the rawStore + await mockKV.setProperty('/file/testhash001/videoType', 'movie'); + + const response = await app.inject({ + method: 'GET', + url: '/api/metadata/testhash001/property?property=videoType' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('value', 'movie'); + }); + + it('PUT /api/metadata/:hashId updates metadata', async function() { + const response = await app.inject({ + method: 'PUT', + url: '/api/metadata/testhash001', + payload: { + customField: 'customValue' + } + }); + + expect(response.statusCode).to.equal(200); + + // Verify the update + const verify = await app.inject({ + method: 'GET', + url: '/api/metadata/testhash001' + }); + const body = JSON.parse(verify.payload); + // Response is flat metadata + expect(body).to.have.property('customField', 'customValue'); + }); + + it('DELETE /api/metadata/:hashId deletes file metadata', async function() { + // First create a test entry + mockKV.seedData('deleteme', { fileName: 'delete.mp4' }); + + const response = await app.inject({ + method: 'DELETE', + url: '/api/metadata/deleteme' + }); + + expect(response.statusCode).to.equal(200); + + // Verify deletion + const verify = await app.inject({ + method: 'GET', + url: '/api/metadata/deleteme' + }); + expect(verify.statusCode).to.equal(404); + }); + + it('POST /api/metadata/search searches files', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/metadata/search', + payload: { + property: 'videoType', + propertyValue: 'movie' + } + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('results'); + expect(Array.isArray(body.results)).to.be.true; + }); + }); + + // ========================================================================= + // Stats API (Dashboard) + // ========================================================================= + + describe('Stats API (/api/stats)', function() { + it('GET /api/stats returns statistics', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/stats' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('fileCount'); + expect(body).to.have.property('totalSize'); + }); + }); + + // ========================================================================= + // Metrics API + // ========================================================================= + + describe('Metrics API (/api/metrics)', function() { + it('GET /api/metrics returns performance metrics', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/metrics' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('uptime'); + }); + }); + + // ========================================================================= + // Scan API + // ========================================================================= + + describe('Scan API (/api/scan)', function() { + it('POST /api/scan/trigger triggers a scan', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/scan/trigger' + }); + + expect([200, 202, 204]).to.include(response.statusCode); + }); + }); + + // ========================================================================= + // 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; + }); + }); + + // ========================================================================= + // KV Browser API (Editor) + // ========================================================================= + + describe('KV Browser API (/api/kv/*)', function() { + it('GET /api/kv/info returns database info', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/kv/info' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('fileCount'); + }); + + it('GET /api/kv/keys returns paginated keys', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/kv/keys?cursor=0&count=50' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('keys'); + expect(Array.isArray(body.keys)).to.be.true; + }); + }); + + // ========================================================================= + // Plugins API (Plugins Page) + // ========================================================================= + + describe('Plugins API (/api/plugins/*)', function() { + it('GET /api/plugins returns 503 when plugin manager not initialized', async function() { + // Without a plugin manager configured, should return 503 + const response = await app.inject({ + method: 'GET', + url: '/api/plugins' + }); + + expect(response.statusCode).to.equal(503); + const body = JSON.parse(response.payload); + expect(body).to.have.property('error'); + expect(body.error).to.include('not yet initialized'); + }); + + it('GET /api/plugins/timings returns timing data', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/plugins/timings' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('timings'); + expect(Array.isArray(body.timings)).to.be.true; + }); + + it('POST /api/plugins/:pluginId/activate returns 503 without plugin manager', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/plugins/test-plugin/activate' + }); + + expect(response.statusCode).to.equal(503); + }); + + it('POST /api/plugins/:pluginId/deactivate returns 503 without plugin manager', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/plugins/test-plugin/deactivate' + }); + + expect(response.statusCode).to.equal(503); + }); + + it('PUT /api/plugins/:pluginId/config returns 503 without plugin manager', async function() { + const response = await app.inject({ + method: 'PUT', + url: '/api/plugins/test-plugin/config', + payload: { setting: 'value' } + }); + + expect(response.statusCode).to.equal(503); + }); + + it('POST /api/plugins/:pluginId/clear-cache returns 503 without plugin manager', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/plugins/test-plugin/clear-cache' + }); + + expect(response.statusCode).to.equal(503); + }); + + it('POST /api/plugins/rescan returns 503 without plugin manager', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/plugins/rescan' + }); + + expect(response.statusCode).to.equal(503); + }); + }); + + // ========================================================================= + // Container Plugins API + // ========================================================================= + + describe('Container Plugins API (/api/plugins/containers/*)', function() { + it('GET /api/plugins/containers returns container plugin info', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/plugins/containers' + }); + + // May return 200 with empty data or 503 if not initialized + expect([200, 503]).to.include(response.statusCode); + + if (response.statusCode === 200) { + const body = JSON.parse(response.payload); + expect(body).to.have.property('pluginCount'); + } + }); + + it('POST /api/plugins/callback returns 503 without scheduler', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/plugins/callback', + payload: { + taskId: 'test-task-123', + pluginId: 'test-plugin', + status: 'success', + hashId: 'testhash001', + metadata: { testKey: 'testValue' } + } + }); + + // Returns 503 when container plugin scheduler not available + expect(response.statusCode).to.equal(503); + }); + }); + + // ========================================================================= + // Mounts API (Mounts Page) + // ========================================================================= + + describe('Mounts API (/api/mounts/*)', function() { + it('GET /api/mounts returns mount list', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/mounts' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('mounts'); + expect(Array.isArray(body.mounts)).to.be.true; + }); + + it('GET /api/mounts/rclone/remotes returns rclone remotes', async function() { + const response = await app.inject({ + method: 'GET', + url: '/api/mounts/rclone/remotes' + }); + + // May succeed or fail depending on rclone availability + expect([200, 500]).to.include(response.statusCode); + + if (response.statusCode === 200) { + const body = JSON.parse(response.payload); + expect(body).to.have.property('remotes'); + expect(Array.isArray(body.remotes)).to.be.true; + } + }); + + it('POST /api/mounts validates required fields', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/mounts', + payload: { + // Missing required fields + } + }); + + expect(response.statusCode).to.equal(400); + }); + + it('POST /api/mounts/:id/mount returns 404 for unknown mount', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/mounts/nonexistent-mount/mount' + }); + + expect(response.statusCode).to.equal(404); + }); + + it('POST /api/mounts/:id/unmount returns 404 for unknown mount', async function() { + const response = await app.inject({ + method: 'POST', + url: '/api/mounts/nonexistent-mount/unmount' + }); + + expect(response.statusCode).to.equal(404); + }); + + it('DELETE /api/mounts/:id returns 404 for unknown mount', async function() { + const response = await app.inject({ + method: 'DELETE', + url: '/api/mounts/nonexistent-mount' + }); + + expect(response.statusCode).to.equal(404); + }); + }); + + // ========================================================================= + // Meta-Core Compatible API (for plugins) + // ========================================================================= + + describe('Meta-Core Compatible API (/meta/*)', function() { + it('GET /meta/:hash returns file metadata wrapped', async function() { + const response = await app.inject({ + method: 'GET', + url: '/meta/testhash001' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + // API returns { metadata: {...} } + expect(body).to.have.property('metadata'); + expect(body.metadata).to.have.property('fileName', 'Test Movie.mp4'); + }); + + it('GET /meta/:hash returns empty metadata for unknown hash', async function() { + const response = await app.inject({ + method: 'GET', + url: '/meta/unknownhash' + }); + + // API returns { metadata: {} } for unknown hash, not 404 + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('metadata'); + }); + + it('GET /meta/:hash/:key returns specific property', async function() { + const response = await app.inject({ + method: 'GET', + url: '/meta/testhash001/videoType' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('value', 'movie'); + }); + + it('PUT /meta/:hash/:key sets property value', async function() { + const response = await app.inject({ + method: 'PUT', + url: '/meta/testhash001/metaCustomField', + payload: { value: 'metaCustomValue' } + }); + + expect(response.statusCode).to.equal(200); + + // Verify the update + const verify = await app.inject({ + method: 'GET', + url: '/meta/testhash001/metaCustomField' + }); + const body = JSON.parse(verify.payload); + expect(body).to.have.property('value', 'metaCustomValue'); + }); + + it('PATCH /meta/:hash merges metadata', async function() { + const response = await app.inject({ + method: 'PATCH', + url: '/meta/testhash001', + payload: { + patchedField: 'patchedValue', + anotherField: '123' + } + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('status', 'ok'); + + // Verify the merge + const verify = await app.inject({ + method: 'GET', + url: '/meta/testhash001' + }); + const verifyBody = JSON.parse(verify.payload); + expect(verifyBody.metadata).to.have.property('patchedField', 'patchedValue'); + expect(verifyBody.metadata).to.have.property('fileName', 'Test Movie.mp4'); // Original preserved + }); + + it('DELETE /meta/:hash/:key removes property', async function() { + // First set a property + await app.inject({ + method: 'PUT', + url: '/meta/testhash001/toDeleteMeta', + payload: { value: 'deleteMe' } + }); + + // Verify property was set + const beforeDelete = await app.inject({ + method: 'GET', + url: '/meta/testhash001/toDeleteMeta' + }); + expect(beforeDelete.statusCode).to.equal(200); + + // Then delete it + const response = await app.inject({ + method: 'DELETE', + url: '/meta/testhash001/toDeleteMeta' + }); + + expect(response.statusCode).to.equal(200); + const body = JSON.parse(response.payload); + expect(body).to.have.property('status', 'ok'); + }); + }); + + // ========================================================================= + // File CID API + // ========================================================================= + + describe('File CID API (/file/cid)', function() { + it('POST /file/cid returns 400 without path', async function() { + const response = await app.inject({ + method: 'POST', + url: '/file/cid', + payload: {} + }); + + expect(response.statusCode).to.equal(400); + }); + + it('POST /file/cid returns 404 for nonexistent file', async function() { + const response = await app.inject({ + method: 'POST', + url: '/file/cid', + payload: { path: '/nonexistent/file.mp4' } + }); + + expect(response.statusCode).to.equal(404); + }); + }); +}); diff --git a/packages/meta-sort-core/src/config/EnvConfig.ts b/packages/meta-sort-core/src/config/EnvConfig.ts index 4b6c5cc..b00dbc0 100644 --- a/packages/meta-sort-core/src/config/EnvConfig.ts +++ b/packages/meta-sort-core/src/config/EnvConfig.ts @@ -80,18 +80,12 @@ interface EnvConfig { // TaskScheduler Configuration (Plugin Queue System) // ======================================================================== - /** Concurrency for fast plugin queue (plugins with <1s avg execution time). Default: 32 */ + /** Concurrency for fast plugin queue. Default: 32 */ FAST_QUEUE_CONCURRENCY: number; - /** Concurrency for background plugin queue (plugins with >=1s avg execution time). Default: 8 */ + /** Concurrency for background plugin queue. Default: 8 */ BACKGROUND_QUEUE_CONCURRENCY: number; - /** Threshold in ms to classify plugin as fast vs background. Default: 1000 */ - FAST_THRESHOLD_MS: number; - - /** Minimum samples before using measured timing for classification. Default: 10 */ - MIN_SAMPLES_FOR_MEASUREMENT: number; - // ======================================================================== // Container Plugins Configuration // All plugins run as Docker containers for isolation and flexibility @@ -183,8 +177,6 @@ export const config: EnvConfig = { // TaskScheduler Configuration FAST_QUEUE_CONCURRENCY: parseInt(process.env.FAST_QUEUE_CONCURRENCY || "32", 10), BACKGROUND_QUEUE_CONCURRENCY: parseInt(process.env.BACKGROUND_QUEUE_CONCURRENCY || "8", 10), - FAST_THRESHOLD_MS: parseInt(process.env.FAST_THRESHOLD_MS || "1000", 10), - MIN_SAMPLES_FOR_MEASUREMENT: parseInt(process.env.MIN_SAMPLES_FOR_MEASUREMENT || "10", 10), // Container Plugins Configuration CONTAINER_PLUGINS_CONFIG: process.env.CONTAINER_PLUGINS_CONFIG || '/app/plugins.yml', diff --git a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts index 8070840..8391522 100644 --- a/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts +++ b/packages/meta-sort-core/src/container-plugins/ContainerPluginScheduler.ts @@ -90,8 +90,15 @@ export class ContainerPluginScheduler extends EventEmitter { concurrency: options?.fastConcurrency ?? config.FAST_QUEUE_CONCURRENCY, }); + // Background queue starts paused - only runs when fast queue is idle this.backgroundQueue = new PQueue({ concurrency: options?.backgroundConcurrency ?? config.BACKGROUND_QUEUE_CONCURRENCY, + autoStart: false, + }); + + // When fast queue becomes idle, start background queue + this.fastQueue.on('idle', () => { + this.onFastQueueIdle(); }); this.callbackUrl = options?.callbackUrl ?? `${config.CONTAINER_CALLBACK_URL}/api/plugins/callback`; @@ -262,12 +269,28 @@ export class ContainerPluginScheduler extends EventEmitter { const queue = task.queue === 'background' ? this.backgroundQueue : this.fastQueue; + // If adding to fast queue, pause background queue (fast takes priority) + if (task.queue === 'fast' && !this.backgroundQueue.isPaused) { + console.log('[ContainerScheduler] Fast task added, pausing background queue'); + this.backgroundQueue.pause(); + } + queue.add(async () => { await this.executeTask(task); }); return true; } + /** + * Called when fast queue becomes idle - start background queue if it has work + */ + private onFastQueueIdle(): void { + if (this.backgroundQueue.size > 0 || this.backgroundQueue.pending > 0) { + console.log('[ContainerScheduler] Fast queue idle, starting background queue'); + this.backgroundQueue.start(); + } + } + /** * Execute a task (dispatch to container) */ @@ -709,10 +732,15 @@ export class ContainerPluginScheduler extends EventEmitter { /** * Resume queues + * Note: Background queue only starts if fast queue is idle (fast queue has priority) */ resume(): void { this.fastQueue.start(); - this.backgroundQueue.start(); + // Only start background if fast queue is empty + if (this.fastQueue.size === 0 && this.fastQueue.pending === 0) { + this.backgroundQueue.start(); + } + // Otherwise, background will start when fast queue idle event fires } /** diff --git a/packages/meta-sort-core/src/index.ts b/packages/meta-sort-core/src/index.ts index 0c324b6..4e02ef7 100644 --- a/packages/meta-sort-core/src/index.ts +++ b/packages/meta-sort-core/src/index.ts @@ -121,8 +121,15 @@ const runProgressiveDiscovery = async () => { // Note: Stremio updates happen automatically via incremental add/remove notifications // during file processing (like FUSE), so no explicit refresh needed here const triggerScan = async () => { - console.log('Triggering manual scan from API...'); + console.log('[Scan] Triggering manual scan from API...'); + + // Reset pipeline counters and state before fresh scan + // Also publishes reset event to meta-fuse + await pipeline.reset(); + + console.log('[Scan] Starting fresh discovery...'); await runProgressiveDiscovery(); + console.log('[Scan] Manual scan complete'); }; // Global error handlers - prevent crashes from unforeseen errors @@ -183,10 +190,11 @@ process.on('SIGINT', async () => { await kvManager.start(); await kvManager.waitForReady(); - // Pass KV client to file processor + // Pass KV client to file processor and pipeline const client = kvManager.getClient(); if (client) { fileProcessor.setKVClient(client); + pipeline.setKVClient(client); console.log(`[Startup] KV Manager ready (role: ${kvManager.isLeader() ? 'LEADER' : 'FOLLOWER'})`); } } diff --git a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts index 006ef77..96942a3 100644 --- a/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts +++ b/packages/meta-sort-core/src/logic/WatchedFileProcessor.ts @@ -199,8 +199,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { this.taskScheduler = createTaskScheduler(pluginManager, performanceMetrics, { fastQueueConcurrency: config.FAST_QUEUE_CONCURRENCY, backgroundQueueConcurrency: config.BACKGROUND_QUEUE_CONCURRENCY, - fastThresholdMs: config.FAST_THRESHOLD_MS, - minSamplesForMeasurement: config.MIN_SAMPLES_FOR_MEASUREMENT, }); // Set up event listeners for VFS updates @@ -209,8 +207,6 @@ export class WatchedFileProcessor implements FileProcessorInterface { console.log(`[WatchedFileProcessor] TaskScheduler initialized with:`); console.log(` - Fast queue concurrency: ${config.FAST_QUEUE_CONCURRENCY}`); console.log(` - Background queue concurrency: ${config.BACKGROUND_QUEUE_CONCURRENCY}`); - console.log(` - Fast threshold: ${config.FAST_THRESHOLD_MS}ms`); - console.log(` - Min samples for measurement: ${config.MIN_SAMPLES_FOR_MEASUREMENT}`); } /** diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 80f3995..378d28f 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -1,12 +1,14 @@ import * as path from 'path'; import PQueue from 'p-queue'; import {PipelineConfig} from './PipelineConfig.js'; -import {withTimeoutAndRetry, TimeoutConfig} from '../../utils/TimeoutWrapper.js'; import {performanceMetrics} from '../../metrics/PerformanceMetrics.js'; // Pub/sub channel for batched updates to meta-fuse const UPDATE_CHANNEL = 'meta-sort:file:batch'; +// Pub/sub channel for scan reset notification +const RESET_CHANNEL = 'meta-sort:scan:reset'; + // Batch update interval (5 seconds) const BATCH_INTERVAL_MS = 5000; @@ -39,23 +41,6 @@ export class StreamingPipeline { private batchBuffer: BatchChange[] = []; private batchTimer: NodeJS.Timeout | null = null; - // Retry configuration - private readonly MAX_PROCESSING_RETRIES = 10; - - // Fast queue timeout: Fast metadata extraction + midhash256 - private readonly fastQueueTimeoutConfig: TimeoutConfig = { - baseTimeout: 120000, // 2 minutes base timeout - timeoutMultiplier: 1.5, // Increase by 1.5x each retry - maxTimeout: 600000 // Max 10 minutes - }; - - // Background queue timeout: Full file hashing over NAS (can be very slow) - private readonly backgroundQueueTimeoutConfig: TimeoutConfig = { - baseTimeout: 7200000, // 2 hours base timeout (for NAS) - timeoutMultiplier: 1.5, // Increase by 1.5x each retry - maxTimeout: 14400000 // Max 4 hours - }; - constructor(config: PipelineConfig) { this.config = config; @@ -223,23 +208,14 @@ export class StreamingPipeline { * - Compute midhash256 (< 1s) * - Store to KV (file becomes accessible in VFS) * - Queue for background processing - * - Includes timeout and retry logic */ private async processLightPhase(filePath: string): Promise { try { - // Get current retry count - const retryCount = this.config.stateManager.getRetryCount(filePath); - - // Call light processing with timeout + // Call light processing const current = this.fastProcessedCount + 1; const queueSize = this.discoveredCount; - await withTimeoutAndRetry( - () => this.config.fileProcessor.processLightPhase(filePath, current, queueSize), - retryCount, - this.fastQueueTimeoutConfig, - `Fast queue processing file ${filePath}` - ); + await this.config.fileProcessor.processLightPhase(filePath, current, queueSize); // After light processing, add to VFS const metadata = this.config.fileProcessor.getDatabase().get(filePath); @@ -250,7 +226,6 @@ export class StreamingPipeline { if (virtualPath) { // Add to VFS immediately (file appears with permanent midhash256 ID) this.config.virtualFileSystem.addFile(virtualPath, filePath, metadata); - //console.log(`✓ File added to VFS: ${filePath} → ${virtualPath} (midhash256: ${metadata.cid_midhash256})`); // Queue for pub/sub notification to meta-fuse if (metadata.cid_midhash256) { @@ -278,30 +253,14 @@ export class StreamingPipeline { this.fastProcessedCount++; } catch (error: any) { - // Handle timeout or processing error - const retryCount = this.config.stateManager.getRetryCount(filePath); - - if (retryCount < this.MAX_PROCESSING_RETRIES) { - // Retry: move back to pending queue with incremented retry count - const success = this.config.stateManager.retryProcessing(filePath); - if (success) { - console.warn(`[Pipeline] Fast queue processing failed for ${filePath} (attempt ${retryCount + 1}/${this.MAX_PROCESSING_RETRIES}), retrying with longer timeout. Error: ${error.message}`); - - // Re-queue for fast processing - this.fastQueue.add(() => this.processLightPhase(filePath)) - .catch(err => console.error(`[Pipeline] Retry failed for ${filePath}:`, err.message)); - return; - } - } - - // Max retries exceeded or retry failed - mark as permanently failed - console.error(`[Pipeline] Fast queue processing permanently failed for ${filePath} after ${retryCount + 1} attempts: ${error.message}`); + // Processing failed - mark as failed (manual retry available via UI) + console.error(`[Pipeline] Fast queue processing failed for ${filePath}: ${error.message}`); // Record failed file in metrics performanceMetrics.recordFailedFile( filePath, error.message || 'Unknown error', - retryCount + 1, + 1, 'processing' ); @@ -315,47 +274,25 @@ export class StreamingPipeline { * - Compute full hashes (SHA-256, SHA-1, MD5, CRC32) * - Update KV with additional hash metadata * - Mark processing as complete - * - Includes timeout and retry logic */ private async processHashPhase(filePath: string): Promise { try { - // Get current retry count - const retryCount = this.config.stateManager.getRetryCount(filePath); - - // Call hash processing with timeout + // Call hash processing const current = this.backgroundProcessedCount + 1; const queueSize = this.discoveredCount; - await withTimeoutAndRetry( - () => this.config.fileProcessor.processHashPhase(filePath, current, queueSize), - retryCount, - this.backgroundQueueTimeoutConfig, - `Background queue processing file ${filePath}` - ); + await this.config.fileProcessor.processHashPhase(filePath, current, queueSize); this.backgroundProcessedCount++; } catch (error: any) { - // Handle timeout or processing error - const retryCount = this.config.stateManager.getRetryCount(filePath); - - if (retryCount < this.MAX_PROCESSING_RETRIES) { - // Retry: move back to background queue with incremented retry count - console.warn(`[Pipeline] Background queue processing failed for ${filePath} (attempt ${retryCount + 1}/${this.MAX_PROCESSING_RETRIES}), retrying. Error: ${error.message}`); - - // Re-queue for background processing - this.backgroundQueue.add(() => this.processHashPhase(filePath)) - .catch(err => console.error(`[Pipeline] Retry failed for ${filePath}:`, err.message)); - return; - } - - // Max retries exceeded - mark as permanently failed - console.error(`[Pipeline] Background queue processing permanently failed for ${filePath} after ${retryCount + 1} attempts: ${error.message}`); + // Processing failed - mark as failed (manual retry available via UI) + console.error(`[Pipeline] Background queue processing failed for ${filePath}: ${error.message}`); // Record failed file in metrics performanceMetrics.recordFailedFile( filePath, error.message || 'Unknown error', - retryCount + 1, + 1, 'hash' ); @@ -432,6 +369,43 @@ export class StreamingPipeline { }; } + /** + * Reset pipeline counters and clear state manager + * Called when triggering a fresh scan to reset statistics + * Publishes reset event to notify meta-fuse to clear its VFS + */ + async reset(): Promise { + console.log('[Pipeline] Resetting counters and state...'); + + // Reset counters + this.discoveredCount = 0; + this.validatedCount = 0; + this.fastProcessedCount = 0; + this.backgroundProcessedCount = 0; + + // Clear batch buffer + this.batchBuffer = []; + + // Clear state manager (discovered, processing, done states) + this.config.stateManager.clear(); + + // Publish reset event to notify meta-fuse + if (this.config.kvClient && typeof (this.config.kvClient as any).publish === 'function') { + try { + const message = JSON.stringify({ + timestamp: Date.now(), + action: 'reset' + }); + await (this.config.kvClient as any).publish(RESET_CHANNEL, message); + console.log('[Pipeline] Reset event published to meta-fuse'); + } catch (error) { + console.error('[Pipeline] Failed to publish reset event:', error); + } + } + + console.log('[Pipeline] Reset complete'); + } + /** * Pause all pipeline queues * Running tasks will complete, but no new tasks will start @@ -520,6 +494,15 @@ export class StreamingPipeline { console.log('[Pipeline] Container plugin scheduler connected'); } + /** + * Set the KV client (called after initialization when Redis is ready) + * This enables pub/sub notifications to meta-fuse + */ + setKVClient(kvClient: import('../../kv/IKVClient.js').IKVClient): void { + (this.config as any).kvClient = kvClient; + console.log('[Pipeline] KV client connected - pub/sub enabled'); + } + /** * Dispatch tasks to container plugins for metadata extraction * This is fire-and-forget - container plugins write results directly to meta-core diff --git a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts b/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts index 9824246..55716fd 100644 --- a/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts +++ b/packages/meta-sort-core/src/plugin-engine/TaskScheduler.ts @@ -18,7 +18,7 @@ import type { PluginKVStore, } from './types.js'; import type { PluginManager } from './PluginManager.js'; -import type { PerformanceMetrics, PluginTimingStats } from '../metrics/PerformanceMetrics.js'; +import type { PerformanceMetrics } from '../metrics/PerformanceMetrics.js'; import type { ContainerPluginScheduler } from '../container-plugins/index.js'; import { KVStore } from './PluginContext.js'; @@ -31,17 +31,11 @@ export interface TaskSchedulerConfig { fastQueueConcurrency?: number; /** Concurrency for background queue (default: 8) */ backgroundQueueConcurrency?: number; - /** Threshold in ms for fast classification (default: 1000) */ - fastThresholdMs?: number; - /** Minimum samples before using measured timing (default: 10) */ - minSamplesForMeasurement?: number; } const DEFAULT_CONFIG: Required = { fastQueueConcurrency: 32, backgroundQueueConcurrency: 8, - fastThresholdMs: 1000, - minSamplesForMeasurement: 10, }; // ============================================================================= @@ -379,25 +373,11 @@ export class TaskScheduler extends EventEmitter { // ========================================================================= /** - * Classify a plugin as fast or background based on timing data and manifest + * Classify a plugin as fast or background based on manifest defaultQueue */ classifyPlugin(pluginId: string): TaskQueueType { - const plugins = this.pluginManager.getPlugins(); - const plugin = plugins.find(p => p.id === pluginId); - const timing = this.performanceMetrics.getPluginTiming(pluginId); - - // Get manifest default queue (need to access via plugin manager internals) - // For now, we'll use a workaround - check if plugin has defaultQueue in manifest const manifest = this.getPluginManifest(pluginId); - const defaultQueue = manifest?.defaultQueue || 'fast'; - - // If we have enough samples, use measured timing - if (timing && timing.count >= this.config.minSamplesForMeasurement) { - return timing.average < this.config.fastThresholdMs ? 'fast' : 'background'; - } - - // Otherwise use manifest default - return defaultQueue; + return manifest?.defaultQueue || 'fast'; } // ========================================================================= diff --git a/packages/meta-sort-core/tsconfig.test.json b/packages/meta-sort-core/tsconfig.test.json new file mode 100644 index 0000000..219efa0 --- /dev/null +++ b/packages/meta-sort-core/tsconfig.test.json @@ -0,0 +1,8 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "outDir": "./dist" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "src/plugin-engine/plugin.test.ts"] +}