From c414a4a3035aaf10ccfe6833d973aaf3f765235b Mon Sep 17 00:00:00 2001 From: Sucukdeluxe <259325684+Sucukdeluxe@users.noreply.github.com> Date: Wed, 26 Aug 2026 09:10:44 +0200 Subject: [PATCH] feat: reconcile monitored folders --- lib/folder-monitor.js | 232 ++++++++++++++++++++++++++++++----- tests/folder-monitor.test.js | 206 ++++++++++++++++++++++++++++++- 2 files changed, 398 insertions(+), 40 deletions(-) diff --git a/lib/folder-monitor.js b/lib/folder-monitor.js index 387a8d5..8716a5e 100644 --- a/lib/folder-monitor.js +++ b/lib/folder-monitor.js @@ -1,17 +1,47 @@ const { EventEmitter } = require('events'); +const fs = require('fs'); const path = require('path'); const chokidar = require('chokidar'); +const { walkFolderAsync } = require('./file-discovery'); class FolderMonitor extends EventEmitter { - constructor({ watch = chokidar.watch } = {}) { + constructor({ + watch = chokidar.watch, + walkFolder = walkFolderAsync, + access = fs.promises.access, + stat = fs.promises.stat, + setTimeoutFn = setTimeout, + clearTimeoutFn = clearTimeout, + setIntervalFn = setInterval, + clearIntervalFn = clearInterval, + now = Date.now + } = {}) { super(); this._watch = watch; + this._walkFolder = walkFolder; + this._access = access; + this._stat = stat; + this._setTimeout = setTimeoutFn; + this._clearTimeout = clearTimeoutFn; + this._setInterval = setIntervalFn; + this._clearInterval = clearIntervalFn; + this._now = now; this._watcher = null; this._settings = null; this._seenFiles = new Set(); this._batchBuffer = []; this._batchTimer = null; this._initialScopes = new Set(); + this._reconcileTimer = null; + this._scanPromise = null; + this._followUpRequested = false; + this._followUpOptions = null; + this._paused = false; + this._reachable = null; + this._scanning = false; + this._lastScanAt = null; + this._lastScanTrigger = ''; + this._lastError = ''; } get running() { @@ -21,6 +51,9 @@ class FolderMonitor extends EventEmitter { start(settings) { this.stop(); this._settings = settings; + this._paused = false; + this._reachable = null; + this._lastError = ''; const folderPath = String(settings.folderPath || '').trim(); if (!folderPath) throw new Error('Kein Ordnerpfad angegeben'); @@ -50,62 +83,96 @@ class FolderMonitor extends EventEmitter { } this._watcher.on('add', (filePath) => this._onNewFile(filePath)); this._watcher.on('unlink', (filePath) => { - // Allow re-added files (e.g. re-encoded) to be detected again - const normalized = filePath.replace(/\\/g, '/').toLowerCase(); - this._seenFiles.delete(normalized); + this._seenFiles.delete(this._normalizePath(filePath)); }); - this._watcher.on('error', (err) => this.emit('error', err)); + this._watcher.on('error', (err) => { + this._lastError = err instanceof Error ? err.message : String(err); + this._emitStatus(); + this.emit('error', err); + }); + const intervalMinutes = Number(settings.reconcileIntervalMinutes) || 5; + this._reconcileTimer = this._setInterval(() => this._reconcile(), intervalMinutes * 60 * 1000); + this._emitStatus(); return { includesExisting: includeInitial }; } stop() { + const changed = !!(this._watcher || this._reconcileTimer || this._batchTimer || this._batchBuffer.length || this._seenFiles.size); if (this._watcher) { this._watcher.close().catch(() => {}); this._watcher = null; } + if (this._reconcileTimer) { + this._clearInterval(this._reconcileTimer); + this._reconcileTimer = null; + } if (this._batchTimer) { - clearTimeout(this._batchTimer); + this._clearTimeout(this._batchTimer); this._batchTimer = null; } this._batchBuffer = []; this._seenFiles = new Set(); + if (changed) this._emitStatus(); } status() { - return { + return Object.freeze({ running: this.running, + paused: this._paused, + reachable: this._reachable, + scanning: this._scanning, folderPath: this._settings ? this._settings.folderPath : '', - seenCount: this._seenFiles.size - }; + seenCount: this._seenFiles.size, + lastScanAt: this._lastScanAt, + lastScanTrigger: this._lastScanTrigger, + error: this._lastError + }); + } + + scan({ emitFiles = true, trigger = 'manual' } = {}) { + const request = { emitFiles: !!emitFiles, trigger: String(trigger || 'manual') }; + if (this._scanPromise) { + this._followUpRequested = true; + this._followUpOptions = { + emitFiles: !!(this._followUpOptions?.emitFiles || request.emitFiles), + trigger: request.trigger + }; + return this._scanPromise; + } + this._scanPromise = this._runScans(request).finally(() => { + this._scanPromise = null; + }); + return this._scanPromise; + } + + async pause() { + this._paused = true; + if (this._reconcileTimer) { + this._clearInterval(this._reconcileTimer); + this._reconcileTimer = null; + } + const watcher = this._watcher; + this._watcher = null; + if (watcher) await watcher.close().catch(() => {}); + if (this._batchTimer) { + this._clearTimeout(this._batchTimer); + this._batchTimer = null; + } + this._batchBuffer = []; + this._emitStatus(); + } + + async resume(settings = this._settings) { + if (!settings) throw new Error('Keine Ordnerkonfiguration vorhanden'); + this.start(settings); + return this.scan({ emitFiles: true, trigger: 'resume' }); } _onNewFile(filePath) { - const settings = this._settings; - if (!settings) return; - - // Extension filter - const ext = path.extname(filePath).replace(/^\./, '').toLowerCase(); - const rawExtensions = String(settings.extensions || '').trim(); - if (rawExtensions) { - const extList = rawExtensions.split(',').map(e => e.trim().toLowerCase().replace(/^\./, '')).filter(Boolean); - if (extList.length > 0) { - const matches = extList.includes(ext); - if (settings.filterMode === 'include' && !matches) return; - if (settings.filterMode === 'exclude' && matches) return; - } - } - - // Skip duplicates (session-based) - if (settings.skipDuplicates) { - const normalized = filePath.replace(/\\/g, '/').toLowerCase(); - if (this._seenFiles.has(normalized)) return; - this._seenFiles.add(normalized); - } - - // Batch: collect files over 200ms window then emit together + if (!this._settings || !this._classifyPath(filePath).allowed || !this._acceptPath(filePath)) return; this._batchBuffer.push(filePath); - if (this._batchTimer) clearTimeout(this._batchTimer); - this._batchTimer = setTimeout(() => { + if (this._batchTimer) this._clearTimeout(this._batchTimer); + this._batchTimer = this._setTimeout(() => { const files = this._batchBuffer.splice(0); this._batchTimer = null; if (files.length > 0) { @@ -113,6 +180,103 @@ class FolderMonitor extends EventEmitter { } }, 200); } + + async _runScans(initialRequest) { + let request = initialRequest; + let result; + do { + this._followUpRequested = false; + this._followUpOptions = null; + result = await this._performScan(request); + request = this._followUpOptions || request; + } while (this._followUpRequested); + return result; + } + + async _performScan({ emitFiles, trigger }) { + if (!this._settings) throw new Error('Keine Ordnerkonfiguration vorhanden'); + this._scanning = true; + this._lastScanTrigger = trigger; + this._emitStatus(); + const wasReachable = this._reachable; + try { + await this._access(this._settings.folderPath); + } catch (error) { + this._reachable = false; + this._lastScanAt = this._now(); + this._lastError = error instanceof Error ? error.message : String(error); + this._scanning = false; + this._emitStatus(); + return Object.freeze({ files: [], reachable: false, reconnected: false, trigger }); + } + + const discovered = await this._walkFolder(this._settings.folderPath, { recursive: !!this._settings.recursive }); + const files = []; + for (const descriptor of discovered) { + if (!this._settings.recursive && this._isNestedPath(descriptor.path)) continue; + if (!this._classifyPath(descriptor.path).allowed) continue; + let mtimeMs = 0; + try { + mtimeMs = Number((await this._stat(descriptor.path)).mtimeMs) || 0; + } catch {} + files.push(Object.freeze({ + path: descriptor.path, + name: descriptor.name || path.basename(descriptor.path), + size: Number(descriptor.size) || 0, + mtimeMs + })); + } + const emittedFiles = emitFiles ? files.filter((file) => this._acceptPath(file.path)) : []; + if (emittedFiles.length > 0) this.emit('new-files', emittedFiles.map((file) => file.path)); + this._reachable = true; + this._lastScanAt = this._now(); + this._lastError = ''; + this._scanning = false; + this._emitStatus(); + return Object.freeze({ files: Object.freeze(files), reachable: true, reconnected: wasReachable === false, trigger }); + } + + async _reconcile() { + if (this._paused || !this._settings) return; + const trigger = this._reachable === false ? 'reconnect' : 'interval'; + await this.scan({ emitFiles: true, trigger }); + } + + _extensionSet() { + return new Set(String(this._settings?.extensions || '') + .split(',') + .map((extension) => extension.trim().toLowerCase().replace(/^\./, '')) + .filter(Boolean)); + } + + _classifyPath(filePath) { + const extension = path.extname(filePath).replace(/^\./, '').toLowerCase(); + const extensions = this._extensionSet(); + const matches = extensions.size === 0 || extensions.has(extension); + const allowed = this._settings?.filterMode === 'exclude' ? !matches : matches; + return Object.freeze({ allowed, reason: allowed ? 'matched' : 'extension' }); + } + + _acceptPath(filePath) { + if (!this._settings?.skipDuplicates) return true; + const normalized = this._normalizePath(filePath); + if (this._seenFiles.has(normalized)) return false; + this._seenFiles.add(normalized); + return true; + } + + _normalizePath(filePath) { + return String(filePath).replace(/\\/g, '/').toLowerCase(); + } + + _isNestedPath(filePath) { + const relativePath = path.relative(this._settings.folderPath, filePath); + return relativePath.split(/[\\/]/).length > 1; + } + + _emitStatus() { + this.emit('status', this.status()); + } } module.exports = FolderMonitor; diff --git a/tests/folder-monitor.test.js b/tests/folder-monitor.test.js index abd4667..b674ec6 100644 --- a/tests/folder-monitor.test.js +++ b/tests/folder-monitor.test.js @@ -1,21 +1,121 @@ const test = require('node:test'); const assert = require('node:assert/strict'); const { EventEmitter } = require('node:events'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); const FolderMonitor = require('../lib/folder-monitor'); -function createHarness() { +function createWatcherHarness() { const calls = []; + const timers = createManualTimers(); const watch = (folderPath, options) => { const watcher = new EventEmitter(); watcher.close = async () => {}; calls.push({ folderPath, options, watcher }); return watcher; }; - return { calls, monitor: new FolderMonitor({ watch }) }; + return { calls, monitor: new FolderMonitor({ watch, ...timers }) }; +} + +function createManualTimers() { + const intervals = new Set(); + const timeouts = new Set(); + return { + setIntervalFn(callback) { + intervals.add(callback); + return callback; + }, + clearIntervalFn(callback) { + intervals.delete(callback); + }, + setTimeoutFn(callback) { + timeouts.add(callback); + return callback; + }, + clearTimeoutFn(callback) { + timeouts.delete(callback); + }, + async runInterval() { + for (const callback of [...intervals]) await callback(); + } + }; +} + +function createSilentWatch() { + return () => { + const watcher = new EventEmitter(); + watcher.close = async () => {}; + return watcher; + }; +} + +function createScanHarness({ files = [] } = {}) { + const events = { newFiles: [], statuses: [] }; + const timers = createManualTimers(); + const stats = new Map(files.map((file) => [file.path, { mtimeMs: file.mtimeMs }])); + const monitor = new FolderMonitor({ + watch: createSilentWatch(), + walkFolder: async () => files.map(({ path: filePath, name, size }) => ({ path: filePath, name, size })), + access: async () => {}, + stat: async (filePath) => stats.get(filePath), + now: () => 1234, + ...timers + }); + monitor.on('new-files', (paths) => events.newFiles.push(paths)); + monitor.on('status', (status) => events.statuses.push(status)); + return { monitor, events, runInterval: timers.runInterval }; +} + +function createDeferredScanHarness() { + let calls = 0; + let releaseFirstScan; + const timers = createManualTimers(); + const firstScan = new Promise((resolve) => { releaseFirstScan = resolve; }); + const monitor = new FolderMonitor({ + watch: createSilentWatch(), + access: async () => {}, + stat: async () => ({ mtimeMs: 1 }), + walkFolder: async () => { + calls++; + if (calls === 1) await firstScan; + return []; + }, + ...timers + }); + monitor.start({ folderPath: 'C:\\incoming', reconcileIntervalMinutes: 5 }); + return { monitor, releaseFirstScan, scanCalls: () => calls }; +} + +function createReachabilityHarness(initiallyReachable) { + let reachable = initiallyReachable; + let calls = 0; + const statusEvents = []; + const timers = createManualTimers(); + const monitor = new FolderMonitor({ + watch: createSilentWatch(), + access: async () => { + if (!reachable) throw new Error('unreachable'); + }, + walkFolder: async () => { + calls++; + return []; + }, + stat: async () => ({ mtimeMs: 1 }), + ...timers + }); + monitor.on('status', (status) => statusEvents.push(status)); + return { + monitor, + setReachable(value) { reachable = value; }, + statusEvents, + scanCalls: () => calls, + runInterval: timers.runInterval + }; } test('existing files are included only on the first start of the same watch scope', () => { - const { calls, monitor } = createHarness(); + const { calls, monitor } = createWatcherHarness(); const settings = { folderPath: 'C:\\incoming', includeExisting: true, recursive: false }; monitor.start(settings); monitor.start(settings); @@ -24,7 +124,7 @@ test('existing files are included only on the first start of the same watch scop }); test('existing files remain ignored unless the option is enabled', () => { - const { calls, monitor } = createHarness(); + const { calls, monitor } = createWatcherHarness(); monitor.start({ folderPath: 'C:\\incoming', includeExisting: false, recursive: false }); monitor.start({ folderPath: 'C:\\incoming', includeExisting: true, recursive: false }); assert.equal(calls[0].options.ignoreInitial, true); @@ -32,7 +132,7 @@ test('existing files remain ignored unless the option is enabled', () => { }); test('a changed folder or filter creates a new initial scope', () => { - const { calls, monitor } = createHarness(); + const { calls, monitor } = createWatcherHarness(); monitor.start({ folderPath: 'C:\\incoming', includeExisting: true, recursive: false, extensions: 'mp4' }); monitor.start({ folderPath: 'D:\\incoming', includeExisting: true, recursive: false, extensions: 'mp4' }); monitor.start({ folderPath: 'D:\\incoming', includeExisting: true, recursive: false, extensions: 'mkv' }); @@ -40,7 +140,7 @@ test('a changed folder or filter creates a new initial scope', () => { }); test('initial scan completion is exposed so the one-time option can be persisted as consumed', () => { - const { calls, monitor } = createHarness(); + const { calls, monitor } = createWatcherHarness(); let completed = 0; monitor.on('initial-scan-complete', () => { completed++; }); monitor.start({ folderPath: 'C:\\incoming', includeExisting: true, recursive: false }); @@ -48,3 +148,97 @@ test('initial scan completion is exposed so the one-time option can be persisted calls[0].watcher.emit('ready'); assert.equal(completed, 1); }); + +test('dry scan returns matching descriptors without emitting new files', async () => { + const { monitor, events } = createScanHarness({ + files: [ + { path: 'C:\\incoming\\a.mkv', name: 'a.mkv', size: 10, mtimeMs: 1 }, + { path: 'C:\\incoming\\b.txt', name: 'b.txt', size: 10, mtimeMs: 2 } + ] + }); + monitor.start({ folderPath: 'C:\\incoming', extensions: 'mkv', filterMode: 'include', recursive: true, reconcileIntervalMinutes: 5 }); + const result = await monitor.scan({ emitFiles: false, trigger: 'test' }); + assert.deepEqual(result.files, [{ path: 'C:\\incoming\\a.mkv', name: 'a.mkv', size: 10, mtimeMs: 1 }]); + assert.equal(events.newFiles.length, 0); +}); + +test('overlapping reconcile requests serialize and collapse to one follow-up scan', async () => { + const { monitor, releaseFirstScan, scanCalls } = createDeferredScanHarness(); + const first = monitor.scan({ emitFiles: true, trigger: 'interval' }); + const second = monitor.scan({ emitFiles: true, trigger: 'reconnect' }); + const third = monitor.scan({ emitFiles: true, trigger: 'manual' }); + releaseFirstScan(); + await Promise.all([first, second, third]); + assert.equal(scanCalls(), 2); +}); + +test('disconnect preserves configuration and reconnect performs one immediate scan', async () => { + const { monitor, setReachable, statusEvents, scanCalls, runInterval } = createReachabilityHarness(false); + monitor.start({ folderPath: 'Z:\\watch', reconcileIntervalMinutes: 5 }); + await runInterval(); + assert.equal(statusEvents.at(-1).reachable, false); + assert.equal(monitor.status().folderPath, 'Z:\\watch'); + setReachable(true); + await runInterval(); + assert.equal(statusEvents.at(-1).reachable, true); + assert.equal(scanCalls(), 1); +}); + +test('reachable reconciliation intervals scan the configured folder', async () => { + const { monitor, runInterval, scanCalls } = createReachabilityHarness(true); + monitor.start({ folderPath: 'C:\\watch', reconcileIntervalMinutes: 5 }); + await runInterval(); + await runInterval(); + assert.equal(scanCalls(), 2); +}); + +test('pause stops watcher and reconciliation until explicit resume', async () => { + const { monitor, runInterval, scanCalls } = createReachabilityHarness(true); + monitor.start({ folderPath: 'C:\\watch', reconcileIntervalMinutes: 5 }); + await monitor.pause(); + await runInterval(); + assert.equal(scanCalls(), 0); + assert.equal(monitor.status().paused, true); + await monitor.resume({ folderPath: 'C:\\watch', reconcileIntervalMinutes: 5 }); + assert.equal(scanCalls(), 1); + assert.equal(monitor.status().paused, false); +}); + +test('stop emits an immutable non-running status snapshot', () => { + const { monitor } = createWatcherHarness(); + const statuses = []; + monitor.on('status', (status) => statuses.push(status)); + monitor.start({ folderPath: 'C:\\watch', reconcileIntervalMinutes: 5 }); + const statusCount = statuses.length; + monitor.stop(); + assert.equal(statuses.length, statusCount + 1); + assert.equal(statuses.at(-1).running, false); + assert.equal(Object.isFrozen(statuses.at(-1)), true); +}); + +test('real temporary folder scan survives disconnect and reconnect', async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), 'mhu-automation-folder-')); + const detached = `${root}-detached`; + try { + fs.mkdirSync(path.join(root, 'nested')); + fs.writeFileSync(path.join(root, 'a.mkv'), Buffer.from('a')); + fs.writeFileSync(path.join(root, 'ignored.txt'), Buffer.from('b')); + fs.writeFileSync(path.join(root, 'nested', 'c.mkv'), Buffer.from('c')); + const monitor = new FolderMonitor(); + monitor.start({ folderPath: root, recursive: true, extensions: 'mkv', filterMode: 'include', reconcileIntervalMinutes: 5 }); + const first = await monitor.scan({ emitFiles: false, trigger: 'test' }); + assert.deepEqual(first.files.map((file) => file.name).sort(), ['a.mkv', 'c.mkv']); + assert.equal(first.files.every((file) => Number.isFinite(file.mtimeMs)), true); + fs.renameSync(root, detached); + const disconnected = await monitor.scan({ emitFiles: false, trigger: 'interval' }); + assert.equal(disconnected.reachable, false); + fs.renameSync(detached, root); + const reconnected = await monitor.scan({ emitFiles: false, trigger: 'interval' }); + assert.equal(reconnected.reachable, true); + assert.equal(reconnected.reconnected, true); + await monitor.pause(); + } finally { + if (fs.existsSync(detached)) fs.renameSync(detached, root); + fs.rmSync(root, { recursive: true, force: true }); + } +});