metastar/meta-sort
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move feature from meta-sort to meta-core

Browse Source
This commit is contained in:
Pierre
2026-01-25 22:02:30 +01:00
parent 759009c019
commit c7abc09509
44 changed files with 488 additions and 8073 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/meta-sort-core/package.json
+2
View File
@@ -25,6 +25,7 @@
"chokidar": "^3.5.3",
"dockerode": "^4.0.2",
"dotenv": "^16.4.1",
"eventsource": "^4.1.0",
"fastify": "^5.6.1",
"fluent-ffmpeg": "^2.1.2",
"franc-min": "^6.2.0",
@@ -42,6 +43,7 @@
"devDependencies": {
"@types/chai": "^4.3.11",
"@types/dockerode": "^3.3.31",
"@types/eventsource": "^3.0.0",
"@types/fluent-ffmpeg": "^2.1.24",
"@types/lodash": "^4.17.0",
"@types/mocha": "^10.0.6",
packages/meta-sort-core/src/api/UnifiedAPIServer.ts
+36 -992
View File
File diff suppressed because it is too large Load Diff
packages/meta-sort-core/src/api/api.test.ts
+2 -2
View File
@@ -839,10 +839,10 @@ describe('UnifiedAPIServer', function() {
});
// =========================================================================
// Mounts API (Mounts Page)
// Mounts API (Mounts Page) - SKIPPED in Architecture V3 (moved to meta-core)
// =========================================================================
describe('Mounts API (/api/mounts/*)', function() {
describe.skip('Mounts API (/api/mounts/*)', function() {
it('GET /api/mounts returns mount list', async function() {
const response = await app.inject({
method: 'GET',
packages/meta-sort-core/src/config/EnvConfig.ts
+20 -62
View File
@@ -1,24 +1,13 @@
import dotenv from 'dotenv';
dotenv.config();
export interface PollingPathConfig {
path: string;
intervalMs: number;
}
interface EnvConfig {
/** Comma-separated list of folders to watch for changes.(mandatory) */
WATCH_FOLDER_LIST: string;
/** Path to the index file within the NFO CID library. */
INDEX_FOLDER_PATH: string;
/** Interval in milliseconds to update/check for changes. (default 30000) */
UPDATE_INTERVAL_MS: number;
/** Interval in milliseconds for polling watch folders. Set to 0 to disable polling. (default 0) */
WATCH_FOLDER_POOLING_INTERVAL_MS: number;
/** CHMOD for new folder format (default 0o777) */
CHMOD_FOR_NEW_FOLDER: number;
@@ -41,11 +30,8 @@ interface EnvConfig {
/** Maximum number of worker threads for Piscina worker pool AND max concurrent file processing (default: CPU count) */
MAX_WORKER_THREADS?: number;
/** JSON array of polling path configurations. Example: '[{"path": "/data/watch/smb/downloads", "intervalMs": 60000}]' (optional) */
POLLING_PATHS?: PollingPathConfig[];
// ========================================================================
// KV Storage Configuration (NEW)
// KV Storage Configuration
// ========================================================================
/** Path to META_CORE_VOLUME - infrastructure volume for KV DB, locks, services (default '/meta-core') */
@@ -123,54 +109,30 @@ interface EnvConfig {
// NOTE: Plugin output files are written via WebDAV (WEBDAV_URL/plugin/<pluginId>/)
// No PLUGIN_OUTPUT_HOST_PATH needed - plugins use HTTP PUT to write files
}
// Parse POLLING_PATHS from JSON string
function parsePollingPaths(): PollingPathConfig[] | undefined {
if (!process.env.POLLING_PATHS) {
return undefined;
}
// ========================================================================
// Event Subscriber Configuration (Architecture V3)
// ========================================================================
try {
const parsed = JSON.parse(process.env.POLLING_PATHS);
if (!Array.isArray(parsed)) {
console.error('POLLING_PATHS must be a JSON array');
return undefined;
}
// Validate structure
const valid = parsed.every(item =>
item && typeof item === 'object' &&
typeof item.path === 'string' &&
typeof item.intervalMs === 'number'
);
if (!valid) {
console.error('POLLING_PATHS entries must have {path: string, intervalMs: number}');
return undefined;
}
return parsed as PollingPathConfig[];
} catch (error) {
console.error('Failed to parse POLLING_PATHS:', error);
return undefined;
}
/**
* URL of meta-core API for event subscription (SSE).
* REQUIRED: meta-sort receives file events from meta-core via SSE.
* Example: "http://meta-core" or "http://localhost:8083"
*/
META_CORE_URL?: string;
}
export const config: EnvConfig = {
WATCH_FOLDER_LIST: process.env.WATCH_FOLDER_LIST,
INDEX_FOLDER_PATH: process.env.INDEX_FOLDER_PATH || '/data/cache/hash-index',//default /data/cache/hash-index
UPDATE_INTERVAL_MS: parseInt(process.env.UPDATE_INTERVAL_MS || "30000", 10),//default 30000
WATCH_FOLDER_POOLING_INTERVAL_MS: parseInt(process.env.WATCH_FOLDER_POOLING_INTERVAL_MS || "0", 10),//default 0
CHMOD_FOR_NEW_FOLDER: process.env.CHMOD_FOR_NEW_FOLDER && parseInt(process.env.CHMOD_FOR_NEW_FOLDER, 8),//default 0o777 (this is the default for mkdir)
CACHE_FOLDER_PATH: process.env.CACHE_FOLDER_PATH || '/data/cache',//default /data/cache
INDEX_FOLDER_PATH: process.env.INDEX_FOLDER_PATH || '/data/cache/hash-index',
UPDATE_INTERVAL_MS: parseInt(process.env.UPDATE_INTERVAL_MS || "30000", 10),
CHMOD_FOR_NEW_FOLDER: process.env.CHMOD_FOR_NEW_FOLDER && parseInt(process.env.CHMOD_FOR_NEW_FOLDER, 8),
CACHE_FOLDER_PATH: process.env.CACHE_FOLDER_PATH || '/data/cache',
JELLYFIN_ENDPOINT: process.env.JELLYFIN_ENDPOINT!,
JELLYFIN_API_KEY: process.env.JELLYFIN_API_KEY!,
FUSE_API_PORT: parseInt(process.env.FUSE_API_PORT || "3000", 10),//default 3000
FUSE_API_HOST: process.env.FUSE_API_HOST || '0.0.0.0',//default 0.0.0.0
METADATA_FORMATS: (process.env.METADATA_FORMATS || 'meta').split(',').map(f => f.trim()).filter(f => f.length > 0),//default 'meta'
FUSE_API_PORT: parseInt(process.env.FUSE_API_PORT || "3000", 10),
FUSE_API_HOST: process.env.FUSE_API_HOST || '0.0.0.0',
METADATA_FORMATS: (process.env.METADATA_FORMATS || 'meta').split(',').map(f => f.trim()).filter(f => f.length > 0),
MAX_WORKER_THREADS: process.env.MAX_WORKER_THREADS ? parseInt(process.env.MAX_WORKER_THREADS, 10) : undefined,
POLLING_PATHS: parsePollingPaths(),
// KV Storage Configuration
META_CORE_PATH: process.env.META_CORE_PATH || '/meta-core',
@@ -194,11 +156,7 @@ export const config: EnvConfig = {
PLUGIN_WEBDAV_URL: process.env.PLUGIN_WEBDAV_URL,
PLUGIN_STACK_NAME: process.env.PLUGIN_STACK_NAME,
PLUGIN_CACHE_HOST_PATH: process.env.PLUGIN_CACHE_HOST_PATH,
// Event Subscriber Configuration
META_CORE_URL: process.env.META_CORE_URL,
};
if(process.env.TEST!=='true') {
if (!config.WATCH_FOLDER_LIST) {
console.error('Invalid configuration: WATCH_FOLDER_LIST is required');
}
}
packages/meta-sort-core/src/container-plugins/ContainerManager.ts
+42 -16
View File
@@ -119,9 +119,31 @@ export class ContainerManager extends EventEmitter {
/**
* Discover service URLs from meta-core service discovery file
* Falls back to environment variables if discovery fails
* Explicit env vars (CONTAINER_CALLBACK_URL, PLUGIN_WEBDAV_URL) take priority
* Service discovery is only used as fallback
*/
private async discoverServiceUrls(): Promise<void> {
// Check if explicit env vars are set - these take priority for container communication
const explicitCallbackUrl = config.CONTAINER_CALLBACK_URL;
const explicitWebdavUrl = config.PLUGIN_WEBDAV_URL;
if (explicitCallbackUrl) {
this.callbackUrl = explicitCallbackUrl;
console.log(`[ContainerManager] Using explicit callback URL: ${this.callbackUrl}`);
}
if (explicitWebdavUrl) {
this.webdavUrl = explicitWebdavUrl;
console.log(`[ContainerManager] Using explicit WebDAV URL: ${this.webdavUrl}`);
}
// If both are explicitly set, skip service discovery
if (explicitCallbackUrl && explicitWebdavUrl) {
console.log('[ContainerManager] Using explicit env vars, skipping service discovery');
return;
}
// Fall back to service discovery for any missing URLs
const serviceFile = join(config.META_CORE_PATH, 'services', 'meta-sort.json');
try {
@@ -132,23 +154,27 @@ export class ContainerManager extends EventEmitter {
const api = serviceInfo.api;
const endpoints = serviceInfo.endpoints || {};
// Update URLs from service discovery
if (endpoints.callback) {
this.callbackUrl = endpoints.callback;
console.log(`[ContainerManager] Discovered callback URL: ${this.callbackUrl}`);
} else if (api) {
// Fallback: construct from api base
this.callbackUrl = `${api}/api/plugins/callback`;
console.log(`[ContainerManager] Constructed callback URL: ${this.callbackUrl}`);
// Only use service discovery if env var not set
if (!explicitCallbackUrl) {
if (endpoints.callback) {
this.callbackUrl = endpoints.callback;
console.log(`[ContainerManager] Discovered callback URL: ${this.callbackUrl}`);
} else if (api) {
// Fallback: construct from api base
this.callbackUrl = `${api}/api/plugins/callback`;
console.log(`[ContainerManager] Constructed callback URL: ${this.callbackUrl}`);
}
}
if (endpoints.webdav) {
this.webdavUrl = endpoints.webdav;
console.log(`[ContainerManager] Discovered WebDAV URL: ${this.webdavUrl}`);
} else if (api) {
// Fallback: construct from api base
this.webdavUrl = `${api}/webdav`;
console.log(`[ContainerManager] Constructed WebDAV URL: ${this.webdavUrl}`);
if (!explicitWebdavUrl) {
if (endpoints.webdav) {
this.webdavUrl = endpoints.webdav;
console.log(`[ContainerManager] Discovered WebDAV URL: ${this.webdavUrl}`);
} else if (api) {
// Fallback: construct from api base
this.webdavUrl = `${api}/webdav`;
console.log(`[ContainerManager] Constructed WebDAV URL: ${this.webdavUrl}`);
}
}
if (endpoints.health) {
packages/meta-sort-core/src/events/EventSubscriber.ts
+292
View File
@@ -0,0 +1,292 @@
/**
* EventSubscriber
*
* Subscribes to file events from meta-core via SSE (Server-Sent Events).
* When events are received, they are forwarded to the StreamingPipeline.
*
* This enables meta-sort to run without direct filesystem access - it receives
* file discovery events from meta-core and processes files via WebDAV.
*/
import { EventSource } from 'eventsource';
// File event types matching meta-core's watcher
interface FileEvent {
type: 'add' | 'change' | 'delete' | 'rename';
path: string; // Relative to FILES_PATH
size?: number;
timestamp: number;
partialHash?: string; // Hash of first 64KB
oldPath?: string; // For rename events
}
// Pipeline interface (minimal, matching StreamingPipeline)
interface Pipeline {
handleFileAdded(filePath: string): void;
handleFileChanged(filePath: string): void;
handleFileDeleted(filePath: string): void;
}
export interface EventSubscriberOptions {
/** URL of meta-core API (e.g., http://meta-core or http://localhost:8083) */
metaCoreUrl: string;
/** Pipeline to receive events */
pipeline: Pipeline;
/** Base path for files (e.g., /files). Paths from meta-core are relative and need this prefix. */
filesPath?: string;
/** Whether to request an initial scan when connecting */
requestInitialScan?: boolean;
/** Reconnect delay in milliseconds */
reconnectDelayMs?: number;
/** Maximum reconnect attempts (0 = unlimited) */
maxReconnectAttempts?: number;
}
export class EventSubscriber {
private options: Required<EventSubscriberOptions>;
private eventSource: EventSource | null = null;
private isRunning = false;
private reconnectAttempts = 0;
private reconnectTimer: NodeJS.Timeout | null = null;
constructor(options: EventSubscriberOptions) {
this.options = {
filesPath: '/files',
requestInitialScan: true,
reconnectDelayMs: 5000,
maxReconnectAttempts: 0, // Unlimited by default
...options
};
}
/**
* Start subscribing to events from meta-core
*/
async start(): Promise<void> {
if (this.isRunning) {
console.log('[EventSubscriber] Already running');
return;
}
this.isRunning = true;
console.log(`[EventSubscriber] Connecting to ${this.options.metaCoreUrl}/api/events/subscribe`);
await this.connect();
}
/**
* Stop subscribing to events
*/
async stop(): Promise<void> {
this.isRunning = false;
if (this.reconnectTimer) {
clearTimeout(this.reconnectTimer);
this.reconnectTimer = null;
}
if (this.eventSource) {
this.eventSource.close();
this.eventSource = null;
}
console.log('[EventSubscriber] Stopped');
}
/**
* Trigger a manual scan on meta-core
*/
async triggerScan(): Promise<void> {
try {
const response = await fetch(`${this.options.metaCoreUrl}/api/scan/trigger`, {
method: 'POST'
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
console.log('[EventSubscriber] Scan triggered on meta-core');
} catch (error) {
console.error('[EventSubscriber] Failed to trigger scan:', error);
}
}
/**
* Get watcher status from meta-core
*/
async getStatus(): Promise<{
status: string;
scanning: boolean;
lastScan: number;
fileCount: number;
} | null> {
try {
const response = await fetch(`${this.options.metaCoreUrl}/api/scan/status`);
if (!response.ok) {
return null;
}
return await response.json();
} catch (error) {
console.error('[EventSubscriber] Failed to get status:', error);
return null;
}
}
/**
* Poll for events since a timestamp (fallback for SSE)
*/
async pollEvents(sinceMs: number, limit = 100): Promise<FileEvent[]> {
try {
const url = `${this.options.metaCoreUrl}/api/events/poll?since=${sinceMs}&limit=${limit}`;
const response = await fetch(url);
if (!response.ok) {
return [];
}
const data = await response.json();
return data.events || [];
} catch (error) {
console.error('[EventSubscriber] Failed to poll events:', error);
return [];
}
}
private async connect(): Promise<void> {
if (!this.isRunning) return;
const url = `${this.options.metaCoreUrl}/api/events/subscribe`;
try {
this.eventSource = new EventSource(url);
this.eventSource.onopen = () => {
console.log('[EventSubscriber] Connected to meta-core SSE');
this.reconnectAttempts = 0;
// Request initial scan if configured
if (this.options.requestInitialScan) {
this.triggerScan().catch(err => {
console.error('[EventSubscriber] Failed to trigger initial scan:', err);
});
}
};
this.eventSource.onerror = (error) => {
console.error('[EventSubscriber] SSE connection error:', error);
this.scheduleReconnect();
};
// Handle file events
this.eventSource.addEventListener('file', (event: MessageEvent) => {
try {
const fileEvent: FileEvent = JSON.parse(event.data);
this.handleEvent(fileEvent);
} catch (error) {
console.error('[EventSubscriber] Failed to parse event:', error);
}
});
// Handle connected event
this.eventSource.addEventListener('connected', () => {
console.log('[EventSubscriber] Received connected event');
});
} catch (error) {
console.error('[EventSubscriber] Failed to connect:', error);
this.scheduleReconnect();
}
}
private handleEvent(event: FileEvent): void {
const { pipeline, filesPath } = this.options;
// Convert relative path to absolute path
const toAbsolutePath = (relativePath: string): string => {
// If path is already absolute, return as-is
if (relativePath.startsWith('/')) {
return relativePath;
}
// Otherwise, prepend filesPath
return `${filesPath}/${relativePath}`;
};
const absolutePath = toAbsolutePath(event.path);
switch (event.type) {
case 'add':
pipeline.handleFileAdded(absolutePath);
break;
case 'change':
pipeline.handleFileChanged(absolutePath);
break;
case 'delete':
pipeline.handleFileDeleted(absolutePath);
break;
case 'rename':
// Handle rename as delete + add
if (event.oldPath) {
pipeline.handleFileDeleted(toAbsolutePath(event.oldPath));
}
pipeline.handleFileAdded(absolutePath);
break;
default:
console.warn(`[EventSubscriber] Unknown event type: ${event.type}`);
}
}
private scheduleReconnect(): void {
if (!this.isRunning) return;
// Close existing connection
if (this.eventSource) {
this.eventSource.close();
this.eventSource = null;
}
// Check reconnect limits
if (this.options.maxReconnectAttempts > 0 &&
this.reconnectAttempts >= this.options.maxReconnectAttempts) {
console.error('[EventSubscriber] Max reconnect attempts reached, giving up');
this.isRunning = false;
return;
}
this.reconnectAttempts++;
const delay = this.options.reconnectDelayMs * Math.min(this.reconnectAttempts, 6); // Max 30s
console.log(`[EventSubscriber] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
this.reconnectTimer = setTimeout(() => {
this.reconnectTimer = null;
this.connect();
}, delay);
}
}
/**
* Create an EventSubscriber with default configuration
*/
export function createEventSubscriber(
metaCoreUrl: string,
pipeline: Pipeline,
options?: Partial<EventSubscriberOptions>
): EventSubscriber {
return new EventSubscriber({
metaCoreUrl,
pipeline,
...options
});
}
packages/meta-sort-core/src/index.ts
+37 -78
View File
@@ -1,4 +1,3 @@
import {FolderWatcher, PollingWatcher} from "@metazla/meta-hash";
import {config} from "./config/EnvConfig.js";
import os from 'os';
@@ -12,6 +11,7 @@ import {KVManager} from "./kv/KVManager.js";
import {getPluginManagerInstance, initializePluginManager} from "./logic/fileProcessor/FileProcessorPiscina.js";
import {ContainerManager} from "./container-plugins/ContainerManager.js";
import {ContainerPluginScheduler} from "./container-plugins/ContainerPluginScheduler.js";
import {EventSubscriber} from "./events/EventSubscriber.js";
try {
// Get current user information
@@ -31,7 +31,6 @@ console.log(config);
process.umask(0);//This unable the process to create files with 777 permissions
let fileProcessor = new WatchedFileProcessor();
let folders = config.WATCH_FOLDER_LIST;
// KV Manager for Redis-based storage with leader election (optional)
let kvManager: KVManager | null = null;
@@ -40,6 +39,9 @@ let kvManager: KVManager | null = null;
let containerManager: ContainerManager | null = null;
let containerPluginScheduler: ContainerPluginScheduler | null = null;
// Event Subscriber for receiving events from meta-core (Architecture V3)
let eventSubscriber: EventSubscriber | null = null;
// Calculate concurrency for pipeline stages
const cpuCount = os.cpus().length;
// Smart defaults optimized for network storage + HDD scenarios:
@@ -69,13 +71,6 @@ const pipeline = new StreamingPipeline({
// Connect pipeline to fileProcessor for queue status monitoring
fileProcessor.setPipeline(pipeline);
// Create generic file discovery
const folderWatcher = new FolderWatcher();
// Parse folder list
const folderList = folders.split(',').map(f => f.trim()).filter(f => f.length > 0);
console.log(`Watching folders: ${folderList.join(', ')}`);
// VFS and state manager are available immediately
const vfs = fileProcessor.getVirtualFileSystem();
const unifiedStateManager = fileProcessor.getUnifiedStateManager();
@@ -84,52 +79,21 @@ const getDuplicateResult = () => fileProcessor.getDuplicateResult();
// API Server will be initialized after KV is ready (deferred to async init)
let apiServer: UnifiedAPIServer | null = null;
// Shared progressive discovery and cleanup logic (used by both startup and manual scan)
const runProgressiveDiscovery = async () => {
console.log(`[Discovery] Progressive scan mode: ${folderList.length} folders`);
// Process folders one at a time for progressive cleanup
for (const folder of folderList) {
console.log(`[Discovery] Scanning folder: ${folder}`);
// Track discovered files for this folder
const discoveredFilesInFolder = new Set<string>();
// Wrap discovery stream to track files
async function* trackDiscoveredFiles(stream: AsyncGenerator<string>) {
for await (const filePath of stream) {
discoveredFilesInFolder.add(filePath);
yield filePath;
}
}
const discoveryStream = folderWatcher.discoverFiles([folder]);
const trackedStream = trackDiscoveredFiles(discoveryStream);
await pipeline.start(trackedStream);
// Progressive cleanup: clean up after each folder completes
console.log(`[Discovery] Folder scan complete: ${folder}. Found ${discoveredFilesInFolder.size} files. Running cleanup...`);
await fileProcessor.cleanupStaleEntries([folder], discoveredFilesInFolder);
console.log(`[Discovery] Cleanup complete for folder: ${folder}`);
}
console.log(`[Discovery] All folders scanned and cleaned up`);
};
// Manual scan trigger (called from API)
// Note: Stremio updates happen automatically via incremental add/remove notifications
// during file processing (like FUSE), so no explicit refresh needed here
// Manual scan trigger (called from API) - delegates to meta-core via EventSubscriber
const triggerScan = async () => {
console.log('[Scan] Triggering manual scan from API...');
console.log('[Scan] Triggering manual scan...');
// 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');
// Trigger scan on meta-core (which will send events back via SSE)
if (eventSubscriber) {
await eventSubscriber.triggerScan();
console.log('[Scan] Scan triggered on meta-core');
} else {
console.warn('[Scan] EventSubscriber not available, cannot trigger scan');
}
};
// Global error handlers - prevent crashes from unforeseen errors
@@ -153,6 +117,12 @@ process.on('SIGINT', async () => {
await apiServer.stop();
}
// Stop Event Subscriber if running
if (eventSubscriber) {
console.log('[Shutdown] Stopping Event Subscriber...');
await eventSubscriber.stop();
}
// Stop Container Manager if running
if (containerManager) {
console.log('[Shutdown] Stopping Container Manager...');
@@ -171,6 +141,13 @@ process.on('SIGINT', async () => {
// Start initial discovery and processing (streaming pipeline)
(async () => {
try {
// Validate META_CORE_URL is configured
if (!config.META_CORE_URL) {
console.error('[Startup] ERROR: META_CORE_URL is required. Set META_CORE_URL environment variable.');
console.error('[Startup] Example: META_CORE_URL=http://meta-core');
process.exit(1);
}
// Initialize KV Manager if configured (Redis + leader election)
if (config.META_CORE_PATH) {
console.log('[Startup] Initializing KV Manager (Redis + leader election)...');
@@ -278,10 +255,6 @@ process.on('SIGINT', async () => {
await fileProcessor.rebuildVFSFromKV();
console.log('[Startup] VFS rebuild completed');
console.log('[Startup] Starting initial file discovery with progressive cleanup...');
await runProgressiveDiscovery();
console.log('[Startup] Initial discovery completed, processing continues in background');
// Log cache performance summary
const metrics = performanceMetrics.getMetrics();
if (metrics.cacheHits.midhash256 || metrics.cacheMisses.midhash256) {
@@ -292,32 +265,18 @@ process.on('SIGINT', async () => {
console.log(`[Cache] midhash256 cache: ${hits} hits, ${misses} misses (${hitRate}% hit rate)`);
}
// Setup file watching for changes
console.log('[Startup] Setting up file watching...');
folderWatcher.watch(folderList, {
onAdd: (filePath) => pipeline.handleFileAdded(filePath),
onChange: (filePath) => pipeline.handleFileChanged(filePath),
onUnlink: (filePath) => pipeline.handleFileDeleted(filePath)
}, {
interval: config.WATCH_FOLDER_POOLING_INTERVAL_MS || 30000,
stabilityThreshold: 30000,
pollInterval: 5000
// Connect to meta-core for file events via SSE (Architecture V3)
console.log(`[Startup] Connecting to meta-core at ${config.META_CORE_URL}...`);
eventSubscriber = new EventSubscriber({
metaCoreUrl: config.META_CORE_URL,
pipeline: pipeline,
requestInitialScan: true, // Request initial scan from meta-core
reconnectDelayMs: 5000,
});
console.log('[Startup] File watching active');
await eventSubscriber.start();
console.log('[Startup] EventSubscriber connected to meta-core');
} catch (error) {
console.error('[Startup] Error during initial discovery:', error);
console.error('[Startup] Error during startup:', error);
}
})();
// Start PollingWatcher for path-specific polling (if configured)
if (config.POLLING_PATHS && config.POLLING_PATHS.length > 0) {
console.log(`PollingWatcher: ${config.POLLING_PATHS.length} paths configured`);
const pollingWatcher = new PollingWatcher(fileProcessor, config.POLLING_PATHS);
pollingWatcher.start().then(() => {
console.log('PollingWatcher started successfully');
}).catch((error) => {
console.error('Error starting PollingWatcher:', error);
});
} else {
console.log('PollingWatcher: No polling paths configured (skipped)');
}
packages/meta-sort-core/src/logic/pipeline/StreamingPipeline.ts
+27 -8
View File
@@ -527,18 +527,37 @@ export class StreamingPipeline {
(this.config as any).containerPluginScheduler = scheduler;
// Listen for file:complete events to transition files to done state
scheduler.on('file:complete', ({ fileHash, filePath }) => {
scheduler.on('file:complete', async ({ fileHash, filePath }) => {
console.log(`[Pipeline] Container plugins complete for ${filePath} (${fileHash})`);
// Get virtual path from metadata if available
// Fetch metadata from Redis (container plugins write directly to Redis via /meta API)
let virtualPath: string | undefined;
const metadata = this.config.fileProcessor?.getDatabase().get(filePath);
if (metadata) {
try {
virtualPath = this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath);
} catch {
// Renaming rule may fail for incomplete metadata
let metadata: Record<string, any> | undefined;
try {
if (this.config.kvClient) {
// Get fresh metadata from Redis
const redisMetadata = await this.config.kvClient.getMetadataFlat(fileHash);
if (redisMetadata && Object.keys(redisMetadata).length > 0) {
metadata = redisMetadata;
// Update local database to stay in sync
if (this.config.fileProcessor?.getDatabase()) {
this.config.fileProcessor.getDatabase().set(filePath, metadata);
}
}
}
// Fallback to local database if Redis fetch failed
if (!metadata) {
metadata = this.config.fileProcessor?.getDatabase().get(filePath);
}
if (metadata) {
virtualPath = this.config.metaDataToFolderStruct.renamingRule(metadata as any, filePath);
}
} catch (error) {
// Renaming rule may fail for incomplete metadata
console.warn(`[Pipeline] Failed to compute virtualPath for ${filePath}:`, error);
}
// Transition file to done state