diff --git a/dist/index.cjs b/dist/index.cjs index 992bfc9..8b5f764 100644 --- a/dist/index.cjs +++ b/dist/index.cjs @@ -7,14 +7,36 @@ var csvParse = require('csv-parse'); var sync = require('csv-parse/sync'); var sync$1 = require('csv-stringify/sync'); var promises = require('fs/promises'); -var I = require('path'); +var n = require('path'); var timers = require('timers'); +var w = require('chokidar'); +var b = require('p-queue'); function _interopDefault (e) { return e && e.__esModule ? e : { default: e }; } -var I__default = /*#__PURE__*/_interopDefault(I); +function _interopNamespace(e) { + if (e && e.__esModule) return e; + var n = Object.create(null); + if (e) { + Object.keys(e).forEach(function (k) { + if (k !== 'default') { + var d = Object.getOwnPropertyDescriptor(e, k); + Object.defineProperty(n, k, d.get ? d : { + enumerable: true, + get: function () { return e[k]; } + }); + } + }); + } + n.default = e; + return Object.freeze(n); +} -var h=undefined;h=h.replace("src","dist");h=h+"/worker.js";var f=new URL(h).href;console.log("Worker URL: ",f);var c=class{piscina;constructor(e){this.piscina=new piscina.Piscina({maxThreads:4,filename:e||process.env.WORKER_URL||f});}async computeCIDs(e,t){return this.piscina.run({filePath:e,algorithms:t})}};var m=class{constructor(e,t){this.targetHash=e;this.fileIDComputer=new c(t);}fileIDComputer;async computeMissingHash(e,t){let i=this.targetHash.filter(r=>!t[r]);if(i.length===0)return;(await this.fileIDComputer.computeCIDs(e,i)).forEach((r,a)=>{let n=i[a];t[n]=r;});}};async function l(o){try{return await promises.access(o,promises.constants.F_OK),!0}catch{return !1}}var q=["path","size","mtime"],p=class{constructor(e,t=["cid_sha1","cid_sha2-256"]){this.filePath=e;this.targetHash=t;if(!e)throw new Error("Invalid index file path")}cache=new Map;intervalId;intervalTime=3e4;lastIndexFileSize=0;lastCacheFile;indexOpsInProgress=!1;hasChanged=!1;initialLoad;getCache(){return new Map(this.cache)}async init(e=!0){return this.initialLoad||(this.initialLoad=new Promise(async(t,i)=>{try{if(!this.checkCSVHeaders(this.filePath))throw new Error("Invalid index file headers");await this.loadIndex(),e&&this.start(),t();}catch(s){i(s);}})),this.initialLoad}checkCSVHeaders(e){let t=sync.parse(e,{bom:!0,columns:!0,skip_empty_lines:!0});if(t.length===0)return !0;let i=Object.keys(t[0]);return ["path","size","mtime",...this.targetHash].every(r=>i.includes(r))}start(){this.startAutoSave(this.intervalTime);}stopAutoSave(){this.intervalId&&timers.clearInterval(this.intervalId);}startAutoSave(e){this.stopAutoSave(),this.intervalTime=e,this.intervalId=setInterval(()=>this.saveCacheToFile(),e);}async loadIndex(){if(await l(this.filePath)){let e=await fs.promises.stat(this.filePath);if(this.lastIndexFileSize!==e.size){let t=await this.readCsv();for(let i of t)this.cache.set(i.path,i);return this.lastIndexFileSize=e.size,this.lastCacheFile=t,t}else return this.lastCacheFile}return []}loadIndexFromCache(){return Array.from(this.cache.values())}async readCsv(){if(!await l(this.filePath))return [];let e=performance.now(),t=csvParse.parse({columns:!0,skip_empty_lines:!0}),i=[];return new Promise((s,r)=>{fs.createReadStream(this.filePath).pipe(t).on("data",a=>{i.push(a);}).on("end",()=>{s(i),console.log(`Index read time ${performance.now()-e}ms`);}).on("error",a=>{r(a);});})}async saveCacheToFile(){if(this.indexOpsInProgress||!this.hasChanged)return;this.hasChanged=!1,this.indexOpsInProgress=!0;let e=performance.now(),t=await this.loadIndex();if(this.cache.size!==0){let r=this.loadIndexFromCache().filter(a=>!t.find(n=>n.path===a.path));if(r.length!==0){let a=sync$1.stringify(r,{header:t.length===0,columns:[{key:"path",header:"path"},{key:"size",header:"size"},{key:"mtime",header:"mtime"},...this.targetHash.map(n=>({key:n,header:n}))]});await fs.promises.appendFile(this.filePath,a);}}let i=performance.now()-e;console.log(`Index saved in ${i}ms`),i*10>this.intervalTime&&(this.startAutoSave(i*10),console.log(`Index save interval increased to ${i*10}ms`)),this.indexOpsInProgress=!1;}getCidForFile(e,t,i){let s=I__default.default.basename(e),r=this.cache.get(s);if(r){if(r.mtime){if(r.size===t+""&&r.mtime===i)return r}else if(r.size===t+"")return r}return null}addFileCid(e,t,i,s){if(!e||!t||!i||!s)throw new Error("Invalid parameters");let r=t+"",a=I__default.default.basename(e),n={path:a,size:r,mtime:i,...s};this.cache.set(a,n),this.hasChanged=!0;}};var g=class{constructor(e,t=["cid_sha1","cid_sha2-256"],i){this.targetHash=t;this.hashComputer=new m(this.targetHash,i),this.hashIndexManager=new p(e);}hashIndexManager;hashComputer;async computeMissingHash(e,t){await this.hashIndexManager.init();let i=await promises.stat(e);if(this.hashIndexManager.getCache().has(I__default.default.basename(e))){let s=this.hashIndexManager.getCidForFile(e,i.size,i.mtime.toISOString());if(s)for(let r of this.targetHash)!t[r]&&s[r]&&(t[r]=s[r]);}await this.hashComputer.computeMissingHash(e,t),this.hashIndexManager.addFileCid(e,i.size,i.mtime.toISOString(),t);}}; +var n__namespace = /*#__PURE__*/_interopNamespace(n); +var w__default = /*#__PURE__*/_interopDefault(w); +var b__default = /*#__PURE__*/_interopDefault(b); + +var h=undefined;h=h.replace("src","dist");h=h+"/worker.js";var f=new URL(h).href;console.log("Worker URL: ",f);var l=class{piscina;constructor(e){this.piscina=new piscina.Piscina({maxThreads:4,filename:e||process.env.WORKER_URL||f});}async computeCIDs(e,t){return this.piscina.run({filePath:e,algorithms:t})}};var p=class{constructor(e,t){this.targetHash=e;this.fileIDComputer=new l(t);}fileIDComputer;async computeMissingHash(e,t){let r=this.targetHash.filter(i=>!t[i]);if(r.length===0)return;(await this.fileIDComputer.computeCIDs(e,r)).forEach((i,a)=>{let o=r[a];t[o]=i;});}};async function d(c){try{return await promises.access(c,promises.constants.F_OK),!0}catch{return !1}}var Q=["path","size","mtime"],m=class{constructor(e,t=["cid_sha1","cid_sha2-256"]){this.filePath=e;this.targetHash=t;if(!e)throw new Error("Invalid index file path")}cache=new Map;intervalId;intervalTime=3e4;lastIndexFileSize=0;lastCacheFile;indexOpsInProgress=!1;hasChanged=!1;initialLoad;getCache(){return new Map(this.cache)}async init(e=!0){return this.initialLoad||(this.initialLoad=new Promise(async(t,r)=>{try{if(!this.checkCSVHeaders(this.filePath))throw new Error("Invalid index file headers");await this.loadIndex(),e&&this.start(),t();}catch(s){r(s);}})),this.initialLoad}checkCSVHeaders(e){let t=sync.parse(e,{bom:!0,columns:!0,skip_empty_lines:!0});if(t.length===0)return !0;let r=Object.keys(t[0]);return ["path","size","mtime",...this.targetHash].every(i=>r.includes(i))}start(){this.startAutoSave(this.intervalTime);}stopAutoSave(){this.intervalId&&timers.clearInterval(this.intervalId);}startAutoSave(e){this.stopAutoSave(),this.intervalTime=e,this.intervalId=setInterval(()=>this.saveCacheToFile(),e);}async loadIndex(){if(await d(this.filePath)){let e=await fs.promises.stat(this.filePath);if(this.lastIndexFileSize!==e.size){let t=await this.readCsv();for(let r of t)this.cache.set(r.path,r);return this.lastIndexFileSize=e.size,this.lastCacheFile=t,t}else return this.lastCacheFile}return []}loadIndexFromCache(){return Array.from(this.cache.values())}async readCsv(){if(!await d(this.filePath))return [];let e=performance.now(),t=csvParse.parse({columns:!0,skip_empty_lines:!0}),r=[];return new Promise((s,i)=>{fs.createReadStream(this.filePath).pipe(t).on("data",a=>{r.push(a);}).on("end",()=>{s(r),console.log(`Index read time ${performance.now()-e}ms`);}).on("error",a=>{i(a);});})}async saveCacheToFile(){if(this.indexOpsInProgress||!this.hasChanged)return;this.hasChanged=!1,this.indexOpsInProgress=!0;let e=performance.now(),t=await this.loadIndex();if(this.cache.size!==0){let i=this.loadIndexFromCache().filter(a=>!t.find(o=>o.path===a.path));if(i.length!==0){let a=sync$1.stringify(i,{header:t.length===0,columns:[{key:"path",header:"path"},{key:"size",header:"size"},{key:"mtime",header:"mtime"},...this.targetHash.map(o=>({key:o,header:o}))]});await fs.promises.appendFile(this.filePath,a);}}let r=performance.now()-e;console.log(`Index saved in ${r}ms`),r*10>this.intervalTime&&(this.startAutoSave(r*10),console.log(`Index save interval increased to ${r*10}ms`)),this.indexOpsInProgress=!1;}getCidForFile(e,t,r){let s=n__namespace.default.basename(e),i=this.cache.get(s);if(i){if(i.mtime){if(i.size===t+""&&i.mtime===r)return i}else if(i.size===t+"")return i}return null}addFileCid(e,t,r,s){if(!e||!t||!r||!s)throw new Error("Invalid parameters");let i=t+"",a=n__namespace.default.basename(e),o={path:a,size:i,mtime:r,...s};this.cache.set(a,o),this.hasChanged=!0;}};var v=class{constructor(e,t=["cid_sha1","cid_sha2-256"],r){this.targetHash=t;this.hashComputer=new p(this.targetHash,r),this.hashIndexManager=new m(e);}hashIndexManager;hashComputer;async computeMissingHash(e,t){await this.hashIndexManager.init();let r=await promises.stat(e);if(this.hashIndexManager.getCache().has(n__namespace.default.basename(e))){let s=this.hashIndexManager.getCidForFile(e,r.size,r.mtime.toISOString());if(s)for(let i of this.targetHash)!t[i]&&s[i]&&(t[i]=s[i]);}await this.hashComputer.computeMissingHash(e,t),this.hashIndexManager.addFileCid(e,r.size,r.mtime.toISOString(),t);}};var y=class{constructor(e,t,r){this.WATCH_FOLDER_LIST=t;this.config=r;this.fileProcessor=e;}initialized=!1;queue=new b__default.default({concurrency:100,autoStart:!0});fileProcessor;queueSize=-1;current=0;processing=new Set;async processDirectory(e,t){let r=await promises.readdir(e,{withFileTypes:!0});for(let s of r){let i=n__namespace.join(e,s.name);if(s.isDirectory())await this.processDirectory(i,t);else {let a=this.processFile(i);t.push(a);}}}async countFile(e){let t=await promises.readdir(e,{withFileTypes:!0}),r=[];for(let i of t){let a=n__namespace.join(e,i.name);i.isDirectory()?r.push(this.countFile(a)):r.push(this.queue.add(async()=>await this.fileProcessor.canProcessFile(a)?1:0));}return (await Promise.all(r)).reduce((i,a)=>i+a,0)}async processFile(e){if(await this.fileProcessor.canProcessFile(e)&&!this.processing.has(e)){this.processing.add(e);let t=++this.current;await this.queue.add(()=>this.fileProcessor.processFile(t,this.queueSize,e)),this.processing.delete(e);}}async processFileExtended(e){let t=n__namespace.dirname(e);await this.processDirectory(t,[]);}chokidarWatch(e){let t={ignoreInitial:!0,persistent:!0,depth:1/0,awaitWriteFinish:{stabilityThreshold:this.config.stabilityThreshold||3e4,pollInterval:this.config.pollInterval||5e3}},r;(this.config.interval||0)<=0?r=w__default.default.watch(e,t):r=w__default.default.watch(e,{...t,usePolling:!0,interval:this.config.interval||0}),r.on("add",async s=>{try{await this.processFileExtended(s);}catch(i){console.error(`Error processing file ${s}:`,i);}}).on("change",async s=>{try{await this.processFileExtended(s);}catch(i){console.error(`Error processing file ${s}:`,i);}}).on("unlink",async s=>{try{await this.fileProcessor.deleteFile(s),await this.processFileExtended(s);}catch(i){console.error("Error processing file deletion:",i);}}).on("error",s=>console.error(`Watcher error: ${s}`)).on("ready",()=>{console.log(`Watching for file changes on ${e}`);});}async watch(){let e=this.WATCH_FOLDER_LIST;if(!e)throw new Error("No folder to watch");let t=e.split(",");this.chokidarWatch(t);let r=[];for(let a=0;aa+o,0);console.log(`Found ${s} files to process`),this.queueSize=s;let i=[];for(let a=0;a {\n return this.piscina.run({filePath, algorithms});\n }\n}\n","import { FileIDComputer } from \"./fileID/FileIDComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\n\nexport class HashComputer implements ComputeInterface{\n private fileIDComputer:FileIDComputer;\n\n constructor(private targetHash: CID_ALGORITHM_NAMES[],workerPath?:string) {\n this.fileIDComputer = new FileIDComputer(workerPath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n // Dynamically determine which hashes are needed\n const neededHashes = this.targetHash.filter(hashName => !metadata[hashName]);\n\n // If all hashes are already computed, skip the processing\n if (neededHashes.length === 0) {\n return;\n }\n\n // Compute only the needed CIDs\n const cids = await this.fileIDComputer.computeCIDs(filePath, neededHashes);\n\n // Map the computed CIDs back to their respective metadata properties\n cids.forEach((cid, index) => {\n const hashType = neededHashes[index];\n metadata[hashType] = cid;\n });\n }\n}\n","import {createReadStream, promises as fs} from 'fs';\nimport {parse} from 'csv-parse';\nimport {parse as parseSync} from 'csv-parse/sync';\nimport {stringify} from 'csv-stringify/sync';\nimport {existsAsync} from \"./ExistsAsync\";\nimport path from \"path\";\nimport {clearInterval} from \"node:timers\";\nimport {CID_ALGORITHM_NAMES} from \"./MultiHashData\";\n\ninterface IndexLine extends Partial> {\n path: string;\n size: string;\n mtime: string;\n}\nexport const INDEX_HEADERS = ['path', 'size', 'mtime'];\n\nexport class HashIndexManager {\n private cache: Map = new Map();\n private intervalId: NodeJS.Timeout;\n private intervalTime: number = 30000;\n private lastIndexFileSize: number = 0; //size of the index file last time it was read\n private lastCacheFile: IndexLine[]; //state of the file last time it was read\n private indexOpsInProgress: boolean = false;\n private hasChanged: boolean = false;\n private initialLoad;\n\n constructor(private filePath: string,private targetHash:CID_ALGORITHM_NAMES[]= [CID_ALGORITHM_NAMES.sha1,CID_ALGORITHM_NAMES.sha256]) {\n if (!filePath) {\n throw new Error('Invalid index file path');\n }\n }\n\n public getCache(): Map {\n return new Map(this.cache);\n }\n\n /**\n * After init consseutively calls to this method will not reload the index\n * @param autosave\n */\n async init(autosave = true) {\n if (!this.initialLoad) {\n this.initialLoad = new Promise(async (resolve, reject) => {\n try {\n if (!this.checkCSVHeaders(this.filePath)) {\n throw new Error('Invalid index file headers');\n }\n await this.loadIndex();\n if (autosave) {\n this.start();\n }\n resolve();\n }catch (e) {\n reject(e);\n }\n });\n }\n return this.initialLoad;\n }\n\n // Function to check CSV headers\n checkCSVHeaders (csvContent: string): boolean {\n const records = parseSync(csvContent, {\n bom: true,\n columns: true,\n skip_empty_lines: true,\n });\n\n if (records.length === 0) {\n //we will write the headers\n return true;\n }\n\n // Extract headers\n const headers = Object.keys(records[0]);\n\n // Define required headers\n //from the IndexLine interface\n const requiredHeaders = ['path', 'size', 'mtime', ...this.targetHash];\n\n // Check if all required headers are present\n return requiredHeaders.every(header => headers.includes(header));\n };\n\n public start() {\n this.startAutoSave(this.intervalTime);\n }\n\n public stopAutoSave() {\n if (this.intervalId) {\n clearInterval(this.intervalId);\n }\n }\n\n private startAutoSave(time: number) {\n this.stopAutoSave();\n this.intervalTime = time;\n this.intervalId = setInterval(() => this.saveCacheToFile(), time);\n }\n\n public async loadIndex(): Promise {\n if (await existsAsync(this.filePath)) {\n // check the file size and if it did not change, do not read the file\n const stats = await fs.stat(this.filePath);\n if (this.lastIndexFileSize !== stats.size) {\n // Read existing file content and parse it\n const records: IndexLine[] = await this.readCsv();\n for (const record of records) {\n this.cache.set(record.path, record);\n }\n this.lastIndexFileSize = stats.size;\n this.lastCacheFile = records;\n return records;\n } else {\n return this.lastCacheFile;\n }\n }\n return [];\n }\n\n private loadIndexFromCache(): IndexLine[] {\n return Array.from(this.cache.values());\n }\n\n private async readCsv(): Promise {\n if (!(await existsAsync(this.filePath))) {\n return [];\n }\n const start = performance.now();\n\n const parser = parse({\n columns: true,\n skip_empty_lines: true,\n });\n\n const records: IndexLine[] = [];\n\n return new Promise((resolve, reject) => {\n createReadStream(this.filePath)\n .pipe(parser)\n .on('data', (record) => {\n records.push(record);\n }).on('end', () => {\n resolve(records);\n console.log(`Index read time ${performance.now() - start}ms`);\n }).on('error', (err) => {\n reject(err);\n });\n });\n }\n\n private async saveCacheToFile(): Promise {\n if (this.indexOpsInProgress || !this.hasChanged) {\n return;\n }\n this.hasChanged = false;\n this.indexOpsInProgress = true;\n const start = performance.now();\n let existingRows: IndexLine[] = await this.loadIndex();\n\n if (this.cache.size !== 0) {\n const cacheRows: IndexLine[] = this.loadIndexFromCache();\n // Filter out cacheRows that are already in the file\n const newRows = cacheRows.filter(row => !existingRows.find(existingRow => existingRow.path === row.path));\n\n if (newRows.length !== 0) {\n // Serialize new cacheRows to CSV string\n const csvString = stringify(newRows, {\n header: existingRows.length === 0, // Only add header if the file was empty\n columns: [\n {key: 'path', header: 'path'},\n {key: 'size', header: 'size'},\n {key: 'mtime', header: 'mtime'},\n ...this.targetHash.map(hash => ({key: hash, header: hash})),\n ],\n });\n\n // Append new cacheRows to the file\n await fs.appendFile(this.filePath, csvString);\n }\n }\n const totalTime = performance.now() - start;\n console.log(`Index saved in ${totalTime}ms`);\n // Check if the time to save the index is greater than the interval time. increase the interval time if needed\n if (totalTime * 10 > this.intervalTime) {\n this.startAutoSave(totalTime * 10);\n console.log(`Index save interval increased to ${totalTime * 10}ms`);\n }\n this.indexOpsInProgress = false;\n }\n\n public getCidForFile(filePath: string, fileSize: number, mtime: string): IndexLine {\n const fileName = path.basename(filePath);\n let fileNameIndex = this.cache.get(fileName);\n if(fileNameIndex) {\n if(fileNameIndex.mtime) {\n //if we have a mtime, we need to check it\n if (fileNameIndex.size === (fileSize + \"\") && fileNameIndex.mtime === mtime) {\n return fileNameIndex;\n }\n }else {\n //mtime is optional\n if (fileNameIndex.size === (fileSize + \"\")) {\n return fileNameIndex;\n }\n }\n }\n\n // 3 - if not found, delete the entry (keeps the index clean)\n /*if (fileNameIndex && fileNameIndex.size !== (fileSize + \"\") && pathIndex && pathIndex.size !== (fileSize + \"\")) {\n this.cache.delete(fileName);\n this.cache.delete(filePath);\n }*/\n\n return null;\n }\n\n public addFileCid(filePath: string, fileSize: number, mtime: string, hashs: Partial>): void {\n if (!filePath || !fileSize || !mtime || !hashs) {\n throw new Error('Invalid parameters');\n }\n const size = fileSize + \"\";\n const baseName = path.basename(filePath);\n const data = {path: baseName, size: size, mtime: mtime, ...hashs};\n this.cache.set(baseName, data);\n this.hasChanged = true;\n }\n\n}","// Example of checking if a file exists asynchronously\n// Note: `existsSync` does not have a direct async equivalent, so we use `access` instead.\nimport {access,constants} from \"fs/promises\";\n\nexport async function existsAsync(filePath: string): Promise {\n try {\n await access(filePath, constants.F_OK);\n return true; // File exists\n } catch {\n return false; // File does not exist\n }\n}","import {HashComputer} from \"./HashComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\nimport {HashIndexManager} from \"./HashIndexManager\";\nimport {stat} from \"fs/promises\";\nimport path from \"path\";\n\nexport class ComputeHashIndexCache implements ComputeInterface {\n hashIndexManager: HashIndexManager;\n hashComputer: HashComputer;\n\n constructor(indexFilePath: string,private targetHash: CID_ALGORITHM_NAMES[] = [CID_ALGORITHM_NAMES.sha1, CID_ALGORITHM_NAMES.sha256],workerPath?:string) {\n this.hashComputer = new HashComputer(this.targetHash,workerPath);\n this.hashIndexManager = new HashIndexManager(indexFilePath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n await this.hashIndexManager.init();\n let stats = await stat(filePath);\n if (this.hashIndexManager.getCache().has(path.basename(filePath))) {\n const indexLine = this.hashIndexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());\n if(indexLine) {\n for (const hash of this.targetHash) {\n if (!metadata[hash] && indexLine[hash]) {\n metadata[hash] = indexLine[hash];\n }\n }\n }\n }\n await this.hashComputer.computeMissingHash(filePath, metadata);\n this.hashIndexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), metadata);\n }\n}"]} \ No newline at end of file +{"version":3,"sources":["../src/lib/fileID/FileIDComputer.ts","../src/lib/HashComputer.ts","../src/lib/HashIndexManager.ts","../src/lib/ExistsAsync.ts","../src/lib/ComputeHashIndexCache.ts","../src/lib/folder_watcher/FolderWatcher.ts"],"names":["Piscina","distFolder","workerUrl","FileIDComputer","workerPath","filePath","algorithms","HashComputer","targetHash","metadata","neededHashes","hashName","cid","index","hashType","createReadStream","fs","parse","parseSync","stringify","access","constants","existsAsync","path","clearInterval","INDEX_HEADERS","HashIndexManager","autosave","resolve","reject","e","csvContent","records","headers","header","time","stats","record","start","parser","err","existingRows","newRows","row","existingRow","csvString","hash","totalTime","fileSize","mtime","fileName","fileNameIndex","hashs","size","baseName","data","stat","ComputeHashIndexCache","indexFilePath","indexLine","chokidar","readdir","PQueue","FolderWatcher","fileProcessor","WATCH_FOLDER_LIST","config","directory","promises","entries","entry","fullPath","promise","countPromises","acc","current","dirname","folderList","chokidarconfig","watcher","error","folders","i","count"],"mappings":"uDAAA,OAAQ,WAAAA,MAAc,UAItB,IAAIC,EAAa,YAAY,QAC7BA,EAAaA,EAAW,QAAQ,MAAO,MAAM,EAC7CA,EAAaA,EAAa,aAC1B,IAAMC,EAAY,IAAI,IAAID,CAAU,EAAE,KACtC,QAAQ,IAAI,eAAgBC,CAAS,EAE9B,IAAMC,EAAN,KAAqB,CAChB,QAER,YAAYC,EAAoB,CAC5B,KAAK,QAAU,IAAIJ,EAAQ,CACvB,WAAY,EAEZ,SAAUI,GAAc,QAAQ,IAAI,YAAcF,CACtD,CAAC,CACL,CAQA,MAAa,YAAYG,EAAkBC,EAAsD,CAC7F,OAAO,KAAK,QAAQ,IAAI,CAAC,SAAAD,EAAU,WAAAC,CAAU,CAAC,CAClD,CACJ,EC3BO,IAAMC,EAAN,KAA8C,CAGjD,YAAoBC,EAAkCJ,EAAoB,CAAtD,gBAAAI,EAChB,KAAK,eAAiB,IAAIL,EAAeC,CAAU,CACvD,CAJQ,eAMR,MAAM,mBAAmBC,EAAkBI,EAAwC,CAE/E,IAAMC,EAAe,KAAK,WAAW,OAAOC,GAAY,CAACF,EAASE,CAAQ,CAAC,EAG3E,GAAID,EAAa,SAAW,EACxB,QAIS,MAAM,KAAK,eAAe,YAAYL,EAAUK,CAAY,GAGpE,QAAQ,CAACE,EAAKC,IAAU,CACzB,IAAMC,EAAWJ,EAAaG,CAAK,EACnCJ,EAASK,CAAQ,EAAIF,CACzB,CAAC,CACL,CACJ,EC5BA,OAAQ,oBAAAG,EAAkB,YAAYC,MAAS,KAC/C,OAAQ,SAAAC,MAAY,YACpB,OAAQ,SAASC,MAAgB,iBACjC,OAAQ,aAAAC,MAAgB,qBCDxB,OAAQ,UAAAC,EAAO,aAAAC,MAAgB,cAE/B,eAAsBC,EAAYjB,EAAoC,CAClE,GAAI,CACA,aAAMe,EAAOf,EAAUgB,EAAU,IAAI,EAC9B,EACX,MAAQ,CACJ,MAAO,EACX,CACJ,CDNA,OAAOE,MAAU,OACjB,OAAQ,iBAAAC,MAAoB,SAQrB,IAAMC,EAAgB,CAAC,OAAQ,OAAQ,OAAO,EAExCC,EAAN,KAAuB,CAU1B,YAAoBrB,EAAyBG,EAAkC,0BAAoD,EAAG,CAAlH,cAAAH,EAAyB,gBAAAG,EACzC,GAAI,CAACH,EACD,MAAM,IAAI,MAAM,yBAAyB,CAEjD,CAbQ,MAAgC,IAAI,IACpC,WACA,aAAuB,IACvB,kBAA4B,EAC5B,cACA,mBAA8B,GAC9B,WAAsB,GACtB,YAQD,UAAmC,CACtC,OAAO,IAAI,IAAI,KAAK,KAAK,CAC7B,CAMA,MAAM,KAAKsB,EAAW,GAAM,CACxB,OAAK,KAAK,cACN,KAAK,YAAc,IAAI,QAAc,MAAOC,EAASC,IAAW,CAC5D,GAAI,CACA,GAAI,CAAC,KAAK,gBAAgB,KAAK,QAAQ,EACnC,MAAM,IAAI,MAAM,4BAA4B,EAEhD,MAAM,KAAK,UAAU,EACjBF,GACA,KAAK,MAAM,EAEfC,EAAQ,CACZ,OAAQE,EAAG,CACPD,EAAOC,CAAC,CACZ,CACJ,CAAC,GAEE,KAAK,WAChB,CAGA,gBAAiBC,EAA6B,CAC1C,IAAMC,EAAUd,EAAUa,EAAY,CAClC,IAAK,GACL,QAAS,GACT,iBAAkB,EACtB,CAAC,EAED,GAAIC,EAAQ,SAAW,EAEnB,MAAO,GAIX,IAAMC,EAAU,OAAO,KAAKD,EAAQ,CAAC,CAAC,EAOtC,MAHwB,CAAC,OAAQ,OAAQ,QAAS,GAAG,KAAK,UAAU,EAG7C,MAAME,GAAUD,EAAQ,SAASC,CAAM,CAAC,CACnE,CAEO,OAAQ,CACX,KAAK,cAAc,KAAK,YAAY,CACxC,CAEO,cAAe,CACd,KAAK,YACLV,EAAc,KAAK,UAAU,CAErC,CAEQ,cAAcW,EAAc,CAChC,KAAK,aAAa,EAClB,KAAK,aAAeA,EACpB,KAAK,WAAa,YAAY,IAAM,KAAK,gBAAgB,EAAGA,CAAI,CACpE,CAEA,MAAa,WAAkC,CAC3C,GAAI,MAAMb,EAAY,KAAK,QAAQ,EAAG,CAElC,IAAMc,EAAQ,MAAMpB,EAAG,KAAK,KAAK,QAAQ,EACzC,GAAI,KAAK,oBAAsBoB,EAAM,KAAM,CAEvC,IAAMJ,EAAuB,MAAM,KAAK,QAAQ,EAChD,QAAWK,KAAUL,EACjB,KAAK,MAAM,IAAIK,EAAO,KAAMA,CAAM,EAEtC,YAAK,kBAAoBD,EAAM,KAC/B,KAAK,cAAgBJ,EACdA,CACX,KACI,QAAO,KAAK,aAEpB,CACA,MAAO,CAAC,CACZ,CAEQ,oBAAkC,CACtC,OAAO,MAAM,KAAK,KAAK,MAAM,OAAO,CAAC,CACzC,CAEA,MAAc,SAAgC,CAC1C,GAAI,CAAE,MAAMV,EAAY,KAAK,QAAQ,EACjC,MAAO,CAAC,EAEZ,IAAMgB,EAAQ,YAAY,IAAI,EAExBC,EAAStB,EAAM,CACjB,QAAS,GACT,iBAAkB,EACtB,CAAC,EAEKe,EAAuB,CAAC,EAE9B,OAAO,IAAI,QAAQ,CAACJ,EAASC,IAAW,CACpCd,EAAiB,KAAK,QAAQ,EACzB,KAAKwB,CAAM,EACX,GAAG,OAASF,GAAW,CACpBL,EAAQ,KAAKK,CAAM,CACvB,CAAC,EAAE,GAAG,MAAO,IAAM,CACnBT,EAAQI,CAAO,EACf,QAAQ,IAAI,mBAAmB,YAAY,IAAI,EAAIM,CAAK,IAAI,CAChE,CAAC,EAAE,GAAG,QAAUE,GAAQ,CACpBX,EAAOW,CAAG,CACd,CAAC,CACL,CAAC,CACL,CAEA,MAAc,iBAAiC,CAC3C,GAAI,KAAK,oBAAsB,CAAC,KAAK,WACjC,OAEJ,KAAK,WAAa,GAClB,KAAK,mBAAqB,GAC1B,IAAMF,EAAQ,YAAY,IAAI,EAC1BG,EAA4B,MAAM,KAAK,UAAU,EAErD,GAAI,KAAK,MAAM,OAAS,EAAG,CAGvB,IAAMC,EAFyB,KAAK,mBAAmB,EAE7B,OAAOC,GAAO,CAACF,EAAa,KAAKG,GAAeA,EAAY,OAASD,EAAI,IAAI,CAAC,EAExG,GAAID,EAAQ,SAAW,EAAG,CAEtB,IAAMG,EAAY1B,EAAUuB,EAAS,CACjC,OAAQD,EAAa,SAAW,EAChC,QAAS,CACL,CAAC,IAAK,OAAQ,OAAQ,MAAM,EAC5B,CAAC,IAAK,OAAQ,OAAQ,MAAM,EAC5B,CAAC,IAAK,QAAS,OAAQ,OAAO,EAC9B,GAAG,KAAK,WAAW,IAAIK,IAAS,CAAC,IAAKA,EAAM,OAAQA,CAAI,EAAE,CAC9D,CACJ,CAAC,EAGD,MAAM9B,EAAG,WAAW,KAAK,SAAU6B,CAAS,CAChD,CACJ,CACA,IAAME,EAAY,YAAY,IAAI,EAAIT,EACtC,QAAQ,IAAI,kBAAkBS,CAAS,IAAI,EAEvCA,EAAY,GAAK,KAAK,eACtB,KAAK,cAAcA,EAAY,EAAE,EACjC,QAAQ,IAAI,oCAAoCA,EAAY,EAAE,IAAI,GAEtE,KAAK,mBAAqB,EAC9B,CAEO,cAAc1C,EAAkB2C,EAAkBC,EAA0B,CAC/E,IAAMC,EAAW3B,EAAK,SAASlB,CAAQ,EACnC8C,EAAgB,KAAK,MAAM,IAAID,CAAQ,EAC3C,GAAGC,GACC,GAAGA,EAAc,OAEb,GAAIA,EAAc,OAAUH,EAAW,IAAOG,EAAc,QAAUF,EAClE,OAAOE,UAIPA,EAAc,OAAUH,EAAW,GACnC,OAAOG,EAWnB,OAAO,IACX,CAEO,WAAW9C,EAAkB2C,EAAkBC,EAAeG,EAA2D,CAC5H,GAAI,CAAC/C,GAAY,CAAC2C,GAAY,CAACC,GAAS,CAACG,EACrC,MAAM,IAAI,MAAM,oBAAoB,EAExC,IAAMC,EAAOL,EAAW,GAClBM,EAAW/B,EAAK,SAASlB,CAAQ,EACjCkD,EAAO,CAAC,KAAMD,EAAU,KAAMD,EAAM,MAAOJ,EAAO,GAAGG,CAAK,EAChE,KAAK,MAAM,IAAIE,EAAUC,CAAI,EAC7B,KAAK,WAAa,EACtB,CAEJ,EEjOA,OAAQ,QAAAC,MAAW,cACnB,OAAOjC,MAAU,OAEV,IAAMkC,EAAN,KAAwD,CAI3D,YAAYC,EAA8BlD,EAAoC,0BAAqD,EAAEJ,EAAoB,CAA/G,gBAAAI,EACtC,KAAK,aAAe,IAAID,EAAa,KAAK,WAAWH,CAAU,EAC/D,KAAK,iBAAmB,IAAIsB,EAAiBgC,CAAa,CAC9D,CANA,iBACA,aAOA,MAAM,mBAAmBrD,EAAkBI,EAAwC,CAC/E,MAAM,KAAK,iBAAiB,KAAK,EACjC,IAAI2B,EAAQ,MAAMoB,EAAKnD,CAAQ,EAC/B,GAAI,KAAK,iBAAiB,SAAS,EAAE,IAAIkB,EAAK,SAASlB,CAAQ,CAAC,EAAG,CAC/D,IAAMsD,EAAY,KAAK,iBAAiB,cAActD,EAAU+B,EAAM,KAAMA,EAAM,MAAM,YAAY,CAAC,EACrG,GAAGuB,EACC,QAAWb,KAAQ,KAAK,WAChB,CAACrC,EAASqC,CAAI,GAAKa,EAAUb,CAAI,IACjCrC,EAASqC,CAAI,EAAIa,EAAUb,CAAI,EAI/C,CACA,MAAM,KAAK,aAAa,mBAAmBzC,EAAUI,CAAQ,EAC7D,KAAK,iBAAiB,WAAWJ,EAAU+B,EAAM,KAAMA,EAAM,MAAM,YAAY,EAAG3B,CAAQ,CAC9F,CACJ,EC/BA,UAAYc,MAAU,OACtB,OAAOqC,MAAc,WACrB,OAAQ,WAAAC,MAAc,cACtB,OAAOC,MAAY,UAGZ,IAAMC,EAAN,KAAoB,CAQvB,YAAYC,EAA6CC,EAAmCC,EAIzF,CAJsD,uBAAAD,EAAmC,YAAAC,EAKxF,KAAK,cAAgBF,CACzB,CAbA,YAAc,GACd,MAAQ,IAAIF,EAAO,CAAC,YAAa,IAAK,UAAW,EAAI,CAAC,EACtD,cACA,UAAY,GACZ,QAAU,EACV,WAAa,IAAI,IAgBjB,MAAc,iBAAiBK,EAAmBC,EAA0B,CACxE,IAAMC,EAAU,MAAMR,EAAQM,EAAW,CAAC,cAAe,EAAI,CAAC,EAC9D,QAAWG,KAASD,EAAS,CACzB,IAAME,EAAgB,OAAKJ,EAAWG,EAAM,IAAI,EAChD,GAAIA,EAAM,YAAY,EAClB,MAAM,KAAK,iBAAiBC,EAAUH,CAAQ,MAC3C,CACH,IAAMI,EAAU,KAAK,YAAYD,CAAQ,EACzCH,EAAS,KAAKI,CAAO,CACzB,CACJ,CACJ,CAEA,MAAc,UAAUL,EAAoC,CACxD,IAAME,EAAU,MAAMR,EAAQM,EAAW,CAAC,cAAe,EAAI,CAAC,EACxDM,EAAgC,CAAC,EAEvC,QAAWH,KAASD,EAAS,CACzB,IAAME,EAAgB,OAAKJ,EAAWG,EAAM,IAAI,EAC5CA,EAAM,YAAY,EAElBG,EAAc,KAAK,KAAK,UAAUF,CAAQ,CAAC,EAG3CE,EAAc,KAAK,KAAK,MAAM,IAC1B,SAAa,MAAM,KAAK,cAAc,eAAeF,CAAQ,EAAK,EAAI,CAC1E,CAAC,CAET,CAIA,OADe,MAAM,QAAQ,IAAIE,CAAa,GAChC,OAAO,CAACC,EAAKC,IAAYD,EAAMC,EAAS,CAAC,CAC3D,CAEA,MAAc,YAAYtE,EAAgC,CACtD,GAAG,MAAM,KAAK,cAAc,eAAeA,CAAQ,GAC5C,CAAC,KAAK,WAAW,IAAIA,CAAQ,EAAG,CAC/B,KAAK,WAAW,IAAIA,CAAQ,EAC5B,IAAMsE,EAAU,EAAE,KAAK,QACvB,MAAM,KAAK,MAAM,IAAI,IAAM,KAAK,cAAc,YAAYA,EAAS,KAAK,UAAWtE,CAAQ,CAAC,EAC5F,KAAK,WAAW,OAAOA,CAAQ,CACnC,CAER,CAOA,MAAc,oBAAoBA,EAAgC,CAC9D,IAAIuE,EAAe,UAAQvE,CAAQ,EACnC,MAAM,KAAK,iBAAiBuE,EAAS,CAAC,CAAC,CAC3C,CAEQ,cAAcC,EAAqB,CACvC,IAAMC,EAAiB,CACnB,cAAe,GACf,WAAY,GACZ,MAAO,IACP,iBAAkB,CACd,mBAAoB,KAAK,OAAO,oBAAsB,IACtD,aAAc,KAAK,OAAO,cAAgB,GAC9C,CACJ,EAEIC,GACC,KAAK,OAAO,UAAY,IAAM,EAC/BA,EAAUnB,EAAS,MAAMiB,EAAYC,CAAc,EAEnDC,EAAUnB,EAAS,MAAMiB,EAAY,CACjC,GAAGC,EACH,WAAY,GACZ,SAAU,KAAK,OAAO,UAAY,CACtC,CAAC,EAGLC,EACK,GAAG,MAAO,MAAO1E,GAAa,CAC3B,GAAI,CACA,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAAS2E,EAAO,CACZ,QAAQ,MAAM,yBAAyB3E,CAAQ,IAAK2E,CAAK,CAC7D,CACJ,CAAC,EACA,GAAG,SAAU,MAAO3E,GAAa,CAC9B,GAAI,CACA,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAASyB,EAAG,CACR,QAAQ,MAAM,yBAAyBzB,CAAQ,IAAKyB,CAAC,CACzD,CACJ,CAAC,EACA,GAAG,SAAU,MAAOzB,GAAa,CAC9B,GAAI,CACA,MAAM,KAAK,cAAc,WAAWA,CAAQ,EAC5C,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAASyB,EAAG,CACR,QAAQ,MAAM,kCAAmCA,CAAC,CACtD,CACJ,CAAC,EACA,GAAG,QAAUkD,GAAU,QAAQ,MAAM,kBAAkBA,CAAK,EAAE,CAAC,EAC/D,GAAG,QAAS,IAAM,CACf,QAAQ,IAAI,gCAAgCH,CAAU,EAAE,CAC5D,CAAC,CACT,CAEA,MAAM,OAAQ,CACV,IAAII,EAAU,KAAK,kBACnB,GAAI,CAACA,EACD,MAAM,IAAI,MAAM,oBAAoB,EAExC,IAAIJ,EAAaI,EAAQ,MAAM,GAAG,EAElC,KAAK,cAAcJ,CAAU,EAG7B,IAAIJ,EAAgB,CAAC,EACrB,QAASS,EAAI,EAAGA,EAAIL,EAAW,OAAQK,IACnCL,EAAWK,CAAC,EAAS,YAAUL,EAAWK,CAAC,CAAC,EAC5CT,EAAc,KAAK,KAAK,UAAUI,EAAWK,CAAC,CAAC,CAAC,EAEpD,IAAMC,GAAS,MAAM,QAAQ,IAAIV,CAAa,GAAG,OAAO,CAACC,EAAKC,IAAYD,EAAMC,EAAS,CAAC,EAC1F,QAAQ,IAAI,SAASQ,CAAK,mBAAmB,EAC7C,KAAK,UAAYA,EACjB,IAAIf,EAA2B,CAAC,EAChC,QAASc,EAAI,EAAGA,EAAIL,EAAW,OAAQK,IACnCL,EAAWK,CAAC,EAAS,YAAUL,EAAWK,CAAC,CAAC,EAC5C,MAAM,KAAK,iBAAiBL,EAAWK,CAAC,EAAGd,CAAQ,EAEvD,MAAM,QAAQ,IAAIA,CAAQ,EAC1B,KAAK,YAAc,GAEnB,QAAQ,IAAI,eAAe,CAC/B,CACJ","sourcesContent":["import {Piscina} from \"piscina\";\nimport {CID_ALGORITHM_NAMES} from \"../MultiHashData\";\n\n// Construct the URL for the current module\nlet distFolder = import.meta.dirname;\ndistFolder = distFolder.replace('src', 'dist');\ndistFolder = distFolder + \"/worker.js\";\nconst workerUrl = new URL(distFolder).href;\nconsole.log(\"Worker URL: \", workerUrl);\n\nexport class FileIDComputer {\n private piscina: Piscina;\n\n constructor(workerPath?:string) {\n this.piscina = new Piscina({\n maxThreads: 4,\n //filename: new URL('./ShaComputeWorker.ts', import.meta.url).href\n filename: workerPath || process.env.WORKER_URL || workerUrl\n });\n }\n\n /**\n * Compute the CIDs of a file using specific algorithms\n * @param filePath The path to the file\n * @param algorithms Array of algorithms ('sha256', 'sha1')\n * @returns Array of CIDs (in the order of the algorithms)\n */\n public async computeCIDs(filePath: string, algorithms: CID_ALGORITHM_NAMES[]): Promise {\n return this.piscina.run({filePath, algorithms});\n }\n}\n","import { FileIDComputer } from \"./fileID/FileIDComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\n\nexport class HashComputer implements ComputeInterface{\n private fileIDComputer:FileIDComputer;\n\n constructor(private targetHash: CID_ALGORITHM_NAMES[],workerPath?:string) {\n this.fileIDComputer = new FileIDComputer(workerPath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n // Dynamically determine which hashes are needed\n const neededHashes = this.targetHash.filter(hashName => !metadata[hashName]);\n\n // If all hashes are already computed, skip the processing\n if (neededHashes.length === 0) {\n return;\n }\n\n // Compute only the needed CIDs\n const cids = await this.fileIDComputer.computeCIDs(filePath, neededHashes);\n\n // Map the computed CIDs back to their respective metadata properties\n cids.forEach((cid, index) => {\n const hashType = neededHashes[index];\n metadata[hashType] = cid;\n });\n }\n}\n","import {createReadStream, promises as fs} from 'fs';\nimport {parse} from 'csv-parse';\nimport {parse as parseSync} from 'csv-parse/sync';\nimport {stringify} from 'csv-stringify/sync';\nimport {existsAsync} from \"./ExistsAsync\";\nimport path from \"path\";\nimport {clearInterval} from \"node:timers\";\nimport {CID_ALGORITHM_NAMES} from \"./MultiHashData\";\n\ninterface IndexLine extends Partial> {\n path: string;\n size: string;\n mtime: string;\n}\nexport const INDEX_HEADERS = ['path', 'size', 'mtime'];\n\nexport class HashIndexManager {\n private cache: Map = new Map();\n private intervalId: NodeJS.Timeout;\n private intervalTime: number = 30000;\n private lastIndexFileSize: number = 0; //size of the index file last time it was read\n private lastCacheFile: IndexLine[]; //state of the file last time it was read\n private indexOpsInProgress: boolean = false;\n private hasChanged: boolean = false;\n private initialLoad;\n\n constructor(private filePath: string,private targetHash:CID_ALGORITHM_NAMES[]= [CID_ALGORITHM_NAMES.sha1,CID_ALGORITHM_NAMES.sha256]) {\n if (!filePath) {\n throw new Error('Invalid index file path');\n }\n }\n\n public getCache(): Map {\n return new Map(this.cache);\n }\n\n /**\n * After init consseutively calls to this method will not reload the index\n * @param autosave\n */\n async init(autosave = true) {\n if (!this.initialLoad) {\n this.initialLoad = new Promise(async (resolve, reject) => {\n try {\n if (!this.checkCSVHeaders(this.filePath)) {\n throw new Error('Invalid index file headers');\n }\n await this.loadIndex();\n if (autosave) {\n this.start();\n }\n resolve();\n }catch (e) {\n reject(e);\n }\n });\n }\n return this.initialLoad;\n }\n\n // Function to check CSV headers\n checkCSVHeaders (csvContent: string): boolean {\n const records = parseSync(csvContent, {\n bom: true,\n columns: true,\n skip_empty_lines: true,\n });\n\n if (records.length === 0) {\n //we will write the headers\n return true;\n }\n\n // Extract headers\n const headers = Object.keys(records[0]);\n\n // Define required headers\n //from the IndexLine interface\n const requiredHeaders = ['path', 'size', 'mtime', ...this.targetHash];\n\n // Check if all required headers are present\n return requiredHeaders.every(header => headers.includes(header));\n };\n\n public start() {\n this.startAutoSave(this.intervalTime);\n }\n\n public stopAutoSave() {\n if (this.intervalId) {\n clearInterval(this.intervalId);\n }\n }\n\n private startAutoSave(time: number) {\n this.stopAutoSave();\n this.intervalTime = time;\n this.intervalId = setInterval(() => this.saveCacheToFile(), time);\n }\n\n public async loadIndex(): Promise {\n if (await existsAsync(this.filePath)) {\n // check the file size and if it did not change, do not read the file\n const stats = await fs.stat(this.filePath);\n if (this.lastIndexFileSize !== stats.size) {\n // Read existing file content and parse it\n const records: IndexLine[] = await this.readCsv();\n for (const record of records) {\n this.cache.set(record.path, record);\n }\n this.lastIndexFileSize = stats.size;\n this.lastCacheFile = records;\n return records;\n } else {\n return this.lastCacheFile;\n }\n }\n return [];\n }\n\n private loadIndexFromCache(): IndexLine[] {\n return Array.from(this.cache.values());\n }\n\n private async readCsv(): Promise {\n if (!(await existsAsync(this.filePath))) {\n return [];\n }\n const start = performance.now();\n\n const parser = parse({\n columns: true,\n skip_empty_lines: true,\n });\n\n const records: IndexLine[] = [];\n\n return new Promise((resolve, reject) => {\n createReadStream(this.filePath)\n .pipe(parser)\n .on('data', (record) => {\n records.push(record);\n }).on('end', () => {\n resolve(records);\n console.log(`Index read time ${performance.now() - start}ms`);\n }).on('error', (err) => {\n reject(err);\n });\n });\n }\n\n private async saveCacheToFile(): Promise {\n if (this.indexOpsInProgress || !this.hasChanged) {\n return;\n }\n this.hasChanged = false;\n this.indexOpsInProgress = true;\n const start = performance.now();\n let existingRows: IndexLine[] = await this.loadIndex();\n\n if (this.cache.size !== 0) {\n const cacheRows: IndexLine[] = this.loadIndexFromCache();\n // Filter out cacheRows that are already in the file\n const newRows = cacheRows.filter(row => !existingRows.find(existingRow => existingRow.path === row.path));\n\n if (newRows.length !== 0) {\n // Serialize new cacheRows to CSV string\n const csvString = stringify(newRows, {\n header: existingRows.length === 0, // Only add header if the file was empty\n columns: [\n {key: 'path', header: 'path'},\n {key: 'size', header: 'size'},\n {key: 'mtime', header: 'mtime'},\n ...this.targetHash.map(hash => ({key: hash, header: hash})),\n ],\n });\n\n // Append new cacheRows to the file\n await fs.appendFile(this.filePath, csvString);\n }\n }\n const totalTime = performance.now() - start;\n console.log(`Index saved in ${totalTime}ms`);\n // Check if the time to save the index is greater than the interval time. increase the interval time if needed\n if (totalTime * 10 > this.intervalTime) {\n this.startAutoSave(totalTime * 10);\n console.log(`Index save interval increased to ${totalTime * 10}ms`);\n }\n this.indexOpsInProgress = false;\n }\n\n public getCidForFile(filePath: string, fileSize: number, mtime: string): IndexLine {\n const fileName = path.basename(filePath);\n let fileNameIndex = this.cache.get(fileName);\n if(fileNameIndex) {\n if(fileNameIndex.mtime) {\n //if we have a mtime, we need to check it\n if (fileNameIndex.size === (fileSize + \"\") && fileNameIndex.mtime === mtime) {\n return fileNameIndex;\n }\n }else {\n //mtime is optional\n if (fileNameIndex.size === (fileSize + \"\")) {\n return fileNameIndex;\n }\n }\n }\n\n // 3 - if not found, delete the entry (keeps the index clean)\n /*if (fileNameIndex && fileNameIndex.size !== (fileSize + \"\") && pathIndex && pathIndex.size !== (fileSize + \"\")) {\n this.cache.delete(fileName);\n this.cache.delete(filePath);\n }*/\n\n return null;\n }\n\n public addFileCid(filePath: string, fileSize: number, mtime: string, hashs: Partial>): void {\n if (!filePath || !fileSize || !mtime || !hashs) {\n throw new Error('Invalid parameters');\n }\n const size = fileSize + \"\";\n const baseName = path.basename(filePath);\n const data = {path: baseName, size: size, mtime: mtime, ...hashs};\n this.cache.set(baseName, data);\n this.hasChanged = true;\n }\n\n}","// Example of checking if a file exists asynchronously\n// Note: `existsSync` does not have a direct async equivalent, so we use `access` instead.\nimport {access,constants} from \"fs/promises\";\n\nexport async function existsAsync(filePath: string): Promise {\n try {\n await access(filePath, constants.F_OK);\n return true; // File exists\n } catch {\n return false; // File does not exist\n }\n}","import {HashComputer} from \"./HashComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\nimport {HashIndexManager} from \"./HashIndexManager\";\nimport {stat} from \"fs/promises\";\nimport path from \"path\";\n\nexport class ComputeHashIndexCache implements ComputeInterface {\n hashIndexManager: HashIndexManager;\n hashComputer: HashComputer;\n\n constructor(indexFilePath: string,private targetHash: CID_ALGORITHM_NAMES[] = [CID_ALGORITHM_NAMES.sha1, CID_ALGORITHM_NAMES.sha256],workerPath?:string) {\n this.hashComputer = new HashComputer(this.targetHash,workerPath);\n this.hashIndexManager = new HashIndexManager(indexFilePath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n await this.hashIndexManager.init();\n let stats = await stat(filePath);\n if (this.hashIndexManager.getCache().has(path.basename(filePath))) {\n const indexLine = this.hashIndexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());\n if(indexLine) {\n for (const hash of this.targetHash) {\n if (!metadata[hash] && indexLine[hash]) {\n metadata[hash] = indexLine[hash];\n }\n }\n }\n }\n await this.hashComputer.computeMissingHash(filePath, metadata);\n this.hashIndexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), metadata);\n }\n}","import * as path from 'path';\nimport chokidar from 'chokidar';\nimport {readdir} from \"fs/promises\";\nimport PQueue from \"p-queue\";\nimport {FileProcessorInterface} from \"./FileProcessorInterface.js\";\n\nexport class FolderWatcher {\n initialized = false;\n queue = new PQueue({concurrency: 100, autoStart: true});\n fileProcessor: FileProcessorInterface;\n queueSize = -1;\n current = 0;\n processing = new Set();\n\n constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{\n interval?: number,\n stabilityThreshold?: number,\n pollInterval?: number,\n }) {\n this.fileProcessor = fileProcessor;\n }\n\n /**\n * Process the directory and its subdirectories\n * @param directory\n * @param promises\n * @private\n */\n private async processDirectory(directory: string, promises: Promise[]) {\n const entries = await readdir(directory, {withFileTypes: true});\n for (const entry of entries) {\n const fullPath = path.join(directory, entry.name);\n if (entry.isDirectory()) {\n await this.processDirectory(fullPath, promises);\n } else {\n const promise = this.processFile(fullPath);\n promises.push(promise);\n }\n }\n }\n\n private async countFile(directory: string): Promise {\n const entries = await readdir(directory, {withFileTypes: true});\n const countPromises: Promise[] = [];\n\n for (const entry of entries) {\n const fullPath = path.join(directory, entry.name);\n if (entry.isDirectory()) {\n // Push the promise without waiting for it here\n countPromises.push(this.countFile(fullPath));\n } else {\n // Use an IIFE (Immediately Invoked Function Expression) to handle asynchronous condition\n countPromises.push(this.queue.add(\n async () => (await this.fileProcessor.canProcessFile(fullPath)) ? 1 : 0\n ));\n }\n }\n\n // Wait for all promises to resolve and then sum up their results\n const counts = await Promise.all(countPromises);\n return counts.reduce((acc, current) => acc + current, 0);\n }\n\n private async processFile(filePath: string):Promise {\n if(await this.fileProcessor.canProcessFile(filePath)) {\n if(!this.processing.has(filePath)) {\n this.processing.add(filePath);//avoid double processing in the queue\n const current = ++this.current;\n await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath));\n this.processing.delete(filePath);\n }\n }\n }\n\n /**\n * Process the file and its sibling file included other folder it is a recursive function\n * @param filePath\n * @private\n */\n private async processFileExtended(filePath: string):Promise {\n let dirname = path.dirname(filePath);\n await this.processDirectory(dirname, []);\n }\n\n private chokidarWatch(folderList: string[]){\n const chokidarconfig = {\n ignoreInitial: true,//ignore the initial scan we have our own function\n persistent: true,\n depth: Infinity,\n awaitWriteFinish: {\n stabilityThreshold: this.config.stabilityThreshold || 30000,\n pollInterval: this.config.pollInterval || 5000\n }\n };\n\n let watcher: chokidar.FSWatcher;\n if ((this.config.interval || 0) <= 0) {\n watcher = chokidar.watch(folderList, chokidarconfig);\n } else {\n watcher = chokidar.watch(folderList, {\n ...chokidarconfig,\n usePolling: true,\n interval: this.config.interval || 0\n });\n }\n\n watcher\n .on('add', async (filePath) => {\n try {\n await this.processFileExtended(filePath);\n } catch (error) {\n console.error(`Error processing file ${filePath}:`, error);\n }\n })\n .on('change', async (filePath) => {\n try {\n await this.processFileExtended(filePath);\n } catch (e) {\n console.error(`Error processing file ${filePath}:`, e);\n }\n })\n .on('unlink', async (filePath) => {\n try {\n await this.fileProcessor.deleteFile(filePath);\n await this.processFileExtended(filePath);\n } catch (e) {\n console.error(`Error processing file deletion:`, e);\n }\n })\n .on('error', (error) => console.error(`Watcher error: ${error}`))\n .on('ready', () => {\n console.log(`Watching for file changes on ${folderList}`);\n });\n }\n\n async watch() {\n let folders = this.WATCH_FOLDER_LIST;\n if (!folders) {\n throw new Error(`No folder to watch`);\n }\n let folderList = folders.split(',');\n\n this.chokidarWatch(folderList);\n\n //start the initial scan\n let countPromises = [];\n for (let i = 0; i < folderList.length; i++) {\n folderList[i] = path.normalize(folderList[i]);\n countPromises.push(this.countFile(folderList[i]));\n }\n const count = (await Promise.all(countPromises)).reduce((acc, current) => acc + current, 0);\n console.log(`Found ${count} files to process`);\n this.queueSize = count;\n let promises: Promise[] = [];\n for (let i = 0; i < folderList.length; i++) {\n folderList[i] = path.normalize(folderList[i]);\n await this.processDirectory(folderList[i], promises);\n }\n await Promise.all(promises);\n this.initialized = true;\n\n console.log(`Watcher ready`);\n }\n}\n"]} \ No newline at end of file diff --git a/dist/index.d.cts b/dist/index.d.cts index 561cb40..40ffe02 100644 --- a/dist/index.d.cts +++ b/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; + canProcessFile(filePath: string): Promise; + deleteFile(filePath: string): Promise; +} + +declare class FolderWatcher { + private WATCH_FOLDER_LIST; + private config; + initialized: boolean; + queue: PQueue__default; + fileProcessor: FileProcessorInterface; + queueSize: number; + current: number; + processing: Set; + 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; +} + +export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, type FileProcessorInterface, FolderWatcher, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync }; diff --git a/dist/index.d.ts b/dist/index.d.ts index 334ca22..940af05 100644 --- a/dist/index.d.ts +++ b/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; + canProcessFile(filePath: string): Promise; + deleteFile(filePath: string): Promise; +} + +declare class FolderWatcher { + private WATCH_FOLDER_LIST; + private config; + initialized: boolean; + queue: PQueue__default; + fileProcessor: FileProcessorInterface; + queueSize: number; + current: number; + processing: Set; + 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; +} + +export { CID_ALGORITHM_NAMES, ComputeHashIndexCache, ComputeInterface, FileIDComputer, type FileProcessorInterface, FolderWatcher, HashComputer, HashIndexManager, INDEX_HEADERS, MultiHashData, type Properties, existsAsync }; diff --git a/dist/index.js b/dist/index.js index fa7fd1c..6e75969 100644 --- a/dist/index.js +++ b/dist/index.js @@ -4,12 +4,15 @@ import { promises, createReadStream } from 'fs'; import { parse as parse$1 } from 'csv-parse'; import { parse } from 'csv-parse/sync'; import { stringify } from 'csv-stringify/sync'; -import { access, constants, stat } from 'fs/promises'; -import I from 'path'; +import { access, constants, stat, readdir } from 'fs/promises'; +import * as n from 'path'; +import n__default from 'path'; import { clearInterval } from 'node:timers'; +import w from 'chokidar'; +import b from 'p-queue'; -var h=import.meta.dirname;h=h.replace("src","dist");h=h+"/worker.js";var f=new URL(h).href;console.log("Worker URL: ",f);var c=class{piscina;constructor(e){this.piscina=new Piscina({maxThreads:4,filename:e||process.env.WORKER_URL||f});}async computeCIDs(e,t){return this.piscina.run({filePath:e,algorithms:t})}};var m=class{constructor(e,t){this.targetHash=e;this.fileIDComputer=new c(t);}fileIDComputer;async computeMissingHash(e,t){let i=this.targetHash.filter(r=>!t[r]);if(i.length===0)return;(await this.fileIDComputer.computeCIDs(e,i)).forEach((r,a)=>{let n=i[a];t[n]=r;});}};async function l(o){try{return await access(o,constants.F_OK),!0}catch{return !1}}var q=["path","size","mtime"],p=class{constructor(e,t=["cid_sha1","cid_sha2-256"]){this.filePath=e;this.targetHash=t;if(!e)throw new Error("Invalid index file path")}cache=new Map;intervalId;intervalTime=3e4;lastIndexFileSize=0;lastCacheFile;indexOpsInProgress=!1;hasChanged=!1;initialLoad;getCache(){return new Map(this.cache)}async init(e=!0){return this.initialLoad||(this.initialLoad=new Promise(async(t,i)=>{try{if(!this.checkCSVHeaders(this.filePath))throw new Error("Invalid index file headers");await this.loadIndex(),e&&this.start(),t();}catch(s){i(s);}})),this.initialLoad}checkCSVHeaders(e){let t=parse(e,{bom:!0,columns:!0,skip_empty_lines:!0});if(t.length===0)return !0;let i=Object.keys(t[0]);return ["path","size","mtime",...this.targetHash].every(r=>i.includes(r))}start(){this.startAutoSave(this.intervalTime);}stopAutoSave(){this.intervalId&&clearInterval(this.intervalId);}startAutoSave(e){this.stopAutoSave(),this.intervalTime=e,this.intervalId=setInterval(()=>this.saveCacheToFile(),e);}async loadIndex(){if(await l(this.filePath)){let e=await promises.stat(this.filePath);if(this.lastIndexFileSize!==e.size){let t=await this.readCsv();for(let i of t)this.cache.set(i.path,i);return this.lastIndexFileSize=e.size,this.lastCacheFile=t,t}else return this.lastCacheFile}return []}loadIndexFromCache(){return Array.from(this.cache.values())}async readCsv(){if(!await l(this.filePath))return [];let e=performance.now(),t=parse$1({columns:!0,skip_empty_lines:!0}),i=[];return new Promise((s,r)=>{createReadStream(this.filePath).pipe(t).on("data",a=>{i.push(a);}).on("end",()=>{s(i),console.log(`Index read time ${performance.now()-e}ms`);}).on("error",a=>{r(a);});})}async saveCacheToFile(){if(this.indexOpsInProgress||!this.hasChanged)return;this.hasChanged=!1,this.indexOpsInProgress=!0;let e=performance.now(),t=await this.loadIndex();if(this.cache.size!==0){let r=this.loadIndexFromCache().filter(a=>!t.find(n=>n.path===a.path));if(r.length!==0){let a=stringify(r,{header:t.length===0,columns:[{key:"path",header:"path"},{key:"size",header:"size"},{key:"mtime",header:"mtime"},...this.targetHash.map(n=>({key:n,header:n}))]});await promises.appendFile(this.filePath,a);}}let i=performance.now()-e;console.log(`Index saved in ${i}ms`),i*10>this.intervalTime&&(this.startAutoSave(i*10),console.log(`Index save interval increased to ${i*10}ms`)),this.indexOpsInProgress=!1;}getCidForFile(e,t,i){let s=I.basename(e),r=this.cache.get(s);if(r){if(r.mtime){if(r.size===t+""&&r.mtime===i)return r}else if(r.size===t+"")return r}return null}addFileCid(e,t,i,s){if(!e||!t||!i||!s)throw new Error("Invalid parameters");let r=t+"",a=I.basename(e),n={path:a,size:r,mtime:i,...s};this.cache.set(a,n),this.hasChanged=!0;}};var g=class{constructor(e,t=["cid_sha1","cid_sha2-256"],i){this.targetHash=t;this.hashComputer=new m(this.targetHash,i),this.hashIndexManager=new p(e);}hashIndexManager;hashComputer;async computeMissingHash(e,t){await this.hashIndexManager.init();let i=await stat(e);if(this.hashIndexManager.getCache().has(I.basename(e))){let s=this.hashIndexManager.getCidForFile(e,i.size,i.mtime.toISOString());if(s)for(let r of this.targetHash)!t[r]&&s[r]&&(t[r]=s[r]);}await this.hashComputer.computeMissingHash(e,t),this.hashIndexManager.addFileCid(e,i.size,i.mtime.toISOString(),t);}}; +var h=import.meta.dirname;h=h.replace("src","dist");h=h+"/worker.js";var f=new URL(h).href;console.log("Worker URL: ",f);var l=class{piscina;constructor(e){this.piscina=new Piscina({maxThreads:4,filename:e||process.env.WORKER_URL||f});}async computeCIDs(e,t){return this.piscina.run({filePath:e,algorithms:t})}};var p=class{constructor(e,t){this.targetHash=e;this.fileIDComputer=new l(t);}fileIDComputer;async computeMissingHash(e,t){let r=this.targetHash.filter(i=>!t[i]);if(r.length===0)return;(await this.fileIDComputer.computeCIDs(e,r)).forEach((i,a)=>{let o=r[a];t[o]=i;});}};async function d(c){try{return await access(c,constants.F_OK),!0}catch{return !1}}var Q=["path","size","mtime"],m=class{constructor(e,t=["cid_sha1","cid_sha2-256"]){this.filePath=e;this.targetHash=t;if(!e)throw new Error("Invalid index file path")}cache=new Map;intervalId;intervalTime=3e4;lastIndexFileSize=0;lastCacheFile;indexOpsInProgress=!1;hasChanged=!1;initialLoad;getCache(){return new Map(this.cache)}async init(e=!0){return this.initialLoad||(this.initialLoad=new Promise(async(t,r)=>{try{if(!this.checkCSVHeaders(this.filePath))throw new Error("Invalid index file headers");await this.loadIndex(),e&&this.start(),t();}catch(s){r(s);}})),this.initialLoad}checkCSVHeaders(e){let t=parse(e,{bom:!0,columns:!0,skip_empty_lines:!0});if(t.length===0)return !0;let r=Object.keys(t[0]);return ["path","size","mtime",...this.targetHash].every(i=>r.includes(i))}start(){this.startAutoSave(this.intervalTime);}stopAutoSave(){this.intervalId&&clearInterval(this.intervalId);}startAutoSave(e){this.stopAutoSave(),this.intervalTime=e,this.intervalId=setInterval(()=>this.saveCacheToFile(),e);}async loadIndex(){if(await d(this.filePath)){let e=await promises.stat(this.filePath);if(this.lastIndexFileSize!==e.size){let t=await this.readCsv();for(let r of t)this.cache.set(r.path,r);return this.lastIndexFileSize=e.size,this.lastCacheFile=t,t}else return this.lastCacheFile}return []}loadIndexFromCache(){return Array.from(this.cache.values())}async readCsv(){if(!await d(this.filePath))return [];let e=performance.now(),t=parse$1({columns:!0,skip_empty_lines:!0}),r=[];return new Promise((s,i)=>{createReadStream(this.filePath).pipe(t).on("data",a=>{r.push(a);}).on("end",()=>{s(r),console.log(`Index read time ${performance.now()-e}ms`);}).on("error",a=>{i(a);});})}async saveCacheToFile(){if(this.indexOpsInProgress||!this.hasChanged)return;this.hasChanged=!1,this.indexOpsInProgress=!0;let e=performance.now(),t=await this.loadIndex();if(this.cache.size!==0){let i=this.loadIndexFromCache().filter(a=>!t.find(o=>o.path===a.path));if(i.length!==0){let a=stringify(i,{header:t.length===0,columns:[{key:"path",header:"path"},{key:"size",header:"size"},{key:"mtime",header:"mtime"},...this.targetHash.map(o=>({key:o,header:o}))]});await promises.appendFile(this.filePath,a);}}let r=performance.now()-e;console.log(`Index saved in ${r}ms`),r*10>this.intervalTime&&(this.startAutoSave(r*10),console.log(`Index save interval increased to ${r*10}ms`)),this.indexOpsInProgress=!1;}getCidForFile(e,t,r){let s=n__default.basename(e),i=this.cache.get(s);if(i){if(i.mtime){if(i.size===t+""&&i.mtime===r)return i}else if(i.size===t+"")return i}return null}addFileCid(e,t,r,s){if(!e||!t||!r||!s)throw new Error("Invalid parameters");let i=t+"",a=n__default.basename(e),o={path:a,size:i,mtime:r,...s};this.cache.set(a,o),this.hasChanged=!0;}};var v=class{constructor(e,t=["cid_sha1","cid_sha2-256"],r){this.targetHash=t;this.hashComputer=new p(this.targetHash,r),this.hashIndexManager=new m(e);}hashIndexManager;hashComputer;async computeMissingHash(e,t){await this.hashIndexManager.init();let r=await stat(e);if(this.hashIndexManager.getCache().has(n__default.basename(e))){let s=this.hashIndexManager.getCidForFile(e,r.size,r.mtime.toISOString());if(s)for(let i of this.targetHash)!t[i]&&s[i]&&(t[i]=s[i]);}await this.hashComputer.computeMissingHash(e,t),this.hashIndexManager.addFileCid(e,r.size,r.mtime.toISOString(),t);}};var y=class{constructor(e,t,r){this.WATCH_FOLDER_LIST=t;this.config=r;this.fileProcessor=e;}initialized=!1;queue=new b({concurrency:100,autoStart:!0});fileProcessor;queueSize=-1;current=0;processing=new Set;async processDirectory(e,t){let r=await readdir(e,{withFileTypes:!0});for(let s of r){let i=n.join(e,s.name);if(s.isDirectory())await this.processDirectory(i,t);else {let a=this.processFile(i);t.push(a);}}}async countFile(e){let t=await readdir(e,{withFileTypes:!0}),r=[];for(let i of t){let a=n.join(e,i.name);i.isDirectory()?r.push(this.countFile(a)):r.push(this.queue.add(async()=>await this.fileProcessor.canProcessFile(a)?1:0));}return (await Promise.all(r)).reduce((i,a)=>i+a,0)}async processFile(e){if(await this.fileProcessor.canProcessFile(e)&&!this.processing.has(e)){this.processing.add(e);let t=++this.current;await this.queue.add(()=>this.fileProcessor.processFile(t,this.queueSize,e)),this.processing.delete(e);}}async processFileExtended(e){let t=n.dirname(e);await this.processDirectory(t,[]);}chokidarWatch(e){let t={ignoreInitial:!0,persistent:!0,depth:1/0,awaitWriteFinish:{stabilityThreshold:this.config.stabilityThreshold||3e4,pollInterval:this.config.pollInterval||5e3}},r;(this.config.interval||0)<=0?r=w.watch(e,t):r=w.watch(e,{...t,usePolling:!0,interval:this.config.interval||0}),r.on("add",async s=>{try{await this.processFileExtended(s);}catch(i){console.error(`Error processing file ${s}:`,i);}}).on("change",async s=>{try{await this.processFileExtended(s);}catch(i){console.error(`Error processing file ${s}:`,i);}}).on("unlink",async s=>{try{await this.fileProcessor.deleteFile(s),await this.processFileExtended(s);}catch(i){console.error("Error processing file deletion:",i);}}).on("error",s=>console.error(`Watcher error: ${s}`)).on("ready",()=>{console.log(`Watching for file changes on ${e}`);});}async watch(){let e=this.WATCH_FOLDER_LIST;if(!e)throw new Error("No folder to watch");let t=e.split(",");this.chokidarWatch(t);let r=[];for(let a=0;aa+o,0);console.log(`Found ${s} files to process`),this.queueSize=s;let i=[];for(let a=0;a {\n return this.piscina.run({filePath, algorithms});\n }\n}\n","import { FileIDComputer } from \"./fileID/FileIDComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\n\nexport class HashComputer implements ComputeInterface{\n private fileIDComputer:FileIDComputer;\n\n constructor(private targetHash: CID_ALGORITHM_NAMES[],workerPath?:string) {\n this.fileIDComputer = new FileIDComputer(workerPath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n // Dynamically determine which hashes are needed\n const neededHashes = this.targetHash.filter(hashName => !metadata[hashName]);\n\n // If all hashes are already computed, skip the processing\n if (neededHashes.length === 0) {\n return;\n }\n\n // Compute only the needed CIDs\n const cids = await this.fileIDComputer.computeCIDs(filePath, neededHashes);\n\n // Map the computed CIDs back to their respective metadata properties\n cids.forEach((cid, index) => {\n const hashType = neededHashes[index];\n metadata[hashType] = cid;\n });\n }\n}\n","import {createReadStream, promises as fs} from 'fs';\nimport {parse} from 'csv-parse';\nimport {parse as parseSync} from 'csv-parse/sync';\nimport {stringify} from 'csv-stringify/sync';\nimport {existsAsync} from \"./ExistsAsync\";\nimport path from \"path\";\nimport {clearInterval} from \"node:timers\";\nimport {CID_ALGORITHM_NAMES} from \"./MultiHashData\";\n\ninterface IndexLine extends Partial> {\n path: string;\n size: string;\n mtime: string;\n}\nexport const INDEX_HEADERS = ['path', 'size', 'mtime'];\n\nexport class HashIndexManager {\n private cache: Map = new Map();\n private intervalId: NodeJS.Timeout;\n private intervalTime: number = 30000;\n private lastIndexFileSize: number = 0; //size of the index file last time it was read\n private lastCacheFile: IndexLine[]; //state of the file last time it was read\n private indexOpsInProgress: boolean = false;\n private hasChanged: boolean = false;\n private initialLoad;\n\n constructor(private filePath: string,private targetHash:CID_ALGORITHM_NAMES[]= [CID_ALGORITHM_NAMES.sha1,CID_ALGORITHM_NAMES.sha256]) {\n if (!filePath) {\n throw new Error('Invalid index file path');\n }\n }\n\n public getCache(): Map {\n return new Map(this.cache);\n }\n\n /**\n * After init consseutively calls to this method will not reload the index\n * @param autosave\n */\n async init(autosave = true) {\n if (!this.initialLoad) {\n this.initialLoad = new Promise(async (resolve, reject) => {\n try {\n if (!this.checkCSVHeaders(this.filePath)) {\n throw new Error('Invalid index file headers');\n }\n await this.loadIndex();\n if (autosave) {\n this.start();\n }\n resolve();\n }catch (e) {\n reject(e);\n }\n });\n }\n return this.initialLoad;\n }\n\n // Function to check CSV headers\n checkCSVHeaders (csvContent: string): boolean {\n const records = parseSync(csvContent, {\n bom: true,\n columns: true,\n skip_empty_lines: true,\n });\n\n if (records.length === 0) {\n //we will write the headers\n return true;\n }\n\n // Extract headers\n const headers = Object.keys(records[0]);\n\n // Define required headers\n //from the IndexLine interface\n const requiredHeaders = ['path', 'size', 'mtime', ...this.targetHash];\n\n // Check if all required headers are present\n return requiredHeaders.every(header => headers.includes(header));\n };\n\n public start() {\n this.startAutoSave(this.intervalTime);\n }\n\n public stopAutoSave() {\n if (this.intervalId) {\n clearInterval(this.intervalId);\n }\n }\n\n private startAutoSave(time: number) {\n this.stopAutoSave();\n this.intervalTime = time;\n this.intervalId = setInterval(() => this.saveCacheToFile(), time);\n }\n\n public async loadIndex(): Promise {\n if (await existsAsync(this.filePath)) {\n // check the file size and if it did not change, do not read the file\n const stats = await fs.stat(this.filePath);\n if (this.lastIndexFileSize !== stats.size) {\n // Read existing file content and parse it\n const records: IndexLine[] = await this.readCsv();\n for (const record of records) {\n this.cache.set(record.path, record);\n }\n this.lastIndexFileSize = stats.size;\n this.lastCacheFile = records;\n return records;\n } else {\n return this.lastCacheFile;\n }\n }\n return [];\n }\n\n private loadIndexFromCache(): IndexLine[] {\n return Array.from(this.cache.values());\n }\n\n private async readCsv(): Promise {\n if (!(await existsAsync(this.filePath))) {\n return [];\n }\n const start = performance.now();\n\n const parser = parse({\n columns: true,\n skip_empty_lines: true,\n });\n\n const records: IndexLine[] = [];\n\n return new Promise((resolve, reject) => {\n createReadStream(this.filePath)\n .pipe(parser)\n .on('data', (record) => {\n records.push(record);\n }).on('end', () => {\n resolve(records);\n console.log(`Index read time ${performance.now() - start}ms`);\n }).on('error', (err) => {\n reject(err);\n });\n });\n }\n\n private async saveCacheToFile(): Promise {\n if (this.indexOpsInProgress || !this.hasChanged) {\n return;\n }\n this.hasChanged = false;\n this.indexOpsInProgress = true;\n const start = performance.now();\n let existingRows: IndexLine[] = await this.loadIndex();\n\n if (this.cache.size !== 0) {\n const cacheRows: IndexLine[] = this.loadIndexFromCache();\n // Filter out cacheRows that are already in the file\n const newRows = cacheRows.filter(row => !existingRows.find(existingRow => existingRow.path === row.path));\n\n if (newRows.length !== 0) {\n // Serialize new cacheRows to CSV string\n const csvString = stringify(newRows, {\n header: existingRows.length === 0, // Only add header if the file was empty\n columns: [\n {key: 'path', header: 'path'},\n {key: 'size', header: 'size'},\n {key: 'mtime', header: 'mtime'},\n ...this.targetHash.map(hash => ({key: hash, header: hash})),\n ],\n });\n\n // Append new cacheRows to the file\n await fs.appendFile(this.filePath, csvString);\n }\n }\n const totalTime = performance.now() - start;\n console.log(`Index saved in ${totalTime}ms`);\n // Check if the time to save the index is greater than the interval time. increase the interval time if needed\n if (totalTime * 10 > this.intervalTime) {\n this.startAutoSave(totalTime * 10);\n console.log(`Index save interval increased to ${totalTime * 10}ms`);\n }\n this.indexOpsInProgress = false;\n }\n\n public getCidForFile(filePath: string, fileSize: number, mtime: string): IndexLine {\n const fileName = path.basename(filePath);\n let fileNameIndex = this.cache.get(fileName);\n if(fileNameIndex) {\n if(fileNameIndex.mtime) {\n //if we have a mtime, we need to check it\n if (fileNameIndex.size === (fileSize + \"\") && fileNameIndex.mtime === mtime) {\n return fileNameIndex;\n }\n }else {\n //mtime is optional\n if (fileNameIndex.size === (fileSize + \"\")) {\n return fileNameIndex;\n }\n }\n }\n\n // 3 - if not found, delete the entry (keeps the index clean)\n /*if (fileNameIndex && fileNameIndex.size !== (fileSize + \"\") && pathIndex && pathIndex.size !== (fileSize + \"\")) {\n this.cache.delete(fileName);\n this.cache.delete(filePath);\n }*/\n\n return null;\n }\n\n public addFileCid(filePath: string, fileSize: number, mtime: string, hashs: Partial>): void {\n if (!filePath || !fileSize || !mtime || !hashs) {\n throw new Error('Invalid parameters');\n }\n const size = fileSize + \"\";\n const baseName = path.basename(filePath);\n const data = {path: baseName, size: size, mtime: mtime, ...hashs};\n this.cache.set(baseName, data);\n this.hasChanged = true;\n }\n\n}","// Example of checking if a file exists asynchronously\n// Note: `existsSync` does not have a direct async equivalent, so we use `access` instead.\nimport {access,constants} from \"fs/promises\";\n\nexport async function existsAsync(filePath: string): Promise {\n try {\n await access(filePath, constants.F_OK);\n return true; // File exists\n } catch {\n return false; // File does not exist\n }\n}","import {HashComputer} from \"./HashComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\nimport {HashIndexManager} from \"./HashIndexManager\";\nimport {stat} from \"fs/promises\";\nimport path from \"path\";\n\nexport class ComputeHashIndexCache implements ComputeInterface {\n hashIndexManager: HashIndexManager;\n hashComputer: HashComputer;\n\n constructor(indexFilePath: string,private targetHash: CID_ALGORITHM_NAMES[] = [CID_ALGORITHM_NAMES.sha1, CID_ALGORITHM_NAMES.sha256],workerPath?:string) {\n this.hashComputer = new HashComputer(this.targetHash,workerPath);\n this.hashIndexManager = new HashIndexManager(indexFilePath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n await this.hashIndexManager.init();\n let stats = await stat(filePath);\n if (this.hashIndexManager.getCache().has(path.basename(filePath))) {\n const indexLine = this.hashIndexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());\n if(indexLine) {\n for (const hash of this.targetHash) {\n if (!metadata[hash] && indexLine[hash]) {\n metadata[hash] = indexLine[hash];\n }\n }\n }\n }\n await this.hashComputer.computeMissingHash(filePath, metadata);\n this.hashIndexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), metadata);\n }\n}"]} \ No newline at end of file +{"version":3,"sources":["../src/lib/fileID/FileIDComputer.ts","../src/lib/HashComputer.ts","../src/lib/HashIndexManager.ts","../src/lib/ExistsAsync.ts","../src/lib/ComputeHashIndexCache.ts","../src/lib/folder_watcher/FolderWatcher.ts"],"names":["Piscina","distFolder","workerUrl","FileIDComputer","workerPath","filePath","algorithms","HashComputer","targetHash","metadata","neededHashes","hashName","cid","index","hashType","createReadStream","fs","parse","parseSync","stringify","access","constants","existsAsync","path","clearInterval","INDEX_HEADERS","HashIndexManager","autosave","resolve","reject","e","csvContent","records","headers","header","time","stats","record","start","parser","err","existingRows","newRows","row","existingRow","csvString","hash","totalTime","fileSize","mtime","fileName","fileNameIndex","hashs","size","baseName","data","stat","ComputeHashIndexCache","indexFilePath","indexLine","chokidar","readdir","PQueue","FolderWatcher","fileProcessor","WATCH_FOLDER_LIST","config","directory","promises","entries","entry","fullPath","promise","countPromises","acc","current","dirname","folderList","chokidarconfig","watcher","error","folders","i","count"],"mappings":"sDAAA,OAAQ,WAAAA,MAAc,UAItB,IAAIC,EAAa,YAAY,QAC7BA,EAAaA,EAAW,QAAQ,MAAO,MAAM,EAC7CA,EAAaA,EAAa,aAC1B,IAAMC,EAAY,IAAI,IAAID,CAAU,EAAE,KACtC,QAAQ,IAAI,eAAgBC,CAAS,EAE9B,IAAMC,EAAN,KAAqB,CAChB,QAER,YAAYC,EAAoB,CAC5B,KAAK,QAAU,IAAIJ,EAAQ,CACvB,WAAY,EAEZ,SAAUI,GAAc,QAAQ,IAAI,YAAcF,CACtD,CAAC,CACL,CAQA,MAAa,YAAYG,EAAkBC,EAAsD,CAC7F,OAAO,KAAK,QAAQ,IAAI,CAAC,SAAAD,EAAU,WAAAC,CAAU,CAAC,CAClD,CACJ,EC3BO,IAAMC,EAAN,KAA8C,CAGjD,YAAoBC,EAAkCJ,EAAoB,CAAtD,gBAAAI,EAChB,KAAK,eAAiB,IAAIL,EAAeC,CAAU,CACvD,CAJQ,eAMR,MAAM,mBAAmBC,EAAkBI,EAAwC,CAE/E,IAAMC,EAAe,KAAK,WAAW,OAAOC,GAAY,CAACF,EAASE,CAAQ,CAAC,EAG3E,GAAID,EAAa,SAAW,EACxB,QAIS,MAAM,KAAK,eAAe,YAAYL,EAAUK,CAAY,GAGpE,QAAQ,CAACE,EAAKC,IAAU,CACzB,IAAMC,EAAWJ,EAAaG,CAAK,EACnCJ,EAASK,CAAQ,EAAIF,CACzB,CAAC,CACL,CACJ,EC5BA,OAAQ,oBAAAG,EAAkB,YAAYC,MAAS,KAC/C,OAAQ,SAAAC,MAAY,YACpB,OAAQ,SAASC,MAAgB,iBACjC,OAAQ,aAAAC,MAAgB,qBCDxB,OAAQ,UAAAC,EAAO,aAAAC,MAAgB,cAE/B,eAAsBC,EAAYjB,EAAoC,CAClE,GAAI,CACA,aAAMe,EAAOf,EAAUgB,EAAU,IAAI,EAC9B,EACX,MAAQ,CACJ,MAAO,EACX,CACJ,CDNA,OAAOE,MAAU,OACjB,OAAQ,iBAAAC,MAAoB,cAQrB,IAAMC,EAAgB,CAAC,OAAQ,OAAQ,OAAO,EAExCC,EAAN,KAAuB,CAU1B,YAAoBrB,EAAyBG,EAAkC,0BAAoD,EAAG,CAAlH,cAAAH,EAAyB,gBAAAG,EACzC,GAAI,CAACH,EACD,MAAM,IAAI,MAAM,yBAAyB,CAEjD,CAbQ,MAAgC,IAAI,IACpC,WACA,aAAuB,IACvB,kBAA4B,EAC5B,cACA,mBAA8B,GAC9B,WAAsB,GACtB,YAQD,UAAmC,CACtC,OAAO,IAAI,IAAI,KAAK,KAAK,CAC7B,CAMA,MAAM,KAAKsB,EAAW,GAAM,CACxB,OAAK,KAAK,cACN,KAAK,YAAc,IAAI,QAAc,MAAOC,EAASC,IAAW,CAC5D,GAAI,CACA,GAAI,CAAC,KAAK,gBAAgB,KAAK,QAAQ,EACnC,MAAM,IAAI,MAAM,4BAA4B,EAEhD,MAAM,KAAK,UAAU,EACjBF,GACA,KAAK,MAAM,EAEfC,EAAQ,CACZ,OAAQE,EAAG,CACPD,EAAOC,CAAC,CACZ,CACJ,CAAC,GAEE,KAAK,WAChB,CAGA,gBAAiBC,EAA6B,CAC1C,IAAMC,EAAUd,EAAUa,EAAY,CAClC,IAAK,GACL,QAAS,GACT,iBAAkB,EACtB,CAAC,EAED,GAAIC,EAAQ,SAAW,EAEnB,MAAO,GAIX,IAAMC,EAAU,OAAO,KAAKD,EAAQ,CAAC,CAAC,EAOtC,MAHwB,CAAC,OAAQ,OAAQ,QAAS,GAAG,KAAK,UAAU,EAG7C,MAAME,GAAUD,EAAQ,SAASC,CAAM,CAAC,CACnE,CAEO,OAAQ,CACX,KAAK,cAAc,KAAK,YAAY,CACxC,CAEO,cAAe,CACd,KAAK,YACLV,EAAc,KAAK,UAAU,CAErC,CAEQ,cAAcW,EAAc,CAChC,KAAK,aAAa,EAClB,KAAK,aAAeA,EACpB,KAAK,WAAa,YAAY,IAAM,KAAK,gBAAgB,EAAGA,CAAI,CACpE,CAEA,MAAa,WAAkC,CAC3C,GAAI,MAAMb,EAAY,KAAK,QAAQ,EAAG,CAElC,IAAMc,EAAQ,MAAMpB,EAAG,KAAK,KAAK,QAAQ,EACzC,GAAI,KAAK,oBAAsBoB,EAAM,KAAM,CAEvC,IAAMJ,EAAuB,MAAM,KAAK,QAAQ,EAChD,QAAWK,KAAUL,EACjB,KAAK,MAAM,IAAIK,EAAO,KAAMA,CAAM,EAEtC,YAAK,kBAAoBD,EAAM,KAC/B,KAAK,cAAgBJ,EACdA,CACX,KACI,QAAO,KAAK,aAEpB,CACA,MAAO,CAAC,CACZ,CAEQ,oBAAkC,CACtC,OAAO,MAAM,KAAK,KAAK,MAAM,OAAO,CAAC,CACzC,CAEA,MAAc,SAAgC,CAC1C,GAAI,CAAE,MAAMV,EAAY,KAAK,QAAQ,EACjC,MAAO,CAAC,EAEZ,IAAMgB,EAAQ,YAAY,IAAI,EAExBC,EAAStB,EAAM,CACjB,QAAS,GACT,iBAAkB,EACtB,CAAC,EAEKe,EAAuB,CAAC,EAE9B,OAAO,IAAI,QAAQ,CAACJ,EAASC,IAAW,CACpCd,EAAiB,KAAK,QAAQ,EACzB,KAAKwB,CAAM,EACX,GAAG,OAASF,GAAW,CACpBL,EAAQ,KAAKK,CAAM,CACvB,CAAC,EAAE,GAAG,MAAO,IAAM,CACnBT,EAAQI,CAAO,EACf,QAAQ,IAAI,mBAAmB,YAAY,IAAI,EAAIM,CAAK,IAAI,CAChE,CAAC,EAAE,GAAG,QAAUE,GAAQ,CACpBX,EAAOW,CAAG,CACd,CAAC,CACL,CAAC,CACL,CAEA,MAAc,iBAAiC,CAC3C,GAAI,KAAK,oBAAsB,CAAC,KAAK,WACjC,OAEJ,KAAK,WAAa,GAClB,KAAK,mBAAqB,GAC1B,IAAMF,EAAQ,YAAY,IAAI,EAC1BG,EAA4B,MAAM,KAAK,UAAU,EAErD,GAAI,KAAK,MAAM,OAAS,EAAG,CAGvB,IAAMC,EAFyB,KAAK,mBAAmB,EAE7B,OAAOC,GAAO,CAACF,EAAa,KAAKG,GAAeA,EAAY,OAASD,EAAI,IAAI,CAAC,EAExG,GAAID,EAAQ,SAAW,EAAG,CAEtB,IAAMG,EAAY1B,EAAUuB,EAAS,CACjC,OAAQD,EAAa,SAAW,EAChC,QAAS,CACL,CAAC,IAAK,OAAQ,OAAQ,MAAM,EAC5B,CAAC,IAAK,OAAQ,OAAQ,MAAM,EAC5B,CAAC,IAAK,QAAS,OAAQ,OAAO,EAC9B,GAAG,KAAK,WAAW,IAAIK,IAAS,CAAC,IAAKA,EAAM,OAAQA,CAAI,EAAE,CAC9D,CACJ,CAAC,EAGD,MAAM9B,EAAG,WAAW,KAAK,SAAU6B,CAAS,CAChD,CACJ,CACA,IAAME,EAAY,YAAY,IAAI,EAAIT,EACtC,QAAQ,IAAI,kBAAkBS,CAAS,IAAI,EAEvCA,EAAY,GAAK,KAAK,eACtB,KAAK,cAAcA,EAAY,EAAE,EACjC,QAAQ,IAAI,oCAAoCA,EAAY,EAAE,IAAI,GAEtE,KAAK,mBAAqB,EAC9B,CAEO,cAAc1C,EAAkB2C,EAAkBC,EAA0B,CAC/E,IAAMC,EAAW3B,EAAK,SAASlB,CAAQ,EACnC8C,EAAgB,KAAK,MAAM,IAAID,CAAQ,EAC3C,GAAGC,GACC,GAAGA,EAAc,OAEb,GAAIA,EAAc,OAAUH,EAAW,IAAOG,EAAc,QAAUF,EAClE,OAAOE,UAIPA,EAAc,OAAUH,EAAW,GACnC,OAAOG,EAWnB,OAAO,IACX,CAEO,WAAW9C,EAAkB2C,EAAkBC,EAAeG,EAA2D,CAC5H,GAAI,CAAC/C,GAAY,CAAC2C,GAAY,CAACC,GAAS,CAACG,EACrC,MAAM,IAAI,MAAM,oBAAoB,EAExC,IAAMC,EAAOL,EAAW,GAClBM,EAAW/B,EAAK,SAASlB,CAAQ,EACjCkD,EAAO,CAAC,KAAMD,EAAU,KAAMD,EAAM,MAAOJ,EAAO,GAAGG,CAAK,EAChE,KAAK,MAAM,IAAIE,EAAUC,CAAI,EAC7B,KAAK,WAAa,EACtB,CAEJ,EEjOA,OAAQ,QAAAC,MAAW,cACnB,OAAOjC,MAAU,OAEV,IAAMkC,EAAN,KAAwD,CAI3D,YAAYC,EAA8BlD,EAAoC,0BAAqD,EAAEJ,EAAoB,CAA/G,gBAAAI,EACtC,KAAK,aAAe,IAAID,EAAa,KAAK,WAAWH,CAAU,EAC/D,KAAK,iBAAmB,IAAIsB,EAAiBgC,CAAa,CAC9D,CANA,iBACA,aAOA,MAAM,mBAAmBrD,EAAkBI,EAAwC,CAC/E,MAAM,KAAK,iBAAiB,KAAK,EACjC,IAAI2B,EAAQ,MAAMoB,EAAKnD,CAAQ,EAC/B,GAAI,KAAK,iBAAiB,SAAS,EAAE,IAAIkB,EAAK,SAASlB,CAAQ,CAAC,EAAG,CAC/D,IAAMsD,EAAY,KAAK,iBAAiB,cAActD,EAAU+B,EAAM,KAAMA,EAAM,MAAM,YAAY,CAAC,EACrG,GAAGuB,EACC,QAAWb,KAAQ,KAAK,WAChB,CAACrC,EAASqC,CAAI,GAAKa,EAAUb,CAAI,IACjCrC,EAASqC,CAAI,EAAIa,EAAUb,CAAI,EAI/C,CACA,MAAM,KAAK,aAAa,mBAAmBzC,EAAUI,CAAQ,EAC7D,KAAK,iBAAiB,WAAWJ,EAAU+B,EAAM,KAAMA,EAAM,MAAM,YAAY,EAAG3B,CAAQ,CAC9F,CACJ,EC/BA,UAAYc,MAAU,OACtB,OAAOqC,MAAc,WACrB,OAAQ,WAAAC,MAAc,cACtB,OAAOC,MAAY,UAGZ,IAAMC,EAAN,KAAoB,CAQvB,YAAYC,EAA6CC,EAAmCC,EAIzF,CAJsD,uBAAAD,EAAmC,YAAAC,EAKxF,KAAK,cAAgBF,CACzB,CAbA,YAAc,GACd,MAAQ,IAAIF,EAAO,CAAC,YAAa,IAAK,UAAW,EAAI,CAAC,EACtD,cACA,UAAY,GACZ,QAAU,EACV,WAAa,IAAI,IAgBjB,MAAc,iBAAiBK,EAAmBC,EAA0B,CACxE,IAAMC,EAAU,MAAMR,EAAQM,EAAW,CAAC,cAAe,EAAI,CAAC,EAC9D,QAAWG,KAASD,EAAS,CACzB,IAAME,EAAgB,OAAKJ,EAAWG,EAAM,IAAI,EAChD,GAAIA,EAAM,YAAY,EAClB,MAAM,KAAK,iBAAiBC,EAAUH,CAAQ,MAC3C,CACH,IAAMI,EAAU,KAAK,YAAYD,CAAQ,EACzCH,EAAS,KAAKI,CAAO,CACzB,CACJ,CACJ,CAEA,MAAc,UAAUL,EAAoC,CACxD,IAAME,EAAU,MAAMR,EAAQM,EAAW,CAAC,cAAe,EAAI,CAAC,EACxDM,EAAgC,CAAC,EAEvC,QAAWH,KAASD,EAAS,CACzB,IAAME,EAAgB,OAAKJ,EAAWG,EAAM,IAAI,EAC5CA,EAAM,YAAY,EAElBG,EAAc,KAAK,KAAK,UAAUF,CAAQ,CAAC,EAG3CE,EAAc,KAAK,KAAK,MAAM,IAC1B,SAAa,MAAM,KAAK,cAAc,eAAeF,CAAQ,EAAK,EAAI,CAC1E,CAAC,CAET,CAIA,OADe,MAAM,QAAQ,IAAIE,CAAa,GAChC,OAAO,CAACC,EAAKC,IAAYD,EAAMC,EAAS,CAAC,CAC3D,CAEA,MAAc,YAAYtE,EAAgC,CACtD,GAAG,MAAM,KAAK,cAAc,eAAeA,CAAQ,GAC5C,CAAC,KAAK,WAAW,IAAIA,CAAQ,EAAG,CAC/B,KAAK,WAAW,IAAIA,CAAQ,EAC5B,IAAMsE,EAAU,EAAE,KAAK,QACvB,MAAM,KAAK,MAAM,IAAI,IAAM,KAAK,cAAc,YAAYA,EAAS,KAAK,UAAWtE,CAAQ,CAAC,EAC5F,KAAK,WAAW,OAAOA,CAAQ,CACnC,CAER,CAOA,MAAc,oBAAoBA,EAAgC,CAC9D,IAAIuE,EAAe,UAAQvE,CAAQ,EACnC,MAAM,KAAK,iBAAiBuE,EAAS,CAAC,CAAC,CAC3C,CAEQ,cAAcC,EAAqB,CACvC,IAAMC,EAAiB,CACnB,cAAe,GACf,WAAY,GACZ,MAAO,IACP,iBAAkB,CACd,mBAAoB,KAAK,OAAO,oBAAsB,IACtD,aAAc,KAAK,OAAO,cAAgB,GAC9C,CACJ,EAEIC,GACC,KAAK,OAAO,UAAY,IAAM,EAC/BA,EAAUnB,EAAS,MAAMiB,EAAYC,CAAc,EAEnDC,EAAUnB,EAAS,MAAMiB,EAAY,CACjC,GAAGC,EACH,WAAY,GACZ,SAAU,KAAK,OAAO,UAAY,CACtC,CAAC,EAGLC,EACK,GAAG,MAAO,MAAO1E,GAAa,CAC3B,GAAI,CACA,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAAS2E,EAAO,CACZ,QAAQ,MAAM,yBAAyB3E,CAAQ,IAAK2E,CAAK,CAC7D,CACJ,CAAC,EACA,GAAG,SAAU,MAAO3E,GAAa,CAC9B,GAAI,CACA,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAASyB,EAAG,CACR,QAAQ,MAAM,yBAAyBzB,CAAQ,IAAKyB,CAAC,CACzD,CACJ,CAAC,EACA,GAAG,SAAU,MAAOzB,GAAa,CAC9B,GAAI,CACA,MAAM,KAAK,cAAc,WAAWA,CAAQ,EAC5C,MAAM,KAAK,oBAAoBA,CAAQ,CAC3C,OAASyB,EAAG,CACR,QAAQ,MAAM,kCAAmCA,CAAC,CACtD,CACJ,CAAC,EACA,GAAG,QAAUkD,GAAU,QAAQ,MAAM,kBAAkBA,CAAK,EAAE,CAAC,EAC/D,GAAG,QAAS,IAAM,CACf,QAAQ,IAAI,gCAAgCH,CAAU,EAAE,CAC5D,CAAC,CACT,CAEA,MAAM,OAAQ,CACV,IAAII,EAAU,KAAK,kBACnB,GAAI,CAACA,EACD,MAAM,IAAI,MAAM,oBAAoB,EAExC,IAAIJ,EAAaI,EAAQ,MAAM,GAAG,EAElC,KAAK,cAAcJ,CAAU,EAG7B,IAAIJ,EAAgB,CAAC,EACrB,QAASS,EAAI,EAAGA,EAAIL,EAAW,OAAQK,IACnCL,EAAWK,CAAC,EAAS,YAAUL,EAAWK,CAAC,CAAC,EAC5CT,EAAc,KAAK,KAAK,UAAUI,EAAWK,CAAC,CAAC,CAAC,EAEpD,IAAMC,GAAS,MAAM,QAAQ,IAAIV,CAAa,GAAG,OAAO,CAACC,EAAKC,IAAYD,EAAMC,EAAS,CAAC,EAC1F,QAAQ,IAAI,SAASQ,CAAK,mBAAmB,EAC7C,KAAK,UAAYA,EACjB,IAAIf,EAA2B,CAAC,EAChC,QAASc,EAAI,EAAGA,EAAIL,EAAW,OAAQK,IACnCL,EAAWK,CAAC,EAAS,YAAUL,EAAWK,CAAC,CAAC,EAC5C,MAAM,KAAK,iBAAiBL,EAAWK,CAAC,EAAGd,CAAQ,EAEvD,MAAM,QAAQ,IAAIA,CAAQ,EAC1B,KAAK,YAAc,GAEnB,QAAQ,IAAI,eAAe,CAC/B,CACJ","sourcesContent":["import {Piscina} from \"piscina\";\nimport {CID_ALGORITHM_NAMES} from \"../MultiHashData\";\n\n// Construct the URL for the current module\nlet distFolder = import.meta.dirname;\ndistFolder = distFolder.replace('src', 'dist');\ndistFolder = distFolder + \"/worker.js\";\nconst workerUrl = new URL(distFolder).href;\nconsole.log(\"Worker URL: \", workerUrl);\n\nexport class FileIDComputer {\n private piscina: Piscina;\n\n constructor(workerPath?:string) {\n this.piscina = new Piscina({\n maxThreads: 4,\n //filename: new URL('./ShaComputeWorker.ts', import.meta.url).href\n filename: workerPath || process.env.WORKER_URL || workerUrl\n });\n }\n\n /**\n * Compute the CIDs of a file using specific algorithms\n * @param filePath The path to the file\n * @param algorithms Array of algorithms ('sha256', 'sha1')\n * @returns Array of CIDs (in the order of the algorithms)\n */\n public async computeCIDs(filePath: string, algorithms: CID_ALGORITHM_NAMES[]): Promise {\n return this.piscina.run({filePath, algorithms});\n }\n}\n","import { FileIDComputer } from \"./fileID/FileIDComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\n\nexport class HashComputer implements ComputeInterface{\n private fileIDComputer:FileIDComputer;\n\n constructor(private targetHash: CID_ALGORITHM_NAMES[],workerPath?:string) {\n this.fileIDComputer = new FileIDComputer(workerPath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n // Dynamically determine which hashes are needed\n const neededHashes = this.targetHash.filter(hashName => !metadata[hashName]);\n\n // If all hashes are already computed, skip the processing\n if (neededHashes.length === 0) {\n return;\n }\n\n // Compute only the needed CIDs\n const cids = await this.fileIDComputer.computeCIDs(filePath, neededHashes);\n\n // Map the computed CIDs back to their respective metadata properties\n cids.forEach((cid, index) => {\n const hashType = neededHashes[index];\n metadata[hashType] = cid;\n });\n }\n}\n","import {createReadStream, promises as fs} from 'fs';\nimport {parse} from 'csv-parse';\nimport {parse as parseSync} from 'csv-parse/sync';\nimport {stringify} from 'csv-stringify/sync';\nimport {existsAsync} from \"./ExistsAsync\";\nimport path from \"path\";\nimport {clearInterval} from \"node:timers\";\nimport {CID_ALGORITHM_NAMES} from \"./MultiHashData\";\n\ninterface IndexLine extends Partial> {\n path: string;\n size: string;\n mtime: string;\n}\nexport const INDEX_HEADERS = ['path', 'size', 'mtime'];\n\nexport class HashIndexManager {\n private cache: Map = new Map();\n private intervalId: NodeJS.Timeout;\n private intervalTime: number = 30000;\n private lastIndexFileSize: number = 0; //size of the index file last time it was read\n private lastCacheFile: IndexLine[]; //state of the file last time it was read\n private indexOpsInProgress: boolean = false;\n private hasChanged: boolean = false;\n private initialLoad;\n\n constructor(private filePath: string,private targetHash:CID_ALGORITHM_NAMES[]= [CID_ALGORITHM_NAMES.sha1,CID_ALGORITHM_NAMES.sha256]) {\n if (!filePath) {\n throw new Error('Invalid index file path');\n }\n }\n\n public getCache(): Map {\n return new Map(this.cache);\n }\n\n /**\n * After init consseutively calls to this method will not reload the index\n * @param autosave\n */\n async init(autosave = true) {\n if (!this.initialLoad) {\n this.initialLoad = new Promise(async (resolve, reject) => {\n try {\n if (!this.checkCSVHeaders(this.filePath)) {\n throw new Error('Invalid index file headers');\n }\n await this.loadIndex();\n if (autosave) {\n this.start();\n }\n resolve();\n }catch (e) {\n reject(e);\n }\n });\n }\n return this.initialLoad;\n }\n\n // Function to check CSV headers\n checkCSVHeaders (csvContent: string): boolean {\n const records = parseSync(csvContent, {\n bom: true,\n columns: true,\n skip_empty_lines: true,\n });\n\n if (records.length === 0) {\n //we will write the headers\n return true;\n }\n\n // Extract headers\n const headers = Object.keys(records[0]);\n\n // Define required headers\n //from the IndexLine interface\n const requiredHeaders = ['path', 'size', 'mtime', ...this.targetHash];\n\n // Check if all required headers are present\n return requiredHeaders.every(header => headers.includes(header));\n };\n\n public start() {\n this.startAutoSave(this.intervalTime);\n }\n\n public stopAutoSave() {\n if (this.intervalId) {\n clearInterval(this.intervalId);\n }\n }\n\n private startAutoSave(time: number) {\n this.stopAutoSave();\n this.intervalTime = time;\n this.intervalId = setInterval(() => this.saveCacheToFile(), time);\n }\n\n public async loadIndex(): Promise {\n if (await existsAsync(this.filePath)) {\n // check the file size and if it did not change, do not read the file\n const stats = await fs.stat(this.filePath);\n if (this.lastIndexFileSize !== stats.size) {\n // Read existing file content and parse it\n const records: IndexLine[] = await this.readCsv();\n for (const record of records) {\n this.cache.set(record.path, record);\n }\n this.lastIndexFileSize = stats.size;\n this.lastCacheFile = records;\n return records;\n } else {\n return this.lastCacheFile;\n }\n }\n return [];\n }\n\n private loadIndexFromCache(): IndexLine[] {\n return Array.from(this.cache.values());\n }\n\n private async readCsv(): Promise {\n if (!(await existsAsync(this.filePath))) {\n return [];\n }\n const start = performance.now();\n\n const parser = parse({\n columns: true,\n skip_empty_lines: true,\n });\n\n const records: IndexLine[] = [];\n\n return new Promise((resolve, reject) => {\n createReadStream(this.filePath)\n .pipe(parser)\n .on('data', (record) => {\n records.push(record);\n }).on('end', () => {\n resolve(records);\n console.log(`Index read time ${performance.now() - start}ms`);\n }).on('error', (err) => {\n reject(err);\n });\n });\n }\n\n private async saveCacheToFile(): Promise {\n if (this.indexOpsInProgress || !this.hasChanged) {\n return;\n }\n this.hasChanged = false;\n this.indexOpsInProgress = true;\n const start = performance.now();\n let existingRows: IndexLine[] = await this.loadIndex();\n\n if (this.cache.size !== 0) {\n const cacheRows: IndexLine[] = this.loadIndexFromCache();\n // Filter out cacheRows that are already in the file\n const newRows = cacheRows.filter(row => !existingRows.find(existingRow => existingRow.path === row.path));\n\n if (newRows.length !== 0) {\n // Serialize new cacheRows to CSV string\n const csvString = stringify(newRows, {\n header: existingRows.length === 0, // Only add header if the file was empty\n columns: [\n {key: 'path', header: 'path'},\n {key: 'size', header: 'size'},\n {key: 'mtime', header: 'mtime'},\n ...this.targetHash.map(hash => ({key: hash, header: hash})),\n ],\n });\n\n // Append new cacheRows to the file\n await fs.appendFile(this.filePath, csvString);\n }\n }\n const totalTime = performance.now() - start;\n console.log(`Index saved in ${totalTime}ms`);\n // Check if the time to save the index is greater than the interval time. increase the interval time if needed\n if (totalTime * 10 > this.intervalTime) {\n this.startAutoSave(totalTime * 10);\n console.log(`Index save interval increased to ${totalTime * 10}ms`);\n }\n this.indexOpsInProgress = false;\n }\n\n public getCidForFile(filePath: string, fileSize: number, mtime: string): IndexLine {\n const fileName = path.basename(filePath);\n let fileNameIndex = this.cache.get(fileName);\n if(fileNameIndex) {\n if(fileNameIndex.mtime) {\n //if we have a mtime, we need to check it\n if (fileNameIndex.size === (fileSize + \"\") && fileNameIndex.mtime === mtime) {\n return fileNameIndex;\n }\n }else {\n //mtime is optional\n if (fileNameIndex.size === (fileSize + \"\")) {\n return fileNameIndex;\n }\n }\n }\n\n // 3 - if not found, delete the entry (keeps the index clean)\n /*if (fileNameIndex && fileNameIndex.size !== (fileSize + \"\") && pathIndex && pathIndex.size !== (fileSize + \"\")) {\n this.cache.delete(fileName);\n this.cache.delete(filePath);\n }*/\n\n return null;\n }\n\n public addFileCid(filePath: string, fileSize: number, mtime: string, hashs: Partial>): void {\n if (!filePath || !fileSize || !mtime || !hashs) {\n throw new Error('Invalid parameters');\n }\n const size = fileSize + \"\";\n const baseName = path.basename(filePath);\n const data = {path: baseName, size: size, mtime: mtime, ...hashs};\n this.cache.set(baseName, data);\n this.hasChanged = true;\n }\n\n}","// Example of checking if a file exists asynchronously\n// Note: `existsSync` does not have a direct async equivalent, so we use `access` instead.\nimport {access,constants} from \"fs/promises\";\n\nexport async function existsAsync(filePath: string): Promise {\n try {\n await access(filePath, constants.F_OK);\n return true; // File exists\n } catch {\n return false; // File does not exist\n }\n}","import {HashComputer} from \"./HashComputer\";\nimport {CID_ALGORITHM_NAMES, ComputeInterface, MultiHashData} from \"./MultiHashData\";\nimport {HashIndexManager} from \"./HashIndexManager\";\nimport {stat} from \"fs/promises\";\nimport path from \"path\";\n\nexport class ComputeHashIndexCache implements ComputeInterface {\n hashIndexManager: HashIndexManager;\n hashComputer: HashComputer;\n\n constructor(indexFilePath: string,private targetHash: CID_ALGORITHM_NAMES[] = [CID_ALGORITHM_NAMES.sha1, CID_ALGORITHM_NAMES.sha256],workerPath?:string) {\n this.hashComputer = new HashComputer(this.targetHash,workerPath);\n this.hashIndexManager = new HashIndexManager(indexFilePath);\n }\n\n async computeMissingHash(filePath: string, metadata: MultiHashData): Promise {\n await this.hashIndexManager.init();\n let stats = await stat(filePath);\n if (this.hashIndexManager.getCache().has(path.basename(filePath))) {\n const indexLine = this.hashIndexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString());\n if(indexLine) {\n for (const hash of this.targetHash) {\n if (!metadata[hash] && indexLine[hash]) {\n metadata[hash] = indexLine[hash];\n }\n }\n }\n }\n await this.hashComputer.computeMissingHash(filePath, metadata);\n this.hashIndexManager.addFileCid(filePath, stats.size, stats.mtime.toISOString(), metadata);\n }\n}","import * as path from 'path';\nimport chokidar from 'chokidar';\nimport {readdir} from \"fs/promises\";\nimport PQueue from \"p-queue\";\nimport {FileProcessorInterface} from \"./FileProcessorInterface.js\";\n\nexport class FolderWatcher {\n initialized = false;\n queue = new PQueue({concurrency: 100, autoStart: true});\n fileProcessor: FileProcessorInterface;\n queueSize = -1;\n current = 0;\n processing = new Set();\n\n constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{\n interval?: number,\n stabilityThreshold?: number,\n pollInterval?: number,\n }) {\n this.fileProcessor = fileProcessor;\n }\n\n /**\n * Process the directory and its subdirectories\n * @param directory\n * @param promises\n * @private\n */\n private async processDirectory(directory: string, promises: Promise[]) {\n const entries = await readdir(directory, {withFileTypes: true});\n for (const entry of entries) {\n const fullPath = path.join(directory, entry.name);\n if (entry.isDirectory()) {\n await this.processDirectory(fullPath, promises);\n } else {\n const promise = this.processFile(fullPath);\n promises.push(promise);\n }\n }\n }\n\n private async countFile(directory: string): Promise {\n const entries = await readdir(directory, {withFileTypes: true});\n const countPromises: Promise[] = [];\n\n for (const entry of entries) {\n const fullPath = path.join(directory, entry.name);\n if (entry.isDirectory()) {\n // Push the promise without waiting for it here\n countPromises.push(this.countFile(fullPath));\n } else {\n // Use an IIFE (Immediately Invoked Function Expression) to handle asynchronous condition\n countPromises.push(this.queue.add(\n async () => (await this.fileProcessor.canProcessFile(fullPath)) ? 1 : 0\n ));\n }\n }\n\n // Wait for all promises to resolve and then sum up their results\n const counts = await Promise.all(countPromises);\n return counts.reduce((acc, current) => acc + current, 0);\n }\n\n private async processFile(filePath: string):Promise {\n if(await this.fileProcessor.canProcessFile(filePath)) {\n if(!this.processing.has(filePath)) {\n this.processing.add(filePath);//avoid double processing in the queue\n const current = ++this.current;\n await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath));\n this.processing.delete(filePath);\n }\n }\n }\n\n /**\n * Process the file and its sibling file included other folder it is a recursive function\n * @param filePath\n * @private\n */\n private async processFileExtended(filePath: string):Promise {\n let dirname = path.dirname(filePath);\n await this.processDirectory(dirname, []);\n }\n\n private chokidarWatch(folderList: string[]){\n const chokidarconfig = {\n ignoreInitial: true,//ignore the initial scan we have our own function\n persistent: true,\n depth: Infinity,\n awaitWriteFinish: {\n stabilityThreshold: this.config.stabilityThreshold || 30000,\n pollInterval: this.config.pollInterval || 5000\n }\n };\n\n let watcher: chokidar.FSWatcher;\n if ((this.config.interval || 0) <= 0) {\n watcher = chokidar.watch(folderList, chokidarconfig);\n } else {\n watcher = chokidar.watch(folderList, {\n ...chokidarconfig,\n usePolling: true,\n interval: this.config.interval || 0\n });\n }\n\n watcher\n .on('add', async (filePath) => {\n try {\n await this.processFileExtended(filePath);\n } catch (error) {\n console.error(`Error processing file ${filePath}:`, error);\n }\n })\n .on('change', async (filePath) => {\n try {\n await this.processFileExtended(filePath);\n } catch (e) {\n console.error(`Error processing file ${filePath}:`, e);\n }\n })\n .on('unlink', async (filePath) => {\n try {\n await this.fileProcessor.deleteFile(filePath);\n await this.processFileExtended(filePath);\n } catch (e) {\n console.error(`Error processing file deletion:`, e);\n }\n })\n .on('error', (error) => console.error(`Watcher error: ${error}`))\n .on('ready', () => {\n console.log(`Watching for file changes on ${folderList}`);\n });\n }\n\n async watch() {\n let folders = this.WATCH_FOLDER_LIST;\n if (!folders) {\n throw new Error(`No folder to watch`);\n }\n let folderList = folders.split(',');\n\n this.chokidarWatch(folderList);\n\n //start the initial scan\n let countPromises = [];\n for (let i = 0; i < folderList.length; i++) {\n folderList[i] = path.normalize(folderList[i]);\n countPromises.push(this.countFile(folderList[i]));\n }\n const count = (await Promise.all(countPromises)).reduce((acc, current) => acc + current, 0);\n console.log(`Found ${count} files to process`);\n this.queueSize = count;\n let promises: Promise[] = [];\n for (let i = 0; i < folderList.length; i++) {\n folderList[i] = path.normalize(folderList[i]);\n await this.processDirectory(folderList[i], promises);\n }\n await Promise.all(promises);\n this.initialized = true;\n\n console.log(`Watcher ready`);\n }\n}\n"]} \ No newline at end of file diff --git a/package.json b/package.json index 8e62f2a..42a70b0 100644 --- a/package.json +++ b/package.json @@ -40,9 +40,11 @@ }, "packageManager": "yarn@4.1.1", "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" } } diff --git a/src/lib/folder_watcher/FileProcessorInterface.ts b/src/lib/folder_watcher/FileProcessorInterface.ts new file mode 100644 index 0000000..4d5e654 --- /dev/null +++ b/src/lib/folder_watcher/FileProcessorInterface.ts @@ -0,0 +1,7 @@ +export interface FileProcessorInterface { + processFile(current: number, queueSize: number, nfoFilePath: string): Promise; + + canProcessFile(filePath: string): Promise; + + deleteFile(filePath: string): Promise +} \ No newline at end of file diff --git a/src/lib/folder_watcher/FolderWatcher.ts b/src/lib/folder_watcher/FolderWatcher.ts new file mode 100644 index 0000000..8857504 --- /dev/null +++ b/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(); + + 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[]) { + 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 { + const entries = await readdir(directory, {withFileTypes: true}); + const countPromises: Promise[] = []; + + 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 { + 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 { + 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[] = []; + 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`); + } +} diff --git a/src/lib/index.ts b/src/lib/index.ts index 087e561..c6a79dc 100644 --- a/src/lib/index.ts +++ b/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'; \ No newline at end of file diff --git a/yarn.lock b/yarn.lock index cc4a1a5..54f4ca8 100644 --- a/yarn.lock +++ b/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"