feat(plugin-kit): @punktfunk/plugin-kit 0.1.0 — the Effect plugin framework

The ~80% of every plugin that was copy-pasted (rom-manager ↔ playnite),
extracted as one Effect-v4 package:

- runtime: definePluginKit — async-main boundary hiding a ManagedRuntime
  (two-effect-instances discipline), signal-driven interruption with a
  bounded shutdown grace
- config: Schema-driven raw round-trip (defaults only in the Schema via
  withDecodingDefaultKey + encodingStrategy omit; file stays authored-shape),
  atomic writes, world-writable refusal, changes stream
- cache-store: disposable derived state, corrupt→empty, write-through
- reconcile: kit-owned provider wire schemas + typed client over the
  skew-safe untyped pf.request seam
- sync-engine: generic poll/watch/debounce/single-flight-coalesce/
  fingerprint-skip engine with a status PubSub (the SSE feed)
- ui-server + sse: effect/unstable/httpapi behind servePluginUi with
  core-only env layers (validated by the phase-0 spikes); raw SSE route
  (httpapi has no event-stream media type)
- cli: minimal command dispatcher reusing the plugin layer graph
  (deliberately not effect/unstable/cli — it needs platform packages)
- react subpath: plugin router (path→hash→fallback deep-link restore +
  pf-ui:navigate bridge), ResultGate, sseAtom, resolvePluginBase
- theme.css: the console's violet identity packaged for plugin UIs

18 bun tests incl. the two phase-0 spikes; publish workflow mirrors
sdk-publish (tag plugin-kit-v*; 0.1.0 published manually).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

plugin-kit/src/sync-engine.ts

