1// @effect-diagnostics nodeBuiltinImport:off - the guarded file read needs open
2// flags (O_NOFOLLOW, O_NONBLOCK) the FileSystem service does not expose.
3import * as NodeFS from "node:fs";
4
5import {
6 EnvironmentTheme,
7 EnvironmentThemeFile,
8 EnvironmentThemeId,
9 environmentThemeFileHasColors,
10} from "@t3tools/contracts";
11import { UNPUBLISHABLE_THEME_IDS } from "@t3tools/shared/themePalettes";
12import * as Cause from "effect/Cause";
13import * as Context from "effect/Context";
14import * as Duration from "effect/Duration";
15import * as Effect from "effect/Effect";
16import * as Equal from "effect/Equal";
17import * as Exit from "effect/Exit";
18import * as FileSystem from "effect/FileSystem";
19import * as Layer from "effect/Layer";
20import * as PubSub from "effect/PubSub";
21import * as Ref from "effect/Ref";
22import * as Schema from "effect/Schema";
23import * as Semaphore from "effect/Semaphore";
24import * as Scope from "effect/Scope";
25import * as Stream from "effect/Stream";
26
27import * as ServerConfig from "./config.ts";
28
29const decodeEnvironmentThemeFileJsonExit = Schema.decodeUnknownExit(
30 Schema.fromJsonString(EnvironmentThemeFile),
31);
32const isEnvironmentThemeId = Schema.is(EnvironmentThemeId);
33
34const THEME_FILE_SUFFIX = ".json";
35
36const MAX_THEME_FILES = 32;
37/** Shared with publishing so it cannot accept files the watcher skips. */
38export const MAX_THEME_FILE_BYTES = 32 * 1024;
39const MAX_THEME_TOTAL_BYTES = 192 * 1024;
40
41interface PublishedThemes {
42 readonly seq: number;
43 readonly themes: ReadonlyArray<EnvironmentTheme>;
44}
45
46export class EnvironmentThemeService extends Context.Service<
47 EnvironmentThemeService,
48 {
49 readonly current: Effect.Effect<ReadonlyArray<EnvironmentTheme>>;
50
51 /** Subscribes before reading to retain changes during connection. */
52 readonly streamChanges: Stream.Stream<ReadonlyArray<EnvironmentTheme>>;
53 }
54>()("t3/environmentTheme/EnvironmentThemeService") {}
55
56/** Validates and reads one handle; rejects symlinked files, allows symlinked directories.
57 * Uses non-blocking opens against FIFOs; Windows lacks these flags and checks symlinks with lstat. */
58export const readThemeFileGuarded = (filePath: string, maxBytes: number): string | null => {
59 let fd: number;
60 try {
61 if (NodeFS.constants.O_NOFOLLOW === undefined && NodeFS.lstatSync(filePath).isSymbolicLink()) {
62 return null;
63 }
64 fd = NodeFS.openSync(
65 filePath,
66 NodeFS.constants.O_RDONLY |
67 (NodeFS.constants.O_NOFOLLOW ?? 0) |
68 (NodeFS.constants.O_NONBLOCK ?? 0),
69 );
70 } catch {
71 return null;
72 }
73 try {
74 const info = NodeFS.fstatSync(fd);
75 if (!info.isFile() || info.size > maxBytes) return null;
76 const contents = Buffer.alloc(info.size);
77 let offset = 0;
78 while (offset < contents.length) {
79 const read = NodeFS.readSync(fd, contents, offset, contents.length - offset, offset);
80 if (read <= 0) break;
81 offset += read;
82 }
83 // Unlike Buffer#toString, TextDecoder drops a UTF-8 BOM, which JSON.parse rejects.
84 return new TextDecoder().decode(contents.subarray(0, offset));
85 } catch {
86 return null;
87 } finally {
88 NodeFS.closeSync(fd);
89 }
90};
91
92export const readPublishedThemes = Effect.fn(function* (themesDir: string) {
93 const fs = yield* FileSystem.FileSystem;
94 const entries = yield* fs
95 .readDirectory(themesDir)
96 .pipe(Effect.orElseSucceed((): Array<string> => []));
97
98 const themes: Array<EnvironmentTheme> = [];
99 let examined = 0;
100 let totalBytes = 0;
101 for (const entry of entries.toSorted()) {
102 if (!entry.endsWith(THEME_FILE_SUFFIX)) continue;
103 const id = entry.slice(0, -THEME_FILE_SUFFIX.length);
104 // Reserved ids shadow built-ins or capture clients that never chose them.
105 if (!isEnvironmentThemeId(id) || UNPUBLISHABLE_THEME_IDS.has(id)) continue;
106
107 // Cap examined files, not accepted themes, to bound work on malformed directories.
108 examined += 1;
109 if (examined > MAX_THEME_FILES) {
110 yield* Effect.logWarning("ignoring environment theme files past the limit", {
111 path: themesDir,
112 limit: MAX_THEME_FILES,
113 });
114 break;
115 }
116
117 const filePath = `${themesDir}/${entry}`;
118 const raw = readThemeFileGuarded(filePath, MAX_THEME_FILE_BYTES);
119 if (raw === null) {
120 yield* Effect.logWarning("ignoring unusable environment theme file", {
121 path: filePath,
122 limit: MAX_THEME_FILE_BYTES,
123 });
124 continue;
125 }
126 if (raw.trim().length === 0) continue;
127
128 const decoded = decodeEnvironmentThemeFileJsonExit(raw);
129 if (decoded._tag === "Failure") {
130 yield* Effect.logWarning("ignoring invalid environment theme", {
131 path: filePath,
132 detail: Cause.pretty(decoded.cause),
133 });
134 continue;
135 }
136 const file = decoded.value;
137 if (!environmentThemeFileHasColors(file)) {
138 yield* Effect.logWarning("ignoring environment theme without colors", { path: filePath });
139 continue;
140 }
141
142 totalBytes += Buffer.byteLength(raw);
143 if (totalBytes > MAX_THEME_TOTAL_BYTES) {
144 yield* Effect.logWarning("ignoring environment themes past the total size limit", {
145 path: themesDir,
146 limit: MAX_THEME_TOTAL_BYTES,
147 });
148 break;
149 }
150
151 themes.push({ id, ...file });
152 }
153 return themes;
154});
155
156const make = Effect.gen(function* () {
157 const { environmentThemesDir } = yield* ServerConfig.ServerConfig;
158 const fs = yield* FileSystem.FileSystem;
159 // Sequence numbers let subscribers drop queued events older than their initial snapshot.
160 const changes = yield* PubSub.sliding<PublishedThemes>(1);
161 const published = yield* Ref.make<PublishedThemes>({ seq: 0, themes: [] });
162 // Serialize read/compare/publish so slow older reads cannot publish with newer sequence numbers.
163 const refreshSemaphore = yield* Semaphore.make(1);
164 const watcherScope = yield* Scope.make("sequential");
165 yield* Effect.addFinalizer(() => Scope.close(watcherScope, Exit.void));
166
167 const refresh = refreshSemaphore.withPermits(1)(
168 Effect.gen(function* () {
169 const themes = yield* readPublishedThemes(environmentThemesDir).pipe(
170 Effect.provideService(FileSystem.FileSystem, fs),
171 );
172 // Compare the whole value so new fields cannot silently miss republishes.
173 const [changed, next] = yield* Ref.modify(
174 published,
175 (previous): readonly [readonly [boolean, PublishedThemes], PublishedThemes] => {
176 if (Equal.equals(previous.themes, themes)) return [[false, previous], previous];
177 const updated: PublishedThemes = { seq: previous.seq + 1, themes };
178 return [[true, updated], updated];
179 },
180 );
181 if (changed) yield* PubSub.publish(changes, next).pipe(Effect.asVoid);
182 return next;
183 }),
184 );
185
186 yield* fs
187 .makeDirectory(environmentThemesDir, { recursive: true })
188 .pipe(Effect.ignoreCause({ log: true }));
189
190 // A save emits several events, and fs.watch can fire before content is flushed.
191 const watchEvents = fs.watch(environmentThemesDir).pipe(Stream.debounce(Duration.millis(100)));
192
193 yield* refresh;
194 yield* Stream.runForEach(watchEvents, () => refresh.pipe(Effect.ignoreCause({ log: true }))).pipe(
195 Effect.ignoreCause({ log: true }),
196 Effect.forkIn(watcherScope),
197 Effect.asVoid,
198 );
199
200 return {
201 current: Effect.map(refresh, (state) => state.themes),
202 get streamChanges() {
203 return Stream.unwrap(
204 Effect.gen(function* () {
205 const subscription = yield* PubSub.subscribe(changes);
206 const snapshot = yield* refresh;
207 return Stream.concat(
208 Stream.make(snapshot.themes),
209 Stream.fromSubscription(subscription).pipe(
210 Stream.filter((update) => update.seq > snapshot.seq),
211 Stream.map((update) => update.themes),
212 ),
213 );
214 }),
215 );
216 },
217 } satisfies EnvironmentThemeService["Service"];
218});
219
220export const layer = Layer.effect(EnvironmentThemeService, make);