diff --git a/Sources/ForgeProxy/WatchFolder.swift b/Sources/ForgeProxy/WatchFolder.swift new file mode 100644 index 0000000..582044d --- /dev/null +++ b/Sources/ForgeProxy/WatchFolder.swift @@ -0,0 +1,101 @@ +import Foundation + +/// Tracks files already handed off — shared across watcher restarts. +public actor ProcessedSkipList { + private var processed = Set() + + public init() {} + + public func contains(_ path: String) -> Bool { + processed.contains(path) + } + + public func mark(_ path: String) { + processed.insert(path) + } +} + +/// Polling watch folder. Emits file paths once they are: +/// - matching extension +/// - size-stable for `stabilityChecks` consecutive polls (partial-write guard) +/// - not in the skip list +/// Polling (not inotify): cheap at DIT-cart scale, identical behavior on +/// network volumes where inotify is unreliable. +public actor WatchFolder { + private let path: String + private let extensions: Set + private let pollInterval: Duration + private let stabilityChecks: Int + private let skipList: ProcessedSkipList + + private var sizeHistory: [String: (size: UInt64, stableCount: Int)] = [:] + private var task: Task? + private var continuation: AsyncStream.Continuation? + + public init( + path: String, + extensions: [String], + pollInterval: Duration = .seconds(2), + stabilityChecks: Int = 2, + skipList: ProcessedSkipList = ProcessedSkipList() + ) { + self.path = path + self.extensions = Set(extensions.map { $0.lowercased() }) + self.pollInterval = pollInterval + self.stabilityChecks = stabilityChecks + self.skipList = skipList + } + + /// Start watching; returns stream of ready file paths. + public func start() -> AsyncStream { + let stream = AsyncStream { continuation in + self.continuation = continuation + } + task = Task { + while !Task.isCancelled { + await pollOnce() + try? await Task.sleep(for: pollInterval) + } + } + return stream + } + + public func stop() { + task?.cancel() + task = nil + continuation?.finish() + continuation = nil + } + + private func pollOnce() async { + let fm = FileManager.default + guard let entries = try? fm.contentsOfDirectory(atPath: path) else { return } + + for entry in entries { + let ext = (entry as NSString).pathExtension.lowercased() + guard extensions.contains(ext) else { continue } + let full = path + "/" + entry + if await skipList.contains(full) { continue } + + guard let attrs = try? fm.attributesOfItem(atPath: full), + let size = attrs[.size] as? UInt64 else { continue } + + if let prev = sizeHistory[full] { + if prev.size == size { + let stable = prev.stableCount + 1 + if stable >= stabilityChecks { + await skipList.mark(full) + sizeHistory.removeValue(forKey: full) + continuation?.yield(full) + } else { + sizeHistory[full] = (size, stable) + } + } else { + sizeHistory[full] = (size, 0) // still growing + } + } else { + sizeHistory[full] = (size, 0) + } + } + } +} diff --git a/Tests/ForgeProxyTests/WatchFolderTests.swift b/Tests/ForgeProxyTests/WatchFolderTests.swift new file mode 100644 index 0000000..0b5f25b --- /dev/null +++ b/Tests/ForgeProxyTests/WatchFolderTests.swift @@ -0,0 +1,132 @@ +import XCTest +import Foundation +@testable import ForgeProxy + +final class WatchFolderTests: XCTestCase { + + var dir: String! + + override func setUpWithError() throws { + dir = NSTemporaryDirectory() + "forge-watch-\(UUID().uuidString)" + try FileManager.default.createDirectory(atPath: dir, withIntermediateDirectories: true) + } + + override func tearDown() { + try? FileManager.default.removeItem(atPath: dir) + } + + // New stable file enqueued exactly once. + func testNewFileEnqueuedOnce() async throws { + let watcher = WatchFolder( + path: dir, + extensions: ["mov"], + pollInterval: .milliseconds(30), + stabilityChecks: 2) + var enqueued = [String]() + let stream = await watcher.start() + + try Data("clip".utf8).write(to: URL(fileURLWithPath: dir + "/C001.mov")) + + let collector = Task { + for await path in stream { + enqueued.append(path) + } + } + try await Task.sleep(for: .milliseconds(400)) + await watcher.stop() + _ = await collector.result + + XCTAssertEqual(enqueued.count, 1) + XCTAssertTrue(enqueued[0].hasSuffix("C001.mov")) + } + + // Growing file (partial write) not enqueued until size stable. + func testGrowingFileWaitsForStability() async throws { + let watcher = WatchFolder( + path: dir, + extensions: ["mov"], + pollInterval: .milliseconds(40), + stabilityChecks: 2) + let path = dir + "/GROW.mov" + try Data("start".utf8).write(to: URL(fileURLWithPath: path)) + + let stream = await watcher.start() + actor Seen { + var paths = [String]() + var at: [ContinuousClock.Instant] = [] + func add(_ p: String) { + paths.append(p) + at.append(.now) + } + } + let seen = Seen() + let collector = Task { + for await p in stream { + await seen.add(p) + } + } + + // Keep growing for ~200ms. + for i in 0..<5 { + try await Task.sleep(for: .milliseconds(40)) + let h = FileHandle(forWritingAtPath: path)! + h.seekToEndOfFile() + h.write(Data("more\(i)".utf8)) + try h.close() + } + let growEnd = ContinuousClock.now + try await Task.sleep(for: .milliseconds(400)) + await watcher.stop() + _ = await collector.result + + let paths = await seen.paths + let times = await seen.at + XCTAssertEqual(paths.count, 1) + // Enqueue happened after growth stopped. + XCTAssertGreaterThan(times[0], growEnd) + } + + // Extension filter. + func testExtensionFilter() async throws { + let watcher = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1) + try Data("x".utf8).write(to: URL(fileURLWithPath: dir + "/note.txt")) + try Data("y".utf8).write(to: URL(fileURLWithPath: dir + "/C002.mov")) + + let stream = await watcher.start() + var enqueued = [String]() + let collector = Task { + for await p in stream { enqueued.append(p) } + } + try await Task.sleep(for: .milliseconds(300)) + await watcher.stop() + _ = await collector.result + + XCTAssertEqual(enqueued.count, 1) + XCTAssertTrue(enqueued[0].hasSuffix(".mov")) + } + + // Already-processed file skipped across watcher runs via skip list. + func testSkipListPersistsAcrossRuns() async throws { + try Data("x".utf8).write(to: URL(fileURLWithPath: dir + "/C003.mov")) + + let skipList = ProcessedSkipList() + let w1 = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1, skipList: skipList) + let s1 = await w1.start() + var first = [String]() + let c1 = Task { for await p in s1 { first.append(p) } } + try await Task.sleep(for: .milliseconds(250)) + await w1.stop() + _ = await c1.result + XCTAssertEqual(first.count, 1) + + // Second watcher with same skip list: nothing new. + let w2 = WatchFolder(path: dir, extensions: ["mov"], pollInterval: .milliseconds(30), stabilityChecks: 1, skipList: skipList) + let s2 = await w2.start() + var second = [String]() + let c2 = Task { for await p in s2 { second.append(p) } } + try await Task.sleep(for: .milliseconds(250)) + await w2.stop() + _ = await c2.result + XCTAssertTrue(second.isEmpty) + } +}