add folder watcher

dist/index.d.cts

@@ -1,5 +1,8 @@
 import { C as ComputeInterface, a as CID_ALGORITHM_NAMES, M as MultiHashData } from './ShaComputeWorker-DWOF-qaH.cjs';
 export { c as CID_ALGORITHM, b as CID_ALGORITHM_CODES, S as SimpleHash } from './ShaComputeWorker-DWOF-qaH.cjs';
+import * as PQueue from 'p-queue';
+import PQueue__default from 'p-queue';
+import * as p_queue_dist_priority_queue_js from 'p-queue/dist/priority-queue.js';
 import 'crypto';
 
 declare class HashComputer implements ComputeInterface {
@@ -72,4 +75,43 @@ interface Properties {
     INDEX_FILE_PATH: string;
 }
 
-export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync };
+interface FileProcessorInterface {
+    processFile(current: number, queueSize: number, nfoFilePath: string): Promise<void>;
+    canProcessFile(filePath: string): Promise<boolean>;
+    deleteFile(filePath: string): Promise<void>;
+}
+
+declare class FolderWatcher {
+    private WATCH_FOLDER_LIST;
+    private config;
+    initialized: boolean;
+    queue: PQueue__default<p_queue_dist_priority_queue_js.default, PQueue.QueueAddOptions>;
+    fileProcessor: FileProcessorInterface;
+    queueSize: number;
+    current: number;
+    processing: Set<string>;
+    constructor(fileProcessor: FileProcessorInterface, WATCH_FOLDER_LIST: string, config: {
+        interval?: number;
+        stabilityThreshold?: number;
+        pollInterval?: number;
+    });
+    /**
+     * Process the directory and its subdirectories
+     * @param directory
+     * @param promises
+     * @private
+     */
+    private processDirectory;
+    private countFile;
+    private processFile;
+    /**
+     * Process the file and its sibling file included other folder it is a recursive function
+     * @param filePath
+     * @private
+     */
+    private processFileExtended;
+    private chokidarWatch;
+    watch(): Promise<void>;
+}
+
+export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, type FileProcessorInterface, FolderWatcher, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync };

dist/index.d.ts

@@ -1,5 +1,8 @@
 import { C as ComputeInterface, a as CID_ALGORITHM_NAMES, M as MultiHashData } from './ShaComputeWorker-DWOF-qaH.js';
 export { c as CID_ALGORITHM, b as CID_ALGORITHM_CODES, S as SimpleHash } from './ShaComputeWorker-DWOF-qaH.js';
+import * as PQueue from 'p-queue';
+import PQueue__default from 'p-queue';
+import * as p_queue_dist_priority_queue_js from 'p-queue/dist/priority-queue.js';
 import 'crypto';
 
 declare class HashComputer implements ComputeInterface {
@@ -72,4 +75,43 @@ interface Properties {
     INDEX_FILE_PATH: string;
 }
 
-export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync };
+interface FileProcessorInterface {
+    processFile(current: number, queueSize: number, nfoFilePath: string): Promise<void>;
+    canProcessFile(filePath: string): Promise<boolean>;
+    deleteFile(filePath: string): Promise<void>;
+}
+
+declare class FolderWatcher {
+    private WATCH_FOLDER_LIST;
+    private config;
+    initialized: boolean;
+    queue: PQueue__default<p_queue_dist_priority_queue_js.default, PQueue.QueueAddOptions>;
+    fileProcessor: FileProcessorInterface;
+    queueSize: number;
+    current: number;
+    processing: Set<string>;
+    constructor(fileProcessor: FileProcessorInterface, WATCH_FOLDER_LIST: string, config: {
+        interval?: number;
+        stabilityThreshold?: number;
+        pollInterval?: number;
+    });
+    /**
+     * Process the directory and its subdirectories
+     * @param directory
+     * @param promises
+     * @private
+     */
+    private processDirectory;
+    private countFile;
+    private processFile;
+    /**
+     * Process the file and its sibling file included other folder it is a recursive function
+     * @param filePath
+     * @private
+     */
+    private processFileExtended;
+    private chokidarWatch;
+    watch(): Promise<void>;
+}
+
+export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, type FileProcessorInterface, FolderWatcher, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync };

