diff --git a/README.md b/README.md index 4b4080a..e53a98f 100644 --- a/README.md +++ b/README.md @@ -508,7 +508,6 @@ services: | `META_CORE_PATH` | `/meta-core` | Path to META_CORE_VOLUME (infrastructure) | | `FILES_PATH` | `/files` | Path to FILES_VOLUME (shared media) | | `REDIS_URL` | Auto-discover | Redis connection (from leader lock file) | -| `WATCH_FOLDER_LIST` | `$FILES_PATH` | Comma-separated folders to watch for media | | `SERVICE_NAME` | Container hostname | Unique service identifier | | `BASE_URL` | `http://localhost:8180` | External URL for this service | diff --git a/packages/meta-sort-core/.env.example b/packages/meta-sort-core/.env.example index 557a417..9c1ef9f 100644 --- a/packages/meta-sort-core/.env.example +++ b/packages/meta-sort-core/.env.example @@ -2,14 +2,7 @@ # Copy this file to .env and adjust values as needed # ============================================================================= -# Required Settings -# ============================================================================= - -# Comma-separated list of folders to watch for media files (REQUIRED) -WATCH_FOLDER_LIST=/files - -# ============================================================================= -# Volume Paths (New Architecture) +# Volume Paths # ============================================================================= # Path to META_CORE_VOLUME - infrastructure volume for KV DB, locks, services diff --git a/packages/meta-sort-core/src/events/FileEventConsumer.ts b/packages/meta-sort-core/src/events/FileEventConsumer.ts index 3fd6ea4..7d10358 100644 --- a/packages/meta-sort-core/src/events/FileEventConsumer.ts +++ b/packages/meta-sort-core/src/events/FileEventConsumer.ts @@ -12,8 +12,8 @@ import type { RedisKVClient, StreamMessage } from '../kv/RedisClient.js'; // Pipeline interface (minimal, matching StreamingPipeline) interface Pipeline { - handleFileAdded(filePath: string): void; - handleFileChanged(filePath: string): void; + handleFileAdded(filePath: string, midhash256?: string): void; + handleFileChanged(filePath: string, midhash256?: string): void; handleFileDeleted(filePath: string): void; } @@ -116,11 +116,11 @@ export class FileEventConsumer { switch (message.type) { case 'add': - this.pipeline.handleFileAdded(absolutePath); + this.pipeline.handleFileAdded(absolutePath, message.midhash256); break; case 'change': - this.pipeline.handleFileChanged(absolutePath); + this.pipeline.handleFileChanged(absolutePath, message.midhash256); break; case 'delete': @@ -132,7 +132,7 @@ export class FileEventConsumer { if (message.oldPath) { this.pipeline.handleFileDeleted(toAbsolutePath(message.oldPath)); } - this.pipeline.handleFileAdded(absolutePath); + this.pipeline.handleFileAdded(absolutePath, message.midhash256); break; default: diff --git a/packages/meta-sort-core/src/kv/RedisClient.ts b/packages/meta-sort-core/src/kv/RedisClient.ts index 18cbdff..61c69cc 100644 --- a/packages/meta-sort-core/src/kv/RedisClient.ts +++ b/packages/meta-sort-core/src/kv/RedisClient.ts @@ -35,7 +35,7 @@ export interface StreamMessage { type: 'add' | 'change' | 'delete' | 'rename'; path: string; size?: string; - partialHash?: string; + midhash256?: string; // midhash256 CID computed by meta-core oldPath?: string; timestamp: string; } @@ -606,7 +606,7 @@ export class RedisKVClient implements IKVClient { type: fieldMap.type as StreamMessage['type'], path: fieldMap.path, size: fieldMap.size, - partialHash: fieldMap.partialHash, + midhash256: fieldMap.midhash256, oldPath: fieldMap.oldPath, timestamp: fieldMap.timestamp || '0', }; diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileAnalyzerInterface.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileAnalyzerInterface.ts index a285076..95f589e 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileAnalyzerInterface.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileAnalyzerInterface.ts @@ -11,8 +11,12 @@ export interface FileAnalyzerInterface { /** * Light processing phase - fast metadata extraction + midhash256 * File becomes accessible in VFS with permanent ID after this phase + * @param filePath - Path to the file + * @param current - Current file number + * @param queueSize - Total queue size + * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ - processLightPhase(filePath: string, current: number, queueSize: number):Promise; + processLightPhase(filePath: string, current: number, queueSize: number, midhash256?: string):Promise; /** * Hash processing phase - background hash computation diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessor.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessor.ts index ce2def2..24569b6 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessor.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessor.ts @@ -1,6 +1,6 @@ import {HashIndexManager} from "@metazla/meta-hash"; import {config} from "../../config/EnvConfig.js"; -import {stat} from "fs/promises"; +import * as webdav from '../../webdav/WebdavClient.js'; import {targetHash} from "../../config/TargetHash.js"; import {HashMeta} from "@metazla/meta-interface"; import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js"; @@ -50,7 +50,10 @@ export class FileProcessor implements FileAnalyzerInterface{ return; } - const stats = await stat(filePath); + const stats = await webdav.stat(filePath); + if (!stats || !stats.exists) { + throw new Error(`File not found via WebDAV: ${filePath}`); + } let indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); let hash: HashMeta = {}; if(indexLine) { diff --git a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts index c0331bc..4012d55 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/FileProcessorPiscina.ts @@ -2,7 +2,7 @@ import {Piscina} from "piscina"; import {Input, Output} from "./worker.js"; import {HashIndexManager} from "@metazla/meta-hash"; import {config} from "../../config/EnvConfig.js"; -import {stat} from "fs/promises"; +import * as webdav from '../../webdav/WebdavClient.js'; // Note: Full hashes are now computed by the full-hash container plugin import {HashMeta} from "@metazla/meta-interface"; import {FileAnalyzerInterface} from "./FileAnalyzerInterface.js"; @@ -160,7 +160,10 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ this.unifiedStateManager.startLightProcessing(filePath); } - const stats = await stat(filePath); + const stats = await webdav.stat(filePath); + if (!stats || !stats.exists) { + throw new Error(`File not found via WebDAV: ${filePath}`); + } // STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files) let midHash256: string | undefined; @@ -325,44 +328,65 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ /** * Light Processing Phase - Fast processing to make file accessible in VFS - * - Compute midhash256 (< 1s) + * - Use midhash256 from meta-core (or compute locally if not provided) * - Extract metadata (filename, FFmpeg analysis) * - Store in KV * - File becomes accessible in VFS with midhash256 ID * + * @param filePath - Path to the file + * @param current - Current file number (for progress reporting) + * @param queueSize - Total queue size (for progress reporting) + * @param providedMidHash256 - Pre-computed midhash256 from meta-core (if available) + * * @deprecated Use TaskScheduler-based processing via WatchedFileProcessor.queueFile() instead. * This method is kept for backward compatibility with StreamingPipeline. */ - async processLightPhase(filePath: string, current: number, queueSize: number) { + async processLightPhase(filePath: string, current: number, queueSize: number, providedMidHash256?: string) { try { // STATE: Start LIGHT processing if (this.unifiedStateManager) { this.unifiedStateManager.startLightProcessing(filePath); } - const stats = await stat(filePath); + const stats = await webdav.stat(filePath); + if (!stats || !stats.exists) { + throw new Error(`File not found via WebDAV: ${filePath}`); + } - // STEP 1: Check cache for midhash256 FIRST (avoids network I/O on unchanged files) + // STEP 1: Use midhash256 from meta-core if provided, otherwise check cache/compute locally let midHash256: string | undefined; - const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); - if (indexLine && indexLine.cid_midhash256) { - // Cache hit! Use cached midhash256 (no network I/O needed) - midHash256 = indexLine.cid_midhash256; + if (providedMidHash256) { + // midhash256 provided by meta-core - use directly (no file I/O needed) + midHash256 = providedMidHash256; performanceMetrics.recordCacheHit('midhash256'); - } else { - // Cache miss - compute midhash256 (< 1 second, but requires network I/O on network drives) - const computeStart = performance.now(); - const { computeMidHash256 } = await import('@metazla/meta-hash'); - midHash256 = await computeMidHash256(filePath); - const computeTime = Math.ceil(performance.now() - computeStart); - // Record midhash256 computation time (excludes cache hits) - performanceMetrics.recordHashComputation('cid_midhash256', computeTime); - performanceMetrics.recordCacheMiss('midhash256'); - - // Store in cache for future lookups + // Store in local cache for future lookups this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 }); + } else { + // Fallback: Check cache or compute locally (legacy path) + const indexLine = this.indexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); + + if (indexLine && indexLine.cid_midhash256) { + // Cache hit! Use cached midhash256 (no network I/O needed) + midHash256 = indexLine.cid_midhash256; + performanceMetrics.recordCacheHit('midhash256'); + } else { + // Cache miss - compute midhash256 locally (requires file access) + // Note: This path should not be hit in normal operation since meta-core provides midhash256 + console.warn(`[FileProcessor] Computing midhash256 locally for ${filePath} - meta-core should provide this`); + const computeStart = performance.now(); + const { computeMidHash256 } = await import('@metazla/meta-hash'); + midHash256 = await computeMidHash256(filePath); + const computeTime = Math.ceil(performance.now() - computeStart); + + // Record midhash256 computation time (excludes cache hits) + performanceMetrics.recordHashComputation('cid_midhash256', computeTime); + performanceMetrics.recordCacheMiss('midhash256'); + + // Store in cache for future lookups + this.indexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), { cid_midhash256: midHash256 }); + } } // Initialize metadata with midhash256 @@ -451,7 +475,10 @@ export class FileProcessorPiscina implements FileAnalyzerInterface{ this.unifiedStateManager.startHashProcessing(filePath); } - const stats = await stat(filePath); + const stats = await webdav.stat(filePath); + if (!stats || !stats.exists) { + throw new Error(`File not found via WebDAV: ${filePath}`); + } 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/fileProcessor/SimpleFileToCid.ts b/packages/meta-sort-core/src/logic/fileProcessor/SimpleFileToCid.ts index e2fb83a..93d2a6e 100644 --- a/packages/meta-sort-core/src/logic/fileProcessor/SimpleFileToCid.ts +++ b/packages/meta-sort-core/src/logic/fileProcessor/SimpleFileToCid.ts @@ -2,7 +2,7 @@ import {CID_ALGORITHM_NAMES, HashComputerFile} from "@metazla/meta-hash"; import {HashIndexManager} from "@metazla/meta-hash"; import {config} from "../../config/EnvConfig.js"; import {targetHash, targetHashForIndex} from "../../config/TargetHash.js"; -import {stat} from "fs/promises"; +import * as webdav from '../../webdav/WebdavClient.js'; export const globalHashIndex = new HashIndexManager(config.INDEX_FOLDER_PATH, targetHashForIndex); @@ -12,7 +12,10 @@ class SimpleFileToCid{ async getcid(filePath: string): Promise { try { - const stats = await stat(filePath); + const stats = await webdav.stat(filePath); + if (!stats || !stats.exists) { + return null; + } let indexLine = globalHashIndex.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); if(indexLine) { return indexLine[CID_ALGORITHM_NAMES.sha256]; diff --git a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts index 9886b6b..773bde9 100644 --- a/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts +++ b/packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts @@ -91,8 +91,10 @@ export class StreamingPipeline { * - Optional MIME validation (reads file header) * - Files already marked as discovered in start() * - After validation passes, file is removed from discovered (tracked by PQueue pending) + * @param filePath - Path to the file + * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ - private async validateFile(filePath: string): Promise { + private async validateFile(filePath: string, midhash256?: string): Promise { try { // Extension check (instant, no I/O) const ext = path.extname(filePath).toLowerCase(); @@ -113,7 +115,8 @@ export class StreamingPipeline { this.config.stateManager.removeFile(filePath); // Stage 2: Fast Queue - metadata + midhash256 (fire and forget) - this.fastQueue.add(() => this.processLightPhase(filePath)) + // Pass midhash256 from meta-core if available + this.fastQueue.add(() => this.processLightPhase(filePath, midhash256)) .catch(err => console.error(`[Pipeline] Fast queue error for ${filePath}:`, err.message)); } catch (error: any) { console.error(`[Pipeline] Validation failed for ${filePath}:`, error.message); @@ -125,17 +128,19 @@ export class StreamingPipeline { /** * Stage 2: Fast Queue Phase * - Extract metadata (filename parsing, FFmpeg analysis) - * - Compute midhash256 (< 1s) + * - Use midhash256 from meta-core (or compute locally if not provided) * - Store to KV (file becomes accessible in VFS) * - Queue for background processing + * @param filePath - Path to the file + * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ - private async processLightPhase(filePath: string): Promise { + private async processLightPhase(filePath: string, midhash256?: string): Promise { try { - // Call light processing + // Call light processing with midhash256 from meta-core const current = this.fastProcessedCount + 1; const queueSize = this.discoveredCount; - await this.config.fileProcessor.processLightPhase(filePath, current, queueSize); + await this.config.fileProcessor.processLightPhase(filePath, current, queueSize, midhash256); // After light processing, add to VFS const metadata = this.config.fileProcessor.getDatabase().get(filePath); @@ -231,23 +236,27 @@ export class StreamingPipeline { } /** - * Handle file added event (from chokidar) + * Handle file added event (from FileEventConsumer via meta-core) + * @param filePath - Path to the file + * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ - async handleFileAdded(filePath: string): Promise { + async handleFileAdded(filePath: string, midhash256?: string): Promise { - // Process through validation stage - this.validationQueue.add(() => this.validateFile(filePath)) + // Process through validation stage, passing midhash256 + this.validationQueue.add(() => this.validateFile(filePath, midhash256)) .catch(err => console.error(`[Pipeline] Error processing added file ${filePath}:`, err.message)); } /** - * Handle file changed event (from chokidar) + * Handle file changed event (from FileEventConsumer via meta-core) + * @param filePath - Path to the file + * @param midhash256 - Pre-computed midhash256 from meta-core (if available) */ - async handleFileChanged(filePath: string): Promise { + async handleFileChanged(filePath: string, midhash256?: string): Promise { // Reprocess through validation stage this.config.stateManager.removeFile(filePath); - this.validationQueue.add(() => this.validateFile(filePath)) + this.validationQueue.add(() => this.validateFile(filePath, midhash256)) .catch(err => console.error(`[Pipeline] Error processing changed file ${filePath}:`, err.message)); }