@@ -0,0 +1,285 @@
+// The generic sync engine — the poll/watch/debounce/coalesce/fingerprint machinery that
+// was ~duplicated between rom-manager and playnite, as one Effect service.
+//
+// Semantics are a faithful port of the original Engine guard:
+//   - single-flight: a sync while one runs records a pending trigger and returns
+//     AlreadyRunning; the running pass re-fires once ("coalesced") when it finishes
+//   - content fingerprint (sha256 of the entries JSON) skips the apply when unchanged
+//   - interval poll + best-effort fs watchers (recursive where the OS supports it, top-dir
+//     fallback on Linux) with debounce; the poll is the real safety net on SMB/NFS
+//   - every transition publishes a SyncStatus (the UI's SSE feed)
+// All loops live in a private scope that `reconfigure` closes and rebuilds, and the whole
+// engine tears down with the Scope it was constructed in.
+import { createHash } from "node:crypto";
+import * as fs from "node:fs";
+import {
+	Cause,
+	type Duration,
+	Effect,
+	Exit,
+	PubSub,
+	Queue,
+	Ref,
+	Scope,
+	Stream,
+} from "effect";
+import { SyncError } from "./errors.js";
+
+export type SyncReason =
+	| "startup"
+	| "poll"
+	| "fs-change"
+	| "config-change"
+	| "manual"
+	| "coalesced";
+
+export interface LastSync {
+	readonly fingerprint: string;
+	readonly count: number;
+	readonly at: number;
+}
+
+export type SyncOutcome<Report> =
+	| { readonly _tag: "Applied"; readonly report: Report; readonly count: number }
+	| { readonly _tag: "Unchanged"; readonly report: Report }
+	| { readonly _tag: "AlreadyRunning" };
+
+export interface SyncStatus<Report> {
+	readonly syncing: boolean;
+	readonly lastReport?: Report;
+	readonly lastSync?: LastSync;
+}
+
+export interface SyncSettings {
+	readonly pollInterval: Duration.Duration;
+	readonly watch: boolean;
+	readonly debounce: Duration.Duration;
+	readonly watchDirs: ReadonlyArray<string>;
+}
+
+export interface SyncEngineOptions<
+	Report,
+	Entries extends ReadonlyArray<unknown>,
+	R,
+> {
+	/** Produce the full desired state. Pure of host effects — `apply` does the write. */
+	readonly compute: (
+		reason: SyncReason,
+	) => Effect.Effect<
+		{ readonly entries: Entries; readonly report: Report },
+		unknown,
+		R
+	>;
+	/** Push the desired state to the host (usually `ProviderClient.reconcile`). */
+	readonly apply: (entries: Entries) => Effect.Effect<void, unknown, R>;
+	/** Override the content fingerprint (default: sha256 of the entries JSON). */
+	readonly fingerprint?: (entries: Entries) => string;
+	/** Durable fingerprint storage (usually the plugin's CacheStore). */
+	readonly lastSync: {
+		readonly get: Effect.Effect<LastSync | undefined, never, R>;
+		readonly set: (last: LastSync) => Effect.Effect<void, never, R>;
+	};
+	/** Re-read on every `reconfigure` — loops restart with fresh settings. */
+	readonly settings: Effect.Effect<SyncSettings, never, R>;
+}
+
+export interface SyncEngine<Report> {
+	readonly sync: (
+		reason: SyncReason,
+	) => Effect.Effect<SyncOutcome<Report>, SyncError>;
+	readonly status: Effect.Effect<SyncStatus<Report>>;
+	/** Emits on every sync start/finish — the UI's SSE feed. */
+	readonly changes: Stream.Stream<SyncStatus<Report>>;
+	/** Initial sync + start poll/watch loops (scoped to the construction Scope). */
+	readonly start: Effect.Effect<void>;
+	/** Restart loops with fresh settings, then sync("config-change"). */
+	readonly reconfigure: Effect.Effect<void>;
+}
+
+const defaultFingerprint = (entries: unknown): string =>
+	createHash("sha256").update(JSON.stringify(entries)).digest("hex");
+
+/** Best-effort watcher set over `dirs`: recursive where supported, top-dir fallback. */
+const openWatchers = (
+	dirs: ReadonlyArray<string>,
+	onEvent: () => void,
+): fs.FSWatcher[] => {
+	const out: fs.FSWatcher[] = [];
+	for (const dir of dirs) {
+		try {
+			out.push(fs.watch(dir, { recursive: true }, onEvent));
+		} catch {
+			try {
+				out.push(fs.watch(dir, onEvent));
+			} catch {
+				// unwatchable (poll covers it)
+			}
+		}
+	}
+	return out;
+};
+
+export const makeSyncEngine = <
+	Report,
+	Entries extends ReadonlyArray<unknown>,
+	R,
+>(
+	opts: SyncEngineOptions<Report, Entries, R>,
+): Effect.Effect<SyncEngine<Report>, never, R | Scope.Scope> =>
+	Effect.gen(function* () {
+		const ctx = yield* Effect.context<R>();
+		const run = <A, E>(eff: Effect.Effect<A, E, R>): Effect.Effect<A, E> =>
+			Effect.provide(eff, ctx);
+		const fingerprint = opts.fingerprint ?? defaultFingerprint;
+
+		const flags = yield* Ref.make({ syncing: false, pending: false });
+		const lastReport = yield* Ref.make<Report | undefined>(undefined);
+		const hub = yield* PubSub.unbounded<SyncStatus<Report>>();
+
+		const status: Effect.Effect<SyncStatus<Report>> = Effect.gen(function* () {
+			const f = yield* Ref.get(flags);
+			return {
+				syncing: f.syncing,
+				lastReport: yield* Ref.get(lastReport),
+				lastSync: yield* run(opts.lastSync.get),
+			};
+		});
+		const publish = status.pipe(
+			Effect.flatMap((s) => PubSub.publish(hub, s)),
+			Effect.asVoid,
+		);
+
+		const doSync = (
+			reason: SyncReason,
+		): Effect.Effect<SyncOutcome<Report>, SyncError> =>
+			Effect.gen(function* () {
+				const { entries, report } = yield* run(opts.compute(reason)).pipe(
+					Effect.mapError((cause) => new SyncError({ reason, cause })),
+				);
+				yield* Ref.set(lastReport, report);
+				const fp = fingerprint(entries);
+				const prev = yield* run(opts.lastSync.get);
+				if (prev?.fingerprint === fp) {
+					yield* Effect.log(
+						`sync (${reason}): no changes (${entries.length} entries)`,
+					);
+					return { _tag: "Unchanged", report } as const;
+				}
+				yield* run(opts.apply(entries)).pipe(
+					Effect.mapError((cause) => new SyncError({ reason, cause })),
+				);
+				yield* run(
+					opts.lastSync.set({
+						fingerprint: fp,
+						count: entries.length,
+						at: Date.now(),
+					}),
+				);
+				yield* Effect.log(
+					`sync (${reason}): reconciled ${entries.length} entries`,
+				);
+				return { _tag: "Applied", report, count: entries.length } as const;
+			});
+
+		// Errors must never kill a loop — log and carry on (the original's catch).
+		const safeSync = (reason: SyncReason): Effect.Effect<void> =>
+			sync(reason).pipe(
+				Effect.catch((e: SyncError) =>
+					Effect.logWarning(`sync (${reason}) failed: ${e.cause}`),
+				),
+				Effect.asVoid,
+			);
+
+		const sync = (
+			reason: SyncReason,
+		): Effect.Effect<SyncOutcome<Report>, SyncError> =>
+			Ref.modify(flags, (f) =>
+				f.syncing
+					? ([false, { ...f, pending: true }] as const)
+					: ([true, { ...f, syncing: true }] as const),
+			).pipe(
+				Effect.flatMap((acquired) => {
+					if (!acquired) {
+						return Effect.succeed({ _tag: "AlreadyRunning" } as const);
+					}
+					return publish.pipe(
+						Effect.andThen(doSync(reason)),
+						Effect.ensuring(
+							Effect.gen(function* () {
+								const pending = yield* Ref.modify(
+									flags,
+									(f) =>
+										[f.pending, { syncing: false, pending: false }] as const,
+								);
+								yield* publish;
+								if (pending) {
+									yield* Effect.forkDetach(safeSync("coalesced"));
+								}
+							}),
+						),
+					);
+				}),
+			);
+
+		// ---------------------------------------------------------------- loops
+		const loopScope = yield* Ref.make<Scope.Closeable | undefined>(undefined);
+
+		const stopLoops = Effect.gen(function* () {
+			const prev = yield* Ref.getAndSet(loopScope, undefined);
+			if (prev) yield* Scope.close(prev, Exit.void);
+		});
+
+		const startLoops: Effect.Effect<void> = Effect.gen(function* () {
+			yield* stopLoops;
+			const scope = yield* Scope.make();
+			yield* Ref.set(loopScope, scope);
+			const settings = yield* run(opts.settings);
+
+			const pollLoop = Effect.forever(
+				Effect.sleep(settings.pollInterval).pipe(
+					Effect.andThen(safeSync("poll")),
+				),
+			);
+			yield* Effect.forkIn(pollLoop, scope);
+
+			if (settings.watch && settings.watchDirs.length > 0) {
+				const watchStream = Stream.callback<void>((queue) =>
+					Effect.acquireRelease(
+						Effect.sync(() =>
+							openWatchers(settings.watchDirs, () => {
+								Queue.offerUnsafe(queue, undefined);
+							}),
+						),
+						(watchers) =>
+							Effect.sync(() => {
+								for (const w of watchers) {
+									try {
+										w.close();
+									} catch {
+										// already closed
+									}
+								}
+							}),
+					),
+				);
+				const watchLoop = watchStream.pipe(
+					Stream.debounce(settings.debounce),
+					Stream.runForEach(() => safeSync("fs-change")),
+				);
+				yield* Effect.forkIn(watchLoop, scope);
+			}
+		});
+
+		// Loop teardown rides the construction Scope (plugin shutdown).
+		yield* Effect.addFinalizer(() => stopLoops);
+
+		return {
+			sync,
+			status,
+			changes: Stream.fromPubSub(hub),
+			start: safeSync("startup").pipe(Effect.andThen(startLoops)),
+			reconfigure: startLoops.pipe(
+				Effect.andThen(safeSync("config-change")),
+			),
+		} satisfies SyncEngine<Report>;
+	});