// @effect-diagnostics nodeBuiltinImport:off - the guarded file read needs open // flags (O_NOFOLLOW, O_NONBLOCK) the FileSystem service does not expose. /** * EnvironmentTheme - palettes this machine publishes for clients to follow. * * A desktop that retints its apps when the user switches system theme writes * `/themes/.json`; this service watches that directory and * streams the published set to connected clients so a theme change lands * without a restart. The filename is the theme id: it stays stable while the * machine rewrites the colors underneath, so `defaultTheme` and a client\'s * selection keep pointing at the same theme across recolors. Theming is * cosmetic, so every failure here degrades to "not published" rather than * propagating. * * @module EnvironmentTheme */ import * as NodeFS from "node:fs"; import { EnvironmentTheme, EnvironmentThemeFile, EnvironmentThemeId, environmentThemeFileHasColors, } from "@t3tools/contracts"; import { UNPUBLISHABLE_THEME_IDS } from "@t3tools/shared/themePalettes"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as ServerConfig from "./config.ts"; const decodeEnvironmentThemeFileJsonExit = Schema.decodeUnknownExit( Schema.fromJsonString(EnvironmentThemeFile), ); const isEnvironmentThemeId = Schema.is(EnvironmentThemeId); const THEME_FILE_SUFFIX = ".json"; /** * Bounds on what a machine can publish. The directory is local, so this is not * a trust boundary -- but an accidental dump of large files there would * otherwise be read in full, streamed to every client, and repainted, so the * cost of a mistake is capped rather than unbounded. */ const MAX_THEME_FILES = 32; /** Exported so the publish path cannot accept a file the watcher will skip. */ export const MAX_THEME_FILE_BYTES = 32 * 1024; /** * The set travels whole in a websocket event to every subscriber, so the sum * matters more than any single file. An exported theme runs a few KB, leaving * this far above any real directory while keeping a mistake off the wire. */ const MAX_THEME_TOTAL_BYTES = 192 * 1024; /** The published set with the sequence number it was observed at. */ interface PublishedThemes { readonly seq: number; readonly themes: ReadonlyArray; } export class EnvironmentThemeService extends Context.Service< EnvironmentThemeService, { /** * The set published right now, read from disk rather than from the * watcher\'s last observation: a client connecting must see what the * machine actually publishes even if it missed a filesystem event. */ readonly current: Effect.Effect>; /** * The current set followed by every change, with repeats dropped. The * subscription is acquired before the current set is read, so a publish * landing while a client connects is delivered rather than lost. */ readonly streamChanges: Stream.Stream>; } >()("t3/environmentTheme/EnvironmentThemeService") {} /** * Reads a theme file through one opened handle, so every check binds to the * file actually read rather than to a path that may have been swapped since: * O_NOFOLLOW rejects a symlink outright (a symlinked themes directory stays * usable, a symlinked file inside it does not), O_NONBLOCK keeps a FIFO from * blocking the open, and the fstat type and size gate examines the open * descriptor. Returns null for anything that is not a small regular file. * * Windows has neither flag (the constants are undefined, and OR-ing them in * is a no-op), so the symlink check there is an lstat before the open. That * leaves a window a swap could slip through, which the descriptor-bound * checks below then narrow to "a regular file at that path". */ export const readThemeFileGuarded = (filePath: string, maxBytes: number): string | null => { let fd: number; try { if (NodeFS.constants.O_NOFOLLOW === undefined && NodeFS.lstatSync(filePath).isSymbolicLink()) { return null; } fd = NodeFS.openSync( filePath, NodeFS.constants.O_RDONLY | (NodeFS.constants.O_NOFOLLOW ?? 0) | (NodeFS.constants.O_NONBLOCK ?? 0), ); } catch { return null; } try { const info = NodeFS.fstatSync(fd); if (!info.isFile() || info.size > maxBytes) return null; const contents = Buffer.alloc(info.size); let offset = 0; while (offset < contents.length) { const read = NodeFS.readSync(fd, contents, offset, contents.length - offset, offset); if (read <= 0) break; offset += read; } return contents.subarray(0, offset).toString("utf8"); } catch { return null; } finally { NodeFS.closeSync(fd); } }; /** * Every theme the directory actually publishes. A file that is missing, * unreadable, malformed, colorless, or misnamed is simply skipped; the rest of * the set is unaffected. The one place that decides what "published" means, so * a caller validating an id cannot disagree with the watcher serving it. */ export const readPublishedThemes = Effect.fn(function* (themesDir: string) { const fs = yield* FileSystem.FileSystem; const entries = yield* fs .readDirectory(themesDir) .pipe(Effect.orElseSucceed((): Array => [])); const themes: Array = []; let examined = 0; let totalBytes = 0; for (const entry of entries.toSorted()) { if (!entry.endsWith(THEME_FILE_SUFFIX)) continue; const id = entry.slice(0, -THEME_FILE_SUFFIX.length); // A reserved id is either shadowed by a built-in on the client or captures // clients that never chose it, so it is not publishable. if (!isEnvironmentThemeId(id) || UNPUBLISHABLE_THEME_IDS.has(id)) continue; // Counts files examined, not themes accepted: capping the output would // let a directory of malformed files be opened, read, and decoded in full // on every refresh and every client connect. examined += 1; if (examined > MAX_THEME_FILES) { yield* Effect.logWarning("ignoring environment theme files past the limit", { path: themesDir, limit: MAX_THEME_FILES, }); break; } const filePath = `${themesDir}/${entry}`; const raw = readThemeFileGuarded(filePath, MAX_THEME_FILE_BYTES); if (raw === null) { yield* Effect.logWarning("ignoring unusable environment theme file", { path: filePath, limit: MAX_THEME_FILE_BYTES, }); continue; } if (raw.trim().length === 0) continue; const decoded = decodeEnvironmentThemeFileJsonExit(raw); if (decoded._tag === "Failure") { yield* Effect.logWarning("ignoring invalid environment theme", { path: filePath, detail: Cause.pretty(decoded.cause), }); continue; } const file = decoded.value; if (!environmentThemeFileHasColors(file)) { yield* Effect.logWarning("ignoring environment theme without colors", { path: filePath }); continue; } // Counted only once accepted: the cap bounds what travels to clients, so // a skipped file must not eat the budget of valid themes sorted after it. // Bytes, not string length -- the cap describes wire weight. totalBytes += Buffer.byteLength(raw); if (totalBytes > MAX_THEME_TOTAL_BYTES) { yield* Effect.logWarning("ignoring environment themes past the total size limit", { path: themesDir, limit: MAX_THEME_TOTAL_BYTES, }); break; } themes.push({ id, ...file }); } return themes; }); /** * Reads the directory and folds it into the sequenced state, publishing only * a genuine change. Every reader goes through here, so the snapshot a client * connects on and the events it then receives come from one ordered source * rather than from disk and the queue independently. */ const make = Effect.gen(function* () { const { environmentThemesDir } = yield* ServerConfig.ServerConfig; const fs = yield* FileSystem.FileSystem; /** * Sliding with capacity 1: every update carries the complete set, so a * subscriber that stops consuming holds at most the newest set rather than * an unbounded backlog. Every observed set carries a sequence number, so a * subscriber can drop queued events that predate the snapshot it started * from. Without it a publish landing between subscribing and reading * replays after the newer value and walks clients backwards onto stale * colors. */ const changes = yield* PubSub.sliding(1); const published = yield* Ref.make({ seq: 0, themes: [] }); /** * Guards the whole read/compare/publish, not just the state update. The * directory read is async, so two concurrent refreshes can finish out of * order and a slower read of an older set would publish under a higher * sequence -- which the subscriber filter, ordering publications rather than * observations, could not then drop. */ const refreshSemaphore = yield* Semaphore.make(1); const watcherScope = yield* Scope.make("sequential"); yield* Effect.addFinalizer(() => Scope.close(watcherScope, Exit.void)); const refresh = refreshSemaphore.withPermits(1)( Effect.gen(function* () { const themes = yield* readPublishedThemes(environmentThemesDir).pipe( Effect.provideService(FileSystem.FileSystem, fs), ); // Structural equality over the whole decoded value: a hand-rolled field // list here silently drops republishes for any field it forgets. const [changed, next] = yield* Ref.modify( published, (previous): readonly [readonly [boolean, PublishedThemes], PublishedThemes] => { if (Equal.equals(previous.themes, themes)) return [[false, previous], previous]; const updated: PublishedThemes = { seq: previous.seq + 1, themes }; return [[true, updated], updated]; }, ); if (changed) yield* PubSub.publish(changes, next).pipe(Effect.asVoid); return next; }), ); // The directory is created up front so the watcher has something to attach // to before the first publisher writes into it. yield* fs .makeDirectory(environmentThemesDir, { recursive: true }) .pipe(Effect.ignoreCause({ log: true })); // Debounced for the same reason settings watching is: a theme script emits // several events per save and `fs.watch` can fire before the content is // flushed. Every event triggers a full re-read, so no event needs filtering. const watchEvents = fs.watch(environmentThemesDir).pipe(Stream.debounce(Duration.millis(100))); // Seeds the dedupe so a watch event that reports no actual change (a touch, // a rewrite with identical contents) does not retint every client. yield* refresh; yield* Stream.runForEach(watchEvents, () => refresh.pipe(Effect.ignoreCause({ log: true }))).pipe( Effect.ignoreCause({ log: true }), Effect.forkIn(watcherScope), Effect.asVoid, ); return { current: Effect.map(refresh, (state) => state.themes), get streamChanges() { return Stream.unwrap( Effect.gen(function* () { // Subscribe first so nothing published during the read is missed, // then drop anything the snapshot already accounts for. const subscription = yield* PubSub.subscribe(changes); const snapshot = yield* refresh; return Stream.concat( Stream.make(snapshot.themes), Stream.fromSubscription(subscription).pipe( Stream.filter((update) => update.seq > snapshot.seq), Stream.map((update) => update.themes), ), ); }), ); }, } satisfies EnvironmentThemeService["Service"]; }); export const layer = Layer.effect(EnvironmentThemeService, make);