From 9438ed973b4b99f01b43e4b1d5f67a7334db5618 Mon Sep 17 00:00:00 2001 From: Pierre Date: Mon, 3 Nov 2025 11:30:39 +0100 Subject: [PATCH] improvement in folder watcher process --- src/lib/folder-watcher/FolderWatcher.ts | 316 +++++++++-------------- src/lib/folder-watcher/PollingWatcher.ts | 168 ++++++++++++ src/lib/index.ts | 1 + 3 files changed, 297 insertions(+), 188 deletions(-) create mode 100644 src/lib/folder-watcher/PollingWatcher.ts diff --git a/src/lib/folder-watcher/FolderWatcher.ts b/src/lib/folder-watcher/FolderWatcher.ts index bb32086..8acc4b3 100644 --- a/src/lib/folder-watcher/FolderWatcher.ts +++ b/src/lib/folder-watcher/FolderWatcher.ts @@ -1,230 +1,170 @@ import * as path from 'path'; import chokidar from 'chokidar'; import {readdir} from "fs/promises"; -import PQueue from "p-queue"; -import debounce from 'debounce'; -import {FileProcessorInterface} from "./FileProcessorInterface.js"; +/** + * Generic file discovery utility + * Discovers files in directory trees and emits paths as async generator + * No filtering logic - just pure file discovery + */ export class FolderWatcher { - initialized = false; - queue: PQueue; // Queue for basic concurrency control during file discovery - fileProcessor: FileProcessorInterface; - queueSize = 0; - current = 0; - processing = new Set(); - private readonly finalizeDebounced: () => Promise; - - constructor(fileProcessor: FileProcessorInterface, private WATCH_FOLDER_LIST: string, private config: { - interval?: number, - stabilityThreshold?: number, - pollInterval?: number, - outputFolderUpdateIntervalMs?: number, - concurrency?: number - }) { - // Initialize queue for basic concurrency control - // Prevents overwhelming the system with too many concurrent file discoveries - const concurrency = config.concurrency || 4; - this.queue = new PQueue({concurrency, autoStart: true}); - console.log(`FolderWatcher initialized with ${concurrency} workers for file processing`); - this.fileProcessor = fileProcessor; - if (this.fileProcessor.finalize) { - const debouncedImmediate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true}); - const debouncedLate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: false}); - this.finalizeDebounced = async () => { - if (this.initialized) { - const ret1 = debouncedImmediate(); - const ret2 = debouncedLate(); - this.current = 0; - this.queueSize = 0; - await Promise.all([ret1, ret2]); - } - } + /** + * Discover all files in given directories + * Returns async generator that yields file paths as they're discovered + * + * @param directories - Array of directory paths to scan + * @returns AsyncGenerator yielding file paths + */ + async *discoverFiles(directories: string[]): AsyncGenerator { + for (const dir of directories) { + const normalizedDir = path.normalize(dir); + yield* this.walkDirectory(normalizedDir); } - } /** - * Process the directory and its subdirectories - * @param directory - * @param promises - * @private + * Recursively walk directory tree + * Yields file paths as they're discovered + * + * @param directory - Directory to walk + * @returns AsyncGenerator yielding file paths */ - private async processDirectory(directory: string, promises: Promise[]) { - const entries = await readdir(directory, {withFileTypes: true}); - for (const entry of entries) { - const fullPath = path.join(directory, entry.name); - if (entry.isDirectory()) { - await this.processDirectory(fullPath, promises); - } else { - const promise = this.processFile(fullPath); - promises.push(promise); - } - } - } + private async *walkDirectory(directory: string): AsyncGenerator { + try { + const entries = await readdir(directory, {withFileTypes: true}); - private async countFile(directory: string): Promise { - const entries = await readdir(directory, {withFileTypes: true}); - const countPromises: Promise[] = []; + // Collect subdirectories for parallel processing + const subdirs: string[] = []; - for (const entry of entries) { - const fullPath = path.join(directory, entry.name); - if (entry.isDirectory()) { - // Push the promise without waiting for it here - countPromises.push(this.countFile(fullPath)); - } else { - // Check if file can be processed (no queue needed for counting) - countPromises.push( - (async () => (await this.fileProcessor.canProcessFile(fullPath)) ? 1 : 0)() - ); - } - } + for (const entry of entries) { + const fullPath = path.join(directory, entry.name); - // Wait for all promises to resolve and then sum up their results - const counts = await Promise.all(countPromises); - return counts.reduce((acc, current) => acc + current, 0); - } - - private async processFile(filePath: string): Promise { - if (!this.processing.has(filePath)) { - this.processing.add(filePath);//avoid double processing in the queue - if (await this.fileProcessor.canProcessFile(filePath)) { - const current = ++this.current; - - // Mark as pending BEFORE adding to queue - // This ensures files appear in "pending" state while waiting for queue - if (this.fileProcessor.markPending) { - this.fileProcessor.markPending(filePath); + if (entry.isDirectory()) { + subdirs.push(fullPath); + } else { + // Yield file path immediately + yield fullPath; } - - // Add to queue for concurrency control - // Queue only controls light processing throughput - hash processing runs independently - await this.queue.add(async () => { - // Light processing (quick metadata extraction) - if (this.fileProcessor.queueFile) { - await this.fileProcessor.queueFile(filePath); - } - - // Heavy processing (hash computation) - fire and forget - // Don't await - let hashQueue handle its own concurrency independently - if (this.fileProcessor.processFile) { - this.fileProcessor.processFile(current, this.queueSize, filePath); - } - }); } + + // Process subdirectories (yields naturally to event loop) + for (const subdir of subdirs) { + yield* this.walkDirectory(subdir); + } + } catch (error: any) { + // Handle permission errors gracefully + if (error.code === 'EACCES' || error.code === 'EPERM') { + console.warn(`[FolderWatcher] Permission denied: ${directory} - skipping`); + } else if (error.code === 'ENOENT') { + console.warn(`[FolderWatcher] Directory not found: ${directory} - skipping`); + } else if (error.code === 'ENOTDIR') { + console.warn(`[FolderWatcher] Not a directory: ${directory} - skipping`); + } else { + console.error(`[FolderWatcher] Error reading directory ${directory}:`, error.message); + } + // Continue processing - don't let one bad directory stop the scan } - this.processing.delete(filePath); } /** - * Process the file and its sibling file included other folder it is a recursive function - * @param filePath - * @private + * Watch directories for file changes using chokidar + * + * @param directories - Directories to watch + * @param callbacks - Event callbacks + * @param options - Chokidar options */ - private async processFileExtended(filePath: string): Promise { - let dirname = path.dirname(filePath); - await this.processDirWithCount([dirname]); - } - - private async processDirWithCount(folderList: string[]): Promise { - let countPromises = []; - for (let i = 0; i < folderList.length; i++) { - folderList[i] = path.normalize(folderList[i]); - countPromises.push(this.countFile(folderList[i])); - } - const count = (await Promise.all(countPromises)).reduce((acc, current) => acc + current, 0); - console.log(`Found ${count} files to process`); - this.queueSize += count; - let promises: Promise[] = []; - for (let i = 0; i < folderList.length; i++) { - folderList[i] = path.normalize(folderList[i]); - await this.processDirectory(folderList[i], promises); - } - await Promise.all(promises); - } - - private chokidarWatch(folderList: string[]) { - const chokidarconfig = { - ignoreInitial: true,//ignore the initial scan we have our own function + watch(directories: string[], callbacks: { + onAdd?: (filePath: string) => void | Promise; + onChange?: (filePath: string) => void | Promise; + onUnlink?: (filePath: string) => void | Promise; + onReady?: () => void | Promise; + onError?: (error: Error) => void | Promise; + }, options?: { + interval?: number; + stabilityThreshold?: number; + pollInterval?: number; + }): chokidar.FSWatcher { + const chokidarConfig: chokidar.WatchOptions = { + ignoreInitial: true, // Don't emit events for initial scan persistent: true, depth: Infinity, awaitWriteFinish: { - stabilityThreshold: this.config.stabilityThreshold || 30000, - pollInterval: this.config.pollInterval || 5000 + stabilityThreshold: options?.stabilityThreshold || 30000, + pollInterval: options?.pollInterval || 5000 } }; let watcher: chokidar.FSWatcher; - if ((this.config.interval || 0) <= 0) { - watcher = chokidar.watch(folderList, chokidarconfig); + if ((options?.interval || 0) <= 0) { + watcher = chokidar.watch(directories, chokidarConfig); } else { - watcher = chokidar.watch(folderList, { - ...chokidarconfig, + watcher = chokidar.watch(directories, { + ...chokidarConfig, usePolling: true, - interval: this.config.interval || 0 + interval: options.interval }); } - watcher - .on('add', async (filePath) => { + if (callbacks.onAdd) { + watcher.on('add', async (filePath) => { try { - await this.processFileExtended(filePath); - if (this.finalizeDebounced) { - await this.finalizeDebounced(); - } + await callbacks.onAdd!(filePath); } catch (error) { - console.error(`Error processing file ${filePath}:`, error); + console.error(`[FolderWatcher] Error in onAdd callback for ${filePath}:`, error); } - }) - .on('change', async (filePath) => { - try { - await this.processFileExtended(filePath); - if (this.finalizeDebounced) { - await this.finalizeDebounced(); - } - } catch (e) { - console.error(`Error processing file ${filePath}:`, e); - } - }) - .on('unlink', async (filePath) => { - try { - if (this.fileProcessor.deleteFile) { - await this.fileProcessor.deleteFile(filePath); - } - this.processing.delete(filePath); - await this.processFileExtended(filePath); - if (this.finalizeDebounced) { - await this.finalizeDebounced(); - } - } catch (e) { - console.error(`Error processing file deletion:`, e); - } - }) - .on('error', (error) => console.error(`Watcher error: ${error}`)) - .on('ready', () => { - console.log(`Watching for file changes on ${folderList}`); }); - } - - async watch() { - let folders = this.WATCH_FOLDER_LIST; - if (!folders) { - throw new Error(`No folder to watch`); - } - let folderList = folders.split(','); - - if (this.fileProcessor.initialize) { - await this.fileProcessor.initialize(); } - this.chokidarWatch(folderList); - - //start the initial scan - await this.processDirWithCount(folderList); - this.initialized = true; - if (this.finalizeDebounced) { - await this.finalizeDebounced(); + if (callbacks.onChange) { + watcher.on('change', async (filePath) => { + try { + await callbacks.onChange!(filePath); + } catch (error) { + console.error(`[FolderWatcher] Error in onChange callback for ${filePath}:`, error); + } + }); } - console.log(`Watcher ready`); + if (callbacks.onUnlink) { + watcher.on('unlink', async (filePath) => { + try { + await callbacks.onUnlink!(filePath); + } catch (error) { + console.error(`[FolderWatcher] Error in onUnlink callback for ${filePath}:`, error); + } + }); + } + + watcher.on('error', (error: Error) => { + if (callbacks.onError) { + callbacks.onError(error); + } else { + // Default error handling + const err = error as any; + if (err.code === 'EACCES' || err.code === 'EPERM') { + console.warn(`[FolderWatcher] Permission denied for file watcher:`, error.message); + } else if (err.code === 'ENOENT') { + console.warn(`[FolderWatcher] Watched path no longer exists:`, error.message); + } else if (err.code === 'ENOSPC') { + console.error(`[FolderWatcher] CRITICAL: No space left on device or inotify watch limit reached:`, error.message); + console.error(`[FolderWatcher] Try increasing fs.inotify.max_user_watches: sysctl fs.inotify.max_user_watches=524288`); + } else { + console.error(`[FolderWatcher] Watcher error:`, error.message || error); + } + } + }); + + if (callbacks.onReady) { + watcher.on('ready', async () => { + try { + console.log(`[FolderWatcher] Watching for file changes on ${directories.join(', ')}`); + await callbacks.onReady!(); + } catch (error) { + console.error(`[FolderWatcher] Error in onReady callback:`, error); + } + }); + } + + return watcher; } } diff --git a/src/lib/folder-watcher/PollingWatcher.ts b/src/lib/folder-watcher/PollingWatcher.ts new file mode 100644 index 0000000..235b9b8 --- /dev/null +++ b/src/lib/folder-watcher/PollingWatcher.ts @@ -0,0 +1,168 @@ +import {readdir} from "fs/promises"; +import * as path from 'path'; +import {FileProcessorInterface} from "./FileProcessorInterface.js"; + +export interface PollingPathConfig { + path: string; + intervalMs: number; +} + +/** + * PollingWatcher - Manual polling-based file watcher (no chokidar) + * + * Polls specific directories at configured intervals using simple setInterval + readdir. + * Perfect for network drives (SMB/NFS) where filesystem events don't work reliably. + * + * Usage: + * - Configure with array of {path, intervalMs} objects + * - Each path can have its own polling interval + * - Runs in parallel with FolderWatcher (chokidar) + */ +export class PollingWatcher { + private fileProcessor: FileProcessorInterface; + private pollingConfigs: PollingPathConfig[]; + private intervals: NodeJS.Timeout[] = []; + private seenFiles: Set = new Set(); + private initialized = false; + + constructor(fileProcessor: FileProcessorInterface, pollingConfigs: PollingPathConfig[]) { + this.fileProcessor = fileProcessor; + this.pollingConfigs = pollingConfigs; + } + + /** + * Recursively scan directory and find all files + */ + private async scanDirectory(directory: string, allFiles: string[]): Promise { + try { + const entries = await readdir(directory, {withFileTypes: true}); + for (const entry of entries) { + const fullPath = path.join(directory, entry.name); + if (entry.isDirectory()) { + await this.scanDirectory(fullPath, allFiles); + } else { + allFiles.push(fullPath); + } + } + } catch (error: any) { + // Ignore permission errors and other scan errors + if (error.code !== 'EACCES' && error.code !== 'EPERM') { + console.error(`Error scanning directory ${directory}:`, error.message); + } + } + } + + /** + * Process a single file (check if new, then queue for processing) + */ + private async processFile(filePath: string): Promise { + // Skip if already seen + if (this.seenFiles.has(filePath)) { + return; + } + + // Check if file type is supported + if (!await this.fileProcessor.canProcessFile(filePath)) { + return; + } + + // Mark as seen + this.seenFiles.add(filePath); + console.log(`[PollingWatcher] New file detected: ${filePath}`); + + // Queue for processing (same pipeline as chokidar) + if (this.fileProcessor.markPending) { + this.fileProcessor.markPending(filePath); + } + + if (this.fileProcessor.queueFile) { + await this.fileProcessor.queueFile(filePath); + } + + if (this.fileProcessor.processFile) { + this.fileProcessor.processFile(0, 0, filePath); + } + } + + /** + * Poll a specific directory once + */ + private async pollDirectory(dirPath: string): Promise { + console.log(`[PollingWatcher] Polling ${dirPath}...`); + const startTime = Date.now(); + + try { + const allFiles: string[] = []; + await this.scanDirectory(dirPath, allFiles); + + console.log(`[PollingWatcher] Found ${allFiles.length} files in ${dirPath} (${Date.now() - startTime}ms)`); + + // Process new files + const promises: Promise[] = []; + for (const filePath of allFiles) { + promises.push(this.processFile(filePath)); + } + await Promise.all(promises); + + console.log(`[PollingWatcher] Poll completed for ${dirPath} (${Date.now() - startTime}ms)`); + } catch (error: any) { + console.error(`[PollingWatcher] Error polling ${dirPath}:`, error.message); + } + } + + /** + * Start polling all configured paths + */ + async start(): Promise { + if (this.pollingConfigs.length === 0) { + console.log('[PollingWatcher] No polling paths configured, skipping'); + return; + } + + console.log(`[PollingWatcher] Starting polling for ${this.pollingConfigs.length} paths`); + + // Initialize file processor if needed + if (!this.initialized && this.fileProcessor.initialize) { + await this.fileProcessor.initialize(); + this.initialized = true; + } + + // Do initial scan for all paths + for (const config of this.pollingConfigs) { + console.log(`[PollingWatcher] Initial scan: ${config.path} (interval: ${config.intervalMs}ms)`); + await this.pollDirectory(config.path); + } + + // Start interval polling for each path + for (const config of this.pollingConfigs) { + const interval = setInterval(async () => { + await this.pollDirectory(config.path); + }, config.intervalMs); + + this.intervals.push(interval); + console.log(`[PollingWatcher] Polling started for ${config.path} every ${config.intervalMs}ms`); + } + + console.log('[PollingWatcher] All polling intervals started'); + } + + /** + * Stop all polling + */ + stop(): void { + console.log('[PollingWatcher] Stopping all polling intervals'); + for (const interval of this.intervals) { + clearInterval(interval); + } + this.intervals = []; + console.log('[PollingWatcher] All polling intervals stopped'); + } + + /** + * Clear the seen files cache (useful for testing/debugging) + */ + clearCache(): void { + console.log(`[PollingWatcher] Clearing seen files cache (${this.seenFiles.size} files)`); + this.seenFiles.clear(); + } +} diff --git a/src/lib/index.ts b/src/lib/index.ts index 6e6880d..48f8d6a 100644 --- a/src/lib/index.ts +++ b/src/lib/index.ts @@ -11,6 +11,7 @@ export * from './hash-compute/HashComputerFile.js'; //folder-watcher export * from './folder-watcher/FolderWatcher.js'; +export * from './folder-watcher/PollingWatcher.js'; export * from './index-interface.js'; export * from './utils/ExistsAsync.js'; \ No newline at end of file