1import { describe, expect, it, vi } from "@effect/vitest";
2import {
3 EnvironmentId,
4 ThreadId,
5 AuthFilesystemReadScope,
6 EnvironmentAuthorizationError,
7 type FilesystemEntryMetadata,
8 type FilesystemGetMetadataInput,
9 WS_METHODS,
10} from "@t3tools/contracts";
11import * as Clock from "effect/Clock";
12import * as Effect from "effect/Effect";
13import * as Deferred from "effect/Deferred";
14import * as Fiber from "effect/Fiber";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Stream from "effect/Stream";
18import * as Schema from "effect/Schema";
19import * as SubscriptionRef from "effect/SubscriptionRef";
20import * as TestClock from "effect/testing/TestClock";
21import { Atom, AtomRegistry } from "effect/reactivity";
22
23import { AVAILABLE_CONNECTION_STATE, BearerConnectionTarget } from "../connection/model.ts";
24import * as EnvironmentRegistry from "../connection/registry.ts";
25import * as EnvironmentSupervisor from "../connection/supervisor.ts";
26import type { RpcSession } from "../rpc/session.ts";
27import { createFileMetadataAtoms } from "./fileMetadata.ts";
28
29const ENVIRONMENT_ID = EnvironmentId.make("metadata-environment");
30const OTHER_ENVIRONMENT_ID = EnvironmentId.make("metadata-other-environment");
31
32const makeHarness = Effect.fnUntraced(function* () {
33 const batches: string[][] = [];
34 const values = new Map<string, FilesystemEntryMetadata | null>();
35 const failure = { denied: false };
36 const response = { wait: Effect.void as Effect.Effect<void> };
37 const makeSession = () =>
38 ({
39 client: {
40 [WS_METHODS.filesystemGetMetadata]: (input: FilesystemGetMetadataInput) =>
41 Effect.gen(function* () {
42 batches.push([...input.paths]);
43 if (failure.denied)
44 return yield* new EnvironmentAuthorizationError({
45 requiredScope: AuthFilesystemReadScope,
46 message: "File access is not granted.",
47 });
48 const entries = input.paths.map((path) => values.get(path) ?? null);
49 yield* response.wait;
50 return { entries };
51 }),
52 },
53 }) as unknown as RpcSession;
54 const session = yield* SubscriptionRef.make(Option.some(makeSession()));
55 const state = yield* SubscriptionRef.make({
56 ...AVAILABLE_CONNECTION_STATE,
57 phase: "connected" as const,
58 });
59 const supervisor = {
60 target: new BearerConnectionTarget({
61 environmentId: ENVIRONMENT_ID,
62 label: "Test",
63 connectionId: "test",
64 }),
65 session,
66 state,
67 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"];
68 const otherSupervisor = {
69 ...supervisor,
70 target: new BearerConnectionTarget({
71 environmentId: OTHER_ENVIRONMENT_ID,
72 label: "Test",
73 connectionId: "test",
74 }),
75 session: yield* SubscriptionRef.make(Option.some(makeSession())),
76 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"];
77 const getSupervisor = (environmentId: EnvironmentId) =>
78 environmentId === OTHER_ENVIRONMENT_ID ? otherSupervisor : supervisor;
79 const environmentRegistry = {
80 run: (environmentId, effect) =>
81 Effect.provideService(
82 effect,
83 EnvironmentSupervisor.EnvironmentSupervisor,
84 getSupervisor(environmentId),
85 ),
86 followStream: (environmentId, stream) =>
87 Stream.provideService(
88 stream,
89 EnvironmentSupervisor.EnvironmentSupervisor,
90 getSupervisor(environmentId),
91 ),
92 } as EnvironmentRegistry.EnvironmentRegistry["Service"];
93 const clock = yield* Clock.Clock;
94 const runtime = Atom.runtime(
95 Layer.merge(
96 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
97 Layer.succeed(Clock.Clock, clock),
98 ),
99 );
100 const atoms = createFileMetadataAtoms(runtime);
101 const registry = AtomRegistry.make();
102 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
103 const atom = (path: string, environmentId = ENVIRONMENT_ID, threadId?: ThreadId) =>
104 atoms.metadata({ environmentId, input: { path, ...(threadId ? { threadId } : {}) } });
105 const read = (path: string, environmentId = ENVIRONMENT_ID, threadId?: ThreadId) =>
106 AtomRegistry.getResult(registry, atom(path, environmentId, threadId), {
107 suspendOnWaiting: true,
108 }).pipe(Effect.tap(() => Effect.yieldNow));
109 const seed = (path: string, kind: "file" | "directory") =>
110 atoms
111 .rememberEntries("/workspace", [{ path, kind }])
112 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
113 return {
114 batches,
115 values,
116 registry,
117 atom,
118 read,
119 seed,
120 session,
121 makeSession,
122 atoms,
123 supervisor,
124 failure,
125 response,
126 };
127});
128
129describe("file metadata", () => {
130 it.effect("refreshes expired mounted misses while retaining known thread entries", () =>
131 Effect.gen(function* () {
132 vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
133 yield* Effect.addFinalizer(() => Effect.sync(() => vi.useRealTimers()));
134 const h = yield* makeHarness();
135 const threadId = ThreadId.make("new-directory-thread");
136 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
137 const path = "/home/user/games/hl1";
138 h.registry.mount(h.atom(path, ENVIRONMENT_ID, threadId));
139 h.values.set("/workspace/known", { kind: "file", byteLength: 5 });
140 expect(yield* h.read(path, ENVIRONMENT_ID, threadId)).toBeNull();
141 yield* h.read("/workspace/known", ENVIRONMENT_ID, threadId);
142 h.values.set(path, { kind: "directory" });
143 h.registry.refresh(h.atom(path, ENVIRONMENT_ID, threadId));
144 expect(yield* h.read(path, ENVIRONMENT_ID, threadId)).toBeNull();
145 h.failure.denied = true;
146 yield* TestClock.adjust(60_001);
147 yield* Effect.promise(() => vi.advanceTimersByTimeAsync(60_000));
148 expect(yield* h.read(path, ENVIRONMENT_ID, threadId).pipe(Effect.flip)).toBeInstanceOf(
149 EnvironmentAuthorizationError,
150 );
151 h.failure.denied = false;
152 yield* TestClock.adjust(60_001);
153 yield* Effect.promise(() => vi.advanceTimersByTimeAsync(60_000));
154 expect(yield* h.read(path, ENVIRONMENT_ID, threadId)).toEqual({ kind: "directory" });
155 h.registry.refresh(h.atom("/workspace/known", ENVIRONMENT_ID, threadId));
156 expect(yield* h.read("/workspace/known", ENVIRONMENT_ID, threadId)).toEqual({
157 kind: "file",
158 byteLength: 5,
159 });
160 yield* TestClock.adjust(60_001);
161 yield* Effect.promise(() => vi.advanceTimersByTimeAsync(60_000));
162 expect(h.batches).toEqual([[path], ["/workspace/known"], [path], [path]]);
163 }),
164 );
165
166 it.effect("does not seed a different path from a literal name with a numeric suffix", () =>
167 Effect.gen(function* () {
168 const h = yield* makeHarness();
169 yield* h.seed("report:123", "directory");
170 h.values.set("/workspace/report", { kind: "file", byteLength: 5 });
171 expect(yield* h.read("/workspace/report")).toEqual({ kind: "file", byteLength: 5 });
172 expect(yield* h.read("/workspace/report:123")).toEqual({ kind: "directory" });
173 yield* h.atoms
174 .rememberFile("/workspace", {
175 relativePath: "/workspace/read:12:3",
176 byteLength: 100,
177 })
178 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
179 expect(yield* h.read("/workspace/read:12:3")).toEqual({ kind: "file", byteLength: 100 });
180 expect(yield* h.read("/workspace/read")).toBeNull();
181 expect(h.batches).toEqual([["/workspace/report"], ["/workspace/read"]]);
182 }),
183 );
184
185 it.effect("retries only paths invalidated while their response is in flight", () =>
186 Effect.gen(function* () {
187 const h = yield* makeHarness();
188 const threadId = ThreadId.make("in-flight-thread");
189 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
190 const started = yield* Deferred.make<void>();
191 const release = yield* Deferred.make<void>();
192 h.response.wait = Deferred.succeed(started, undefined).pipe(
193 Effect.andThen(Deferred.await(release)),
194 );
195 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
196 h.values.set("/workspace/unchanged", { kind: "directory" });
197 const read = yield* Effect.all(
198 [h.read("/workspace/file", ENVIRONMENT_ID, threadId), h.read("/workspace/unchanged")],
199 { concurrency: "unbounded" },
200 ).pipe(Effect.forkChild);
201 yield* Deferred.await(started);
202 yield* h.atoms
203 .invalidate("/workspace/file")
204 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
205 h.values.set("/workspace/file", { kind: "file", byteLength: 10 });
206 h.response.wait = Effect.void;
207 yield* Deferred.succeed(release, undefined);
208 expect(yield* Fiber.join(read)).toEqual([
209 { kind: "file", byteLength: 10 },
210 { kind: "directory" },
211 ]);
212 expect(h.batches).toEqual([["/workspace/file", "/workspace/unchanged"], ["/workspace/file"]]);
213 h.registry.refresh(h.atom("/workspace/file", ENVIRONMENT_ID, threadId));
214 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
215 kind: "file",
216 byteLength: 10,
217 });
218 expect(h.batches).toHaveLength(2);
219 }),
220 );
221
222 it.effect("an offscreen invalidation waits until that path is read again", () =>
223 Effect.gen(function* () {
224 const h = yield* makeHarness();
225 const threadId = ThreadId.make("offscreen-thread");
226 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
227 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
228 yield* h.atoms
229 .invalidate("/workspace/file")
230 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
231 h.atoms.refreshPath(ENVIRONMENT_ID, "/workspace/file", h.registry);
232 yield* Effect.yieldNow;
233 expect(h.batches).toHaveLength(1);
234 h.values.set("/workspace/file", { kind: "directory" });
235 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
236 kind: "directory",
237 });
238 expect(h.batches).toHaveLength(2);
239 }),
240 );
241
242 it.effect("invalidates a changed retained path without refetching other paths", () =>
243 Effect.gen(function* () {
244 const h = yield* makeHarness();
245 const threadId = ThreadId.make("changing-thread");
246 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
247 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
248 const file = h.atom("/workspace/file", ENVIRONMENT_ID, threadId);
249 h.registry.mount(file);
250 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
251 yield* h.read("/workspace/unchanged", ENVIRONMENT_ID, threadId);
252 const before = h.batches.length;
253 h.values.set("/workspace/file", { kind: "file", byteLength: 10 });
254 yield* h.atoms
255 .invalidate("/workspace/file")
256 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
257 h.atoms.refreshPath(ENVIRONMENT_ID, "/workspace/file", h.registry);
258 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
259 kind: "file",
260 byteLength: 10,
261 });
262 h.registry.refresh(h.atom("/workspace/unchanged", ENVIRONMENT_ID, threadId));
263 yield* h.read("/workspace/unchanged", ENVIRONMENT_ID, threadId);
264 expect(h.batches.slice(before)).toEqual([["/workspace/file"]]);
265 }),
266 );
267
268 it.effect("a reconnect discards retained metadata from the previous session", () =>
269 Effect.gen(function* () {
270 const h = yield* makeHarness();
271 const threadId = ThreadId.make("reconnected-thread");
272 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
273 h.values.set("/workspace/file", { kind: "file" });
274 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
275 h.values.set("/workspace/file", { kind: "directory" });
276 yield* SubscriptionRef.set(h.session, Option.some(h.makeSession()));
277 h.registry.refresh(h.atom("/workspace/file", ENVIRONMENT_ID, threadId));
278 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
279 kind: "directory",
280 });
281 expect(h.batches).toHaveLength(2);
282 }),
283 );
284
285 it.effect("keeps batches and cached paths separate across environments", () =>
286 Effect.gen(function* () {
287 const h = yield* makeHarness();
288 yield* h.seed("known", "directory");
289 expect(yield* h.read("/workspace/known", OTHER_ENVIRONMENT_ID)).toBeNull();
290 expect(h.batches).toEqual([["/workspace/known"]]);
291 yield* Effect.all(
292 [h.read("/workspace/new"), h.read("/workspace/new", OTHER_ENVIRONMENT_ID)],
293 { concurrency: "unbounded" },
294 );
295 expect(h.batches.slice(1)).toEqual([["/workspace/new"], ["/workspace/new"]]);
296 }),
297 );
298
299 it.effect("settles a failed batch and can retry after access is granted", () =>
300 Effect.gen(function* () {
301 const h = yield* makeHarness();
302 h.failure.denied = true;
303 const failures = yield* Effect.all(
304 [h.read("/workspace/a").pipe(Effect.flip), h.read("/workspace/b").pipe(Effect.flip)],
305 { concurrency: "unbounded" },
306 );
307 expect(failures.every(Schema.is(EnvironmentAuthorizationError))).toBe(true);
308 h.failure.denied = false;
309 h.values.set("/workspace/a", { kind: "directory" });
310 h.registry.refresh(h.atom("/workspace/a"));
311 expect(yield* h.read("/workspace/a")).toEqual({ kind: "directory" });
312 }),
313 );
314
315 it.effect("reuses file read sizes and normalizes Windows cache keys", () =>
316 Effect.gen(function* () {
317 const h = yield* makeHarness();
318 yield* h.atoms
319 .rememberFile("C:\\Workspace", {
320 relativePath: "file",
321 byteLength: 100,
322 })
323 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
324 expect(yield* h.read("c:/workspace/file/")).toEqual({ kind: "file", byteLength: 100 });
325 expect(h.batches).toHaveLength(0);
326 }),
327 );
328
329 it.effect("batches mounted paths and deduplicates repeated chips", () =>
330 Effect.gen(function* () {
331 const h = yield* makeHarness();
332 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
333 h.values.set("/workspace/folder.ts", { kind: "directory" });
334 const results = yield* Effect.all(
335 [h.read("/workspace/file"), h.read("/workspace/folder.ts"), h.read("/workspace/file")],
336 { concurrency: "unbounded" },
337 );
338 expect(h.batches).toEqual([["/workspace/file", "/workspace/folder.ts"]]);
339 expect(results).toEqual([
340 { kind: "file", byteLength: 5 },
341 { kind: "directory" },
342 { kind: "file", byteLength: 5 },
343 ]);
344 yield* h.read("/workspace/file");
345 expect(h.batches).toHaveLength(1);
346 }),
347 );
348
349 it.effect("uses search and tree kinds without another request", () =>
350 Effect.gen(function* () {
351 const h = yield* makeHarness();
352 yield* h.seed("folder.ts", "directory");
353 expect(yield* h.read("/workspace/folder.ts")).toEqual({ kind: "directory" });
354 expect(h.batches).toHaveLength(0);
355 }),
356 );
357
358 it.effect("splits large mounts into bounded batches and caches misses", () =>
359 Effect.gen(function* () {
360 const h = yield* makeHarness();
361 yield* Effect.all(
362 Array.from({ length: 130 }, (_, i) => h.read(`/workspace/${i}`)),
363 { concurrency: "unbounded" },
364 );
365 expect(h.batches.flat()).toHaveLength(130);
366 expect(h.batches.map((batch) => batch.length)).toEqual([64, 64, 2]);
367 expect(h.batches.every((batch) => batch.length <= 64)).toBe(true);
368 const count = h.batches.length;
369 yield* h.read("/workspace/0");
370 expect(h.batches).toHaveLength(count);
371 }),
372 );
373});