// SyncEngine semantics: fingerprint skip, single-flight coalescing, status feed. import { describe, expect, test } from "bun:test"; import { Duration, Effect, Fiber, Ref, Scope, Stream } from "effect"; import { type LastSync, makeSyncEngine, type SyncOutcome, } from "../src/index.js"; interface Report { readonly included: number; } const harness = (opts?: { computeDelayMs?: number; entries?: () => ReadonlyArray; }) => Effect.gen(function* () { const applied = yield* Ref.make(0); const last = yield* Ref.make(undefined); const entries = opts?.entries ?? (() => ["a", "b"]); const engine = yield* makeSyncEngine, never>( { compute: () => Effect.suspend(() => { const e = entries(); return Effect.succeed({ entries: e, report: { included: e.length }, }); }).pipe( opts?.computeDelayMs ? Effect.delay(Duration.millis(opts.computeDelayMs)) : (x) => x, ), apply: () => Ref.update(applied, (n) => n + 1), lastSync: { get: Ref.get(last), set: (l) => Ref.set(last, l), }, settings: Effect.succeed({ pollInterval: Duration.minutes(60), watch: false, debounce: Duration.millis(10), watchDirs: [], }), }, ); return { engine, applied, last }; }); const run = (eff: Effect.Effect): Promise => Effect.runPromise(Effect.scoped(eff) as Effect.Effect); describe("SyncEngine", () => { test("first sync applies; unchanged content skips the apply", async () => { const { first, second, count } = await run( Effect.gen(function* () { const h = yield* harness(); const first = yield* h.engine.sync("manual"); const second = yield* h.engine.sync("manual"); return { first, second, count: yield* Ref.get(h.applied), }; }), ); expect(first._tag).toBe("Applied"); if (first._tag === "Applied") expect(first.count).toBe(2); expect(second._tag).toBe("Unchanged"); expect(count).toBe(1); }); test("changed content re-applies", async () => { let call = 0; const { outcomes, count } = await run( Effect.gen(function* () { const h = yield* harness({ entries: () => (call++ === 0 ? ["a"] : ["a", "b"]), }); const o1 = yield* h.engine.sync("manual"); const o2 = yield* h.engine.sync("manual"); return { outcomes: [o1, o2], count: yield* Ref.get(h.applied) }; }), ); expect(outcomes.map((o: SyncOutcome) => o._tag)).toEqual([ "Applied", "Applied", ]); expect(count).toBe(2); }); test("concurrent trigger returns AlreadyRunning and coalesces into a follow-up", async () => { const { during, count } = await run( Effect.gen(function* () { const h = yield* harness({ computeDelayMs: 50 }); const fiber = yield* Effect.forkChild(h.engine.sync("manual")); yield* Effect.sleep("10 millis"); const during = yield* h.engine.sync("manual"); yield* Fiber.join(fiber); // the coalesced re-run is forked detached — give it a beat yield* Effect.sleep("120 millis"); return { during, count: yield* Ref.get(h.applied) }; }), ); expect(during._tag).toBe("AlreadyRunning"); // first sync applied; the coalesced pass found identical content → skipped expect(count).toBe(1); }); test("status feed publishes syncing transitions", async () => { const statuses = await run( Effect.gen(function* () { const h = yield* harness(); const fiber = yield* Effect.forkChild( h.engine.changes.pipe(Stream.take(2), Stream.runCollect), ); yield* Effect.sleep("20 millis"); yield* h.engine.sync("manual"); return yield* Fiber.join(fiber); }), ); expect(statuses[0]?.syncing).toBe(true); expect(statuses[1]?.syncing).toBe(false); expect(statuses[1]?.lastSync?.count).toBe(2); }); });