1// @effect-diagnostics nodeBuiltinImport:off - the guarded file read needs open
2// flags (O_NOFOLLOW, O_NONBLOCK) the FileSystem service does not expose.
3/**
4 * EnvironmentTheme - palettes this machine publishes for clients to follow.
5 *
6 * A desktop that retints its apps when the user switches system theme writes
7 * `<stateDir>/themes/<id>.json`; this service watches that directory and
8 * streams the published set to connected clients so a theme change lands
9 * without a restart. The filename is the theme id: it stays stable while the
10 * machine rewrites the colors underneath, so `defaultTheme` and a client\'s
11 * selection keep pointing at the same theme across recolors. Theming is
12 * cosmetic, so every failure here degrades to "not published" rather than
13 * propagating.
14 *
15 * @module EnvironmentTheme
16 */
17import * as NodeFS from "node:fs";
18
19import {
20 EnvironmentTheme,
21 EnvironmentThemeFile,
22 EnvironmentThemeId,
23 environmentThemeFileHasColors,
24} from "@t3tools/contracts";
25import { UNPUBLISHABLE_THEME_IDS } from "@t3tools/shared/themePalettes";
26import * as Cause from "effect/Cause";
27import * as Context from "effect/Context";
28import * as Duration from "effect/Duration";
29import * as Effect from "effect/Effect";
30import * as Equal from "effect/Equal";
31import * as Exit from "effect/Exit";
32import * as FileSystem from "effect/FileSystem";
33import * as Layer from "effect/Layer";
34import * as PubSub from "effect/PubSub";
35import * as Ref from "effect/Ref";
36import * as Schema from "effect/Schema";
37import * as Semaphore from "effect/Semaphore";
38import * as Scope from "effect/Scope";
39import * as Stream from "effect/Stream";
40
41import * as ServerConfig from "./config.ts";
42
43const decodeEnvironmentThemeFileJsonExit = Schema.decodeUnknownExit(
44 Schema.fromJsonString(EnvironmentThemeFile),
45);
46const isEnvironmentThemeId = Schema.is(EnvironmentThemeId);
47
48const THEME_FILE_SUFFIX = ".json";
49
50/**
51 * Bounds on what a machine can publish. The directory is local, so this is not
52 * a trust boundary -- but an accidental dump of large files there would
53 * otherwise be read in full, streamed to every client, and repainted, so the
54 * cost of a mistake is capped rather than unbounded.
55 */
56const MAX_THEME_FILES = 32;
57/** Exported so the publish path cannot accept a file the watcher will skip. */
58export const MAX_THEME_FILE_BYTES = 32 * 1024;
59/**
60 * The set travels whole in a websocket event to every subscriber, so the sum
61 * matters more than any single file. An exported theme runs a few KB, leaving
62 * this far above any real directory while keeping a mistake off the wire.
63 */
64const MAX_THEME_TOTAL_BYTES = 192 * 1024;
65
66/** The published set with the sequence number it was observed at. */
67interface PublishedThemes {
68 readonly seq: number;
69 readonly themes: ReadonlyArray<EnvironmentTheme>;
70}
71
72export class EnvironmentThemeService extends Context.Service<
73 EnvironmentThemeService,
74 {
75 /**
76 * The set published right now, read from disk rather than from the
77 * watcher\'s last observation: a client connecting must see what the
78 * machine actually publishes even if it missed a filesystem event.
79 */
80 readonly current: Effect.Effect<ReadonlyArray<EnvironmentTheme>>;
81
82 /**
83 * The current set followed by every change, with repeats dropped. The
84 * subscription is acquired before the current set is read, so a publish
85 * landing while a client connects is delivered rather than lost.
86 */
87 readonly streamChanges: Stream.Stream<ReadonlyArray<EnvironmentTheme>>;
88 }
89>()("t3/environmentTheme/EnvironmentThemeService") {}
90
91/**
92 * Reads a theme file through one opened handle, so every check binds to the
93 * file actually read rather than to a path that may have been swapped since:
94 * O_NOFOLLOW rejects a symlink outright (a symlinked themes directory stays
95 * usable, a symlinked file inside it does not), O_NONBLOCK keeps a FIFO from
96 * blocking the open, and the fstat type and size gate examines the open
97 * descriptor. Returns null for anything that is not a small regular file.
98 *
99 * Windows has neither flag (the constants are undefined, and OR-ing them in
100 * is a no-op), so the symlink check there is an lstat before the open. That
101 * leaves a window a swap could slip through, which the descriptor-bound
102 * checks below then narrow to "a regular file at that path".
103 */
104export const readThemeFileGuarded = (filePath: string, maxBytes: number): string | null => {
105 let fd: number;
106 try {
107 if (NodeFS.constants.O_NOFOLLOW === undefined && NodeFS.lstatSync(filePath).isSymbolicLink()) {
108 return null;
109 }
110 fd = NodeFS.openSync(
111 filePath,
112 NodeFS.constants.O_RDONLY |
113 (NodeFS.constants.O_NOFOLLOW ?? 0) |
114 (NodeFS.constants.O_NONBLOCK ?? 0),
115 );
116 } catch {
117 return null;
118 }
119 try {
120 const info = NodeFS.fstatSync(fd);
121 if (!info.isFile() || info.size > maxBytes) return null;
122 const contents = Buffer.alloc(info.size);
123 let offset = 0;
124 while (offset < contents.length) {
125 const read = NodeFS.readSync(fd, contents, offset, contents.length - offset, offset);
126 if (read <= 0) break;
127 offset += read;
128 }
129 return contents.subarray(0, offset).toString("utf8");
130 } catch {
131 return null;
132 } finally {
133 NodeFS.closeSync(fd);
134 }
135};
136
137/**
138 * Every theme the directory actually publishes. A file that is missing,
139 * unreadable, malformed, colorless, or misnamed is simply skipped; the rest of
140 * the set is unaffected. The one place that decides what "published" means, so
141 * a caller validating an id cannot disagree with the watcher serving it.
142 */
143export const readPublishedThemes = Effect.fn(function* (themesDir: string) {
144 const fs = yield* FileSystem.FileSystem;
145 const entries = yield* fs
146 .readDirectory(themesDir)
147 .pipe(Effect.orElseSucceed((): Array<string> => []));
148
149 const themes: Array<EnvironmentTheme> = [];
150 let examined = 0;
151 let totalBytes = 0;
152 for (const entry of entries.toSorted()) {
153 if (!entry.endsWith(THEME_FILE_SUFFIX)) continue;
154 const id = entry.slice(0, -THEME_FILE_SUFFIX.length);
155 // A reserved id is either shadowed by a built-in on the client or captures
156 // clients that never chose it, so it is not publishable.
157 if (!isEnvironmentThemeId(id) || UNPUBLISHABLE_THEME_IDS.has(id)) continue;
158
159 // Counts files examined, not themes accepted: capping the output would
160 // let a directory of malformed files be opened, read, and decoded in full
161 // on every refresh and every client connect.
162 examined += 1;
163 if (examined > MAX_THEME_FILES) {
164 yield* Effect.logWarning("ignoring environment theme files past the limit", {
165 path: themesDir,
166 limit: MAX_THEME_FILES,
167 });
168 break;
169 }
170
171 const filePath = `${themesDir}/${entry}`;
172 const raw = readThemeFileGuarded(filePath, MAX_THEME_FILE_BYTES);
173 if (raw === null) {
174 yield* Effect.logWarning("ignoring unusable environment theme file", {
175 path: filePath,
176 limit: MAX_THEME_FILE_BYTES,
177 });
178 continue;
179 }
180 if (raw.trim().length === 0) continue;
181
182 const decoded = decodeEnvironmentThemeFileJsonExit(raw);
183 if (decoded._tag === "Failure") {
184 yield* Effect.logWarning("ignoring invalid environment theme", {
185 path: filePath,
186 detail: Cause.pretty(decoded.cause),
187 });
188 continue;
189 }
190 const file = decoded.value;
191 if (!environmentThemeFileHasColors(file)) {
192 yield* Effect.logWarning("ignoring environment theme without colors", { path: filePath });
193 continue;
194 }
195
196 // Counted only once accepted: the cap bounds what travels to clients, so
197 // a skipped file must not eat the budget of valid themes sorted after it.
198 // Bytes, not string length -- the cap describes wire weight.
199 totalBytes += Buffer.byteLength(raw);
200 if (totalBytes > MAX_THEME_TOTAL_BYTES) {
201 yield* Effect.logWarning("ignoring environment themes past the total size limit", {
202 path: themesDir,
203 limit: MAX_THEME_TOTAL_BYTES,
204 });
205 break;
206 }
207
208 themes.push({ id, ...file });
209 }
210 return themes;
211});
212
213/**
214 * Reads the directory and folds it into the sequenced state, publishing only
215 * a genuine change. Every reader goes through here, so the snapshot a client
216 * connects on and the events it then receives come from one ordered source
217 * rather than from disk and the queue independently.
218 */
219
220const make = Effect.gen(function* () {
221 const { environmentThemesDir } = yield* ServerConfig.ServerConfig;
222 const fs = yield* FileSystem.FileSystem;
223 /**
224 * Sliding with capacity 1: every update carries the complete set, so a
225 * subscriber that stops consuming holds at most the newest set rather than
226 * an unbounded backlog. Every observed set carries a sequence number, so a
227 * subscriber can drop queued events that predate the snapshot it started
228 * from. Without it a publish landing between subscribing and reading
229 * replays after the newer value and walks clients backwards onto stale
230 * colors.
231 */
232 const changes = yield* PubSub.sliding<PublishedThemes>(1);
233 const published = yield* Ref.make<PublishedThemes>({ seq: 0, themes: [] });
234 /**
235 * Guards the whole read/compare/publish, not just the state update. The
236 * directory read is async, so two concurrent refreshes can finish out of
237 * order and a slower read of an older set would publish under a higher
238 * sequence -- which the subscriber filter, ordering publications rather than
239 * observations, could not then drop.
240 */
241 const refreshSemaphore = yield* Semaphore.make(1);
242 const watcherScope = yield* Scope.make("sequential");
243 yield* Effect.addFinalizer(() => Scope.close(watcherScope, Exit.void));
244
245 const refresh = refreshSemaphore.withPermits(1)(
246 Effect.gen(function* () {
247 const themes = yield* readPublishedThemes(environmentThemesDir).pipe(
248 Effect.provideService(FileSystem.FileSystem, fs),
249 );
250 // Structural equality over the whole decoded value: a hand-rolled field
251 // list here silently drops republishes for any field it forgets.
252 const [changed, next] = yield* Ref.modify(
253 published,
254 (previous): readonly [readonly [boolean, PublishedThemes], PublishedThemes] => {
255 if (Equal.equals(previous.themes, themes)) return [[false, previous], previous];
256 const updated: PublishedThemes = { seq: previous.seq + 1, themes };
257 return [[true, updated], updated];
258 },
259 );
260 if (changed) yield* PubSub.publish(changes, next).pipe(Effect.asVoid);
261 return next;
262 }),
263 );
264
265 // The directory is created up front so the watcher has something to attach
266 // to before the first publisher writes into it.
267 yield* fs
268 .makeDirectory(environmentThemesDir, { recursive: true })
269 .pipe(Effect.ignoreCause({ log: true }));
270
271 // Debounced for the same reason settings watching is: a theme script emits
272 // several events per save and `fs.watch` can fire before the content is
273 // flushed. Every event triggers a full re-read, so no event needs filtering.
274 const watchEvents = fs.watch(environmentThemesDir).pipe(Stream.debounce(Duration.millis(100)));
275
276 // Seeds the dedupe so a watch event that reports no actual change (a touch,
277 // a rewrite with identical contents) does not retint every client.
278 yield* refresh;
279 yield* Stream.runForEach(watchEvents, () => refresh.pipe(Effect.ignoreCause({ log: true }))).pipe(
280 Effect.ignoreCause({ log: true }),
281 Effect.forkIn(watcherScope),
282 Effect.asVoid,
283 );
284
285 return {
286 current: Effect.map(refresh, (state) => state.themes),
287 get streamChanges() {
288 return Stream.unwrap(
289 Effect.gen(function* () {
290 // Subscribe first so nothing published during the read is missed,
291 // then drop anything the snapshot already accounts for.
292 const subscription = yield* PubSub.subscribe(changes);
293 const snapshot = yield* refresh;
294 return Stream.concat(
295 Stream.make(snapshot.themes),
296 Stream.fromSubscription(subscription).pipe(
297 Stream.filter((update) => update.seq > snapshot.seq),
298 Stream.map((update) => update.themes),
299 ),
300 );
301 }),
302 );
303 },
304 } satisfies EnvironmentThemeService["Service"];
305});
306
307export const layer = Layer.effect(EnvironmentThemeService, make);