package.json

@@ -40,9 +40,11 @@
   },
   "packageManager": "[email protected]",
   "dependencies": {
+    "chokidar": "^3.6.0",
     "csv-parse": "^5.5.5",
     "csv-stringify": "^6.4.6",
     "multiformats": "^13.1.0",
+    "p-queue": "^8.0.1",
     "piscina": "^4.4.0"
   }
 }

src/lib/folder_watcher/FileProcessorInterface.ts

@@ -0,0 +1,7 @@
+export interface FileProcessorInterface {
+    processFile(current: number, queueSize: number, nfoFilePath: string): Promise<void>;
+
+    canProcessFile(filePath: string): Promise<boolean>;
+
+    deleteFile(filePath: string): Promise<void>
+}

src/lib/folder_watcher/FolderWatcher.ts

@@ -0,0 +1,164 @@
+import * as path from 'path';
+import chokidar from 'chokidar';
+import {readdir} from "fs/promises";
+import PQueue from "p-queue";
+import {FileProcessorInterface} from "./FileProcessorInterface.js";
+
+export class FolderWatcher {
+    initialized = false;
+    queue = new PQueue({concurrency: 100, autoStart: true});
+    fileProcessor: FileProcessorInterface;
+    queueSize = -1;
+    current = 0;
+    processing = new Set<string>();
+
+    constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{
+        interval?: number,
+        stabilityThreshold?: number,
+        pollInterval?: number,
+    }) {
+        this.fileProcessor = fileProcessor;
+    }
+
+    /**
+     * Process the directory and its subdirectories
+     * @param directory
+     * @param promises
+     * @private
+     */
+    private async processDirectory(directory: string, promises: Promise<any>[]) {
+        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 countFile(directory: string): Promise<number> {
+        const entries = await readdir(directory, {withFileTypes: true});
+        const countPromises: Promise<any>[] = [];
+
+        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 {
+                // Use an IIFE (Immediately Invoked Function Expression) to handle asynchronous condition
+                countPromises.push(this.queue.add(
+                    async () => (await this.fileProcessor.canProcessFile(fullPath)) ? 1 : 0
+                ));
+            }
+        }
+
+        // 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<void> {
+        if(await this.fileProcessor.canProcessFile(filePath)) {
+            if(!this.processing.has(filePath)) {
+                this.processing.add(filePath);//avoid double processing in the queue
+                const current = ++this.current;
+                await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath));
+                this.processing.delete(filePath);
+            }
+        }
+    }
+
+    /**
+     * Process the file and its sibling file included other folder it is a recursive function
+     * @param filePath
+     * @private
+     */
+    private async processFileExtended(filePath: string):Promise<void> {
+        let dirname = path.dirname(filePath);
+        await this.processDirectory(dirname, []);
+    }
+
+    private chokidarWatch(folderList: string[]){
+        const chokidarconfig = {
+            ignoreInitial: true,//ignore the initial scan we have our own function
+            persistent: true,
+            depth: Infinity,
+            awaitWriteFinish: {
+                stabilityThreshold: this.config.stabilityThreshold || 30000,
+                pollInterval: this.config.pollInterval || 5000
+            }
+        };
+
+        let watcher: chokidar.FSWatcher;
+        if ((this.config.interval || 0) <= 0) {
+            watcher = chokidar.watch(folderList, chokidarconfig);
+        } else {
+            watcher = chokidar.watch(folderList, {
+                ...chokidarconfig,
+                usePolling: true,
+                interval: this.config.interval || 0
+            });
+        }
+
+        watcher
+            .on('add', async (filePath) => {
+                try {
+                    await this.processFileExtended(filePath);
+                } catch (error) {
+                    console.error(`Error processing file ${filePath}:`, error);
+                }
+            })
+            .on('change', async (filePath) => {
+                try {
+                    await this.processFileExtended(filePath);
+                } catch (e) {
+                    console.error(`Error processing file ${filePath}:`, e);
+                }
+            })
+            .on('unlink', async (filePath) => {
+                try {
+                    await this.fileProcessor.deleteFile(filePath);
+                    await this.processFileExtended(filePath);
+                } 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(',');
+
+        this.chokidarWatch(folderList);
+
+        //start the initial scan
+        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<any>[] = [];
+        for (let i = 0; i < folderList.length; i++) {
+            folderList[i] = path.normalize(folderList[i]);
+            await this.processDirectory(folderList[i], promises);
+        }
+        await Promise.all(promises);
+        this.initialized = true;
+
+        console.log(`Watcher ready`);
+    }
+}

src/lib/index.ts

@@ -9,3 +9,5 @@ export * from './HashIndexManager.js';
 export * from './MultiHashData.js';
 // Assuming Properties.ts exports one or more interfaces; adjust as needed
 export * from './Properties.js';
+export * from './folder_watcher/FolderWatcher.js';
+export * from './folder_watcher/FileProcessorInterface.js';

yarn.lock

@@ -416,9 +416,11 @@ __metadata:
   resolution: "@metazla/meta-hash@workspace:."
   dependencies:
     "@types/node": "npm:^20.12.2"
+    chokidar: "npm:^3.6.0"
     csv-parse: "npm:^5.5.5"
     csv-stringify: "npm:^6.4.6"
     multiformats: "npm:^13.1.0"
+    p-queue: "npm:^8.0.1"
     piscina: "npm:^4.4.0"
     ts-node: "npm:^10.9.2"
     tsup: "npm:^8.0.2"
@@ -904,7 +906,7 @@ __metadata:
   languageName: node
   linkType: hard
 
-"chokidar@npm:^3.5.1":
+"chokidar@npm:^3.5.1, chokidar@npm:^3.6.0":
   version: 3.6.0
   resolution: "chokidar@npm:3.6.0"
   dependencies:
@@ -1249,6 +1251,13 @@ __metadata:
   languageName: node
   linkType: hard
 
+"eventemitter3@npm:^5.0.1":
+  version: 5.0.1
+  resolution: "eventemitter3@npm:5.0.1"
+  checksum: 10c0/4ba5c00c506e6c786b4d6262cfbce90ddc14c10d4667e5c83ae993c9de88aa856033994dd2b35b83e8dc1170e224e66a319fa80adc4c32adcd2379bbc75da814
+  languageName: node
+  linkType: hard
+
 "execa@npm:^5.0.0":
   version: 5.1.1
   resolution: "execa@npm:5.1.1"
@@ -2051,6 +2060,23 @@ __metadata:
   languageName: node
   linkType: hard
 
+"p-queue@npm:^8.0.1":
+  version: 8.0.1
+  resolution: "p-queue@npm:8.0.1"
+  dependencies:
+    eventemitter3: "npm:^5.0.1"
+    p-timeout: "npm:^6.1.2"
+  checksum: 10c0/fe185bc8bbd32d17a5f6dba090077b1bb326b008b4ec9b0646c57a32a6984035aa8ece909a6d0de7f6c4640296dc288197f430e7394cdc76a26d862339494616
+  languageName: node
+  linkType: hard
+
+"p-timeout@npm:^6.1.2":
+  version: 6.1.2
+  resolution: "p-timeout@npm:6.1.2"
+  checksum: 10c0/d46b90a9a5fb7c650a5c56dd5cf7102ea9ab6ce998defa2b3d4672789aaec4e2f45b3b0b5a4a3e17a0fb94301ad5dd26da7d8728402e48db2022ad1847594d19
+  languageName: node
+  linkType: hard
+
 "path-key@npm:^3.0.0, path-key@npm:^3.1.0":
   version: 3.1.1
   resolution: "path-key@npm:3.1.1"