initial commit
This commit is contained in:
Pierre
@@ -0,0 +1,50 @@
|
||||
/**
|
||||
* Interface for file processing implementations used by FolderWatcher.
|
||||
*
|
||||
* Provides hooks for different stages of the file processing pipeline:
|
||||
* 1. markPending() - File discovered, waiting in queue
|
||||
* 2. queueFile() - File starts processing (optional pre-processing phase)
|
||||
* 3. processFile() - Main processing phase
|
||||
* 4. finalize() - After all files processed (optional)
|
||||
*/
|
||||
export interface FileProcessorInterface {
|
||||
/** Optional initialization before processing starts */
|
||||
initialize?(): Promise<void>;
|
||||
|
||||
/**
|
||||
* Main file processing method.
|
||||
* @param current - Current file index in the queue
|
||||
* @param queueSize - Total number of files to process
|
||||
* @param nfoFilePath - Path to the file being processed
|
||||
*/
|
||||
processFile(current: number, queueSize: number, nfoFilePath: string): Promise<void>;
|
||||
|
||||
/**
|
||||
* Determines if a file should be processed.
|
||||
* @param filePath - Path to the file to check
|
||||
* @returns true if the file should be processed
|
||||
*/
|
||||
canProcessFile(filePath: string): Promise<boolean>;
|
||||
|
||||
/** Optional callback when a file is deleted */
|
||||
deleteFile?(filePath: string): Promise<void>;
|
||||
|
||||
/** Optional callback after all files have been processed */
|
||||
finalize?(): Promise<void>;
|
||||
|
||||
/**
|
||||
* Optional callback when a file is discovered and marked as pending.
|
||||
* Called BEFORE the file enters the processing queue.
|
||||
* Useful for tracking/monitoring pending files.
|
||||
* @param filePath - Path to the file being marked as pending
|
||||
*/
|
||||
markPending?(filePath: string): void;
|
||||
|
||||
/**
|
||||
* Optional callback when a file begins processing.
|
||||
* Called WITHIN the queue (respects concurrency limits).
|
||||
* Can be used for quick/light pre-processing before main processing.
|
||||
* @param filePath - Path to the file being queued
|
||||
*/
|
||||
queueFile?(filePath: string): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
import * as path from 'path';
|
||||
import chokidar from 'chokidar';
|
||||
import {readdir} from "fs/promises";
|
||||
|
||||
/**
|
||||
* 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 {
|
||||
/**
|
||||
* 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<string> {
|
||||
for (const dir of directories) {
|
||||
const normalizedDir = path.normalize(dir);
|
||||
yield* this.walkDirectory(normalizedDir);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Recursively walk directory tree
|
||||
* Yields file paths as they're discovered
|
||||
*
|
||||
* @param directory - Directory to walk
|
||||
* @returns AsyncGenerator yielding file paths
|
||||
*/
|
||||
private async *walkDirectory(directory: string): AsyncGenerator<string> {
|
||||
try {
|
||||
const entries = await readdir(directory, {withFileTypes: true});
|
||||
|
||||
// Collect subdirectories for parallel processing
|
||||
const subdirs: string[] = [];
|
||||
|
||||
for (const entry of entries) {
|
||||
const fullPath = path.join(directory, entry.name);
|
||||
|
||||
if (entry.isDirectory()) {
|
||||
subdirs.push(fullPath);
|
||||
} else {
|
||||
// Yield file path immediately
|
||||
yield fullPath;
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Watch directories for file changes using chokidar
|
||||
*
|
||||
* @param directories - Directories to watch
|
||||
* @param callbacks - Event callbacks
|
||||
* @param options - Chokidar options
|
||||
*/
|
||||
watch(directories: string[], callbacks: {
|
||||
onAdd?: (filePath: string) => void | Promise<void>;
|
||||
onChange?: (filePath: string) => void | Promise<void>;
|
||||
onUnlink?: (filePath: string) => void | Promise<void>;
|
||||
onReady?: () => void | Promise<void>;
|
||||
onError?: (error: Error) => void | Promise<void>;
|
||||
}, 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: options?.stabilityThreshold || 30000,
|
||||
pollInterval: options?.pollInterval || 5000
|
||||
}
|
||||
};
|
||||
|
||||
let watcher: chokidar.FSWatcher;
|
||||
if ((options?.interval || 0) <= 0) {
|
||||
watcher = chokidar.watch(directories, chokidarConfig);
|
||||
} else {
|
||||
watcher = chokidar.watch(directories, {
|
||||
...chokidarConfig,
|
||||
usePolling: true,
|
||||
interval: options.interval
|
||||
});
|
||||
}
|
||||
|
||||
if (callbacks.onAdd) {
|
||||
watcher.on('add', async (filePath) => {
|
||||
try {
|
||||
await callbacks.onAdd!(filePath);
|
||||
} catch (error) {
|
||||
console.error(`[FolderWatcher] Error in onAdd callback for ${filePath}:`, error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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<string> = 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<void> {
|
||||
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<void> {
|
||||
// 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<void> {
|
||||
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<void>[] = [];
|
||||
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<void> {
|
||||
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();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user