1import { describe, expect, it } 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 } 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: { environmentId: ENVIRONMENT_ID },
61 session,
62 state,
63 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"];
64 const otherSupervisor = {
65 ...supervisor,
66 target: { environmentId: OTHER_ENVIRONMENT_ID },
67 session: yield* SubscriptionRef.make(Option.some(makeSession())),
68 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"];
69 const getSupervisor = (environmentId: EnvironmentId) =>
70 environmentId === OTHER_ENVIRONMENT_ID ? otherSupervisor : supervisor;
71 const environmentRegistry = {
72 run: (environmentId, effect) =>
73 Effect.provideService(
74 effect,
75 EnvironmentSupervisor.EnvironmentSupervisor,
76 getSupervisor(environmentId),
77 ),
78 followStream: (environmentId, stream) =>
79 Stream.provideService(
80 stream,
81 EnvironmentSupervisor.EnvironmentSupervisor,
82 getSupervisor(environmentId),
83 ),
84 } as EnvironmentRegistry.EnvironmentRegistry["Service"];
85 const clock = yield* Clock.Clock;
86 const runtime = Atom.runtime(
87 Layer.merge(
88 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
89 Layer.succeed(Clock.Clock, clock),
90 ),
91 );
92 const atoms = createFileMetadataAtoms(runtime);
93 const registry = AtomRegistry.make();
94 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
95 const atom = (path: string, environmentId = ENVIRONMENT_ID, threadId?: ThreadId) =>
96 atoms.metadata({ environmentId, input: { path, ...(threadId ? { threadId } : {}) } });
97 const read = (path: string, environmentId = ENVIRONMENT_ID, threadId?: ThreadId) =>
98 AtomRegistry.getResult(registry, atom(path, environmentId, threadId), {
99 suspendOnWaiting: true,
100 }).pipe(Effect.tap(() => Effect.yieldNow));
101 const seed = (path: string, kind: "file" | "directory") =>
102 atoms
103 .rememberEntries("/workspace", [{ path, kind }])
104 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
105 return {
106 batches,
107 values,
108 registry,
109 atom,
110 read,
111 seed,
112 session,
113 makeSession,
114 atoms,
115 supervisor,
116 failure,
117 response,
118 };
119});
120
121describe("file metadata", () => {
122 it.effect("does not seed a different path from a literal name with a numeric suffix", () =>
123 Effect.gen(function* () {
124 const h = yield* makeHarness();
125 yield* h.seed("report:123", "directory");
126 h.values.set("/workspace/report", { kind: "file", byteLength: 5 });
127 expect(yield* h.read("/workspace/report")).toEqual({ kind: "file", byteLength: 5 });
128 expect(yield* h.read("/workspace/report:123")).toEqual({ kind: "directory" });
129 yield* h.atoms
130 .rememberFile("/workspace", {
131 relativePath: "/workspace/read:12:3",
132 byteLength: 100,
133 })
134 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
135 expect(yield* h.read("/workspace/read:12:3")).toEqual({ kind: "file", byteLength: 100 });
136 expect(yield* h.read("/workspace/read")).toBeNull();
137 expect(h.batches).toEqual([["/workspace/report"], ["/workspace/read"]]);
138 }).pipe(Effect.scoped),
139 );
140
141 it.effect("retries only paths invalidated while their response is in flight", () =>
142 Effect.gen(function* () {
143 const h = yield* makeHarness();
144 const threadId = ThreadId.make("in-flight-thread");
145 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
146 const started = yield* Deferred.make<void>();
147 const release = yield* Deferred.make<void>();
148 h.response.wait = Deferred.succeed(started, undefined).pipe(
149 Effect.andThen(Deferred.await(release)),
150 );
151 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
152 h.values.set("/workspace/unchanged", { kind: "directory" });
153 const read = yield* Effect.all(
154 [h.read("/workspace/file", ENVIRONMENT_ID, threadId), h.read("/workspace/unchanged")],
155 { concurrency: "unbounded" },
156 ).pipe(Effect.forkChild);
157 yield* Deferred.await(started);
158 yield* h.atoms
159 .invalidate("/workspace/file")
160 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
161 h.values.set("/workspace/file", { kind: "file", byteLength: 10 });
162 h.response.wait = Effect.void;
163 yield* Deferred.succeed(release, undefined);
164 expect(yield* Fiber.join(read)).toEqual([
165 { kind: "file", byteLength: 10 },
166 { kind: "directory" },
167 ]);
168 expect(h.batches).toEqual([["/workspace/file", "/workspace/unchanged"], ["/workspace/file"]]);
169 h.registry.refresh(h.atom("/workspace/file", ENVIRONMENT_ID, threadId));
170 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
171 kind: "file",
172 byteLength: 10,
173 });
174 expect(h.batches).toHaveLength(2);
175 }).pipe(Effect.scoped),
176 );
177
178 it.effect("does not reuse expired metadata retained by an unrelated thread", () =>
179 Effect.gen(function* () {
180 const h = yield* makeHarness();
181 const threadId = ThreadId.make("unrelated-thread");
182 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
183 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
184 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
185 yield* TestClock.adjust("1 hour");
186 h.values.set("/workspace/file", { kind: "directory" });
187 expect(yield* h.read("/workspace/file")).toEqual({ kind: "directory" });
188 expect(h.batches).toHaveLength(2);
189 }).pipe(Effect.scoped),
190 );
191
192 it.effect("uses a file listing kind without requesting optional metadata", () =>
193 Effect.gen(function* () {
194 const h = yield* makeHarness();
195 yield* h.seed("run", "file");
196 expect(yield* h.read("/workspace/run")).toEqual({ kind: "file" });
197 expect(h.batches).toHaveLength(0);
198 }).pipe(Effect.scoped),
199 );
200
201 it.effect("tree refreshes preserve MIME hints retained by an open thread", () =>
202 Effect.gen(function* () {
203 const h = yield* makeHarness();
204 const threadId = ThreadId.make("rich-metadata-thread");
205 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
206 const value = { kind: "file" as const, byteLength: 5, mimeType: "text/x-python" };
207 h.values.set("/workspace/run", value);
208 yield* h.read("/workspace/run", ENVIRONMENT_ID, threadId);
209 yield* TestClock.adjust("1 hour");
210 yield* h.seed("run", "file");
211 h.registry.refresh(h.atom("/workspace/run", ENVIRONMENT_ID, threadId));
212 expect(yield* h.read("/workspace/run", ENVIRONMENT_ID, threadId)).toEqual(value);
213 expect(h.batches).toHaveLength(1);
214 }).pipe(Effect.scoped),
215 );
216
217 it.effect("an offscreen invalidation waits until that path is read again", () =>
218 Effect.gen(function* () {
219 const h = yield* makeHarness();
220 const threadId = ThreadId.make("offscreen-thread");
221 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
222 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
223 yield* h.atoms
224 .invalidate("/workspace/file")
225 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
226 h.atoms.refreshPath(ENVIRONMENT_ID, "/workspace/file", h.registry);
227 yield* Effect.yieldNow;
228 expect(h.batches).toHaveLength(1);
229 h.values.set("/workspace/file", { kind: "directory" });
230 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
231 kind: "directory",
232 });
233 expect(h.batches).toHaveLength(2);
234 }).pipe(Effect.scoped),
235 );
236
237 it.effect("retains more than 1024 visited paths until the thread closes", () =>
238 Effect.gen(function* () {
239 const h = yield* makeHarness();
240 const threadId = ThreadId.make("large-thread");
241 const release = h.registry.mount(
242 h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }),
243 );
244 const paths = Array.from({ length: 1300 }, (_, i) => `/workspace/visited-${i}`);
245 yield* Effect.forEach(paths, (path) => h.read(path, ENVIRONMENT_ID, threadId), {
246 concurrency: "unbounded",
247 });
248 expect(h.batches).toHaveLength(21);
249 yield* TestClock.adjust("1 hour");
250 for (const path of paths) h.registry.refresh(h.atom(path, ENVIRONMENT_ID, threadId));
251 yield* Effect.forEach(paths, (path) => h.read(path, ENVIRONMENT_ID, threadId), {
252 concurrency: "unbounded",
253 });
254 expect(h.batches).toHaveLength(21);
255 release();
256 yield* Effect.yieldNow;
257 h.registry.refresh(h.atom(paths[0]!, ENVIRONMENT_ID, threadId));
258 yield* h.read(paths[0]!, ENVIRONMENT_ID, threadId);
259 expect(h.batches).toHaveLength(22);
260 }).pipe(Effect.scoped),
261 );
262
263 it.effect("closing one thread does not release another thread's retained paths", () =>
264 Effect.gen(function* () {
265 const h = yield* makeHarness();
266 const a = ThreadId.make("thread-a");
267 const b = ThreadId.make("thread-b");
268 const releaseA = h.registry.mount(
269 h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId: a }),
270 );
271 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId: b }));
272 yield* h.read("/workspace/shared", ENVIRONMENT_ID, a);
273 yield* h.read("/workspace/shared", ENVIRONMENT_ID, b);
274 expect(h.batches).toHaveLength(1);
275 releaseA();
276 yield* Effect.yieldNow;
277 yield* TestClock.adjust("1 hour");
278 h.registry.refresh(h.atom("/workspace/shared", ENVIRONMENT_ID, b));
279 yield* h.read("/workspace/shared", ENVIRONMENT_ID, b);
280 expect(h.batches).toHaveLength(1);
281 }).pipe(Effect.scoped),
282 );
283
284 it.effect("invalidates a changed retained path without refetching other paths", () =>
285 Effect.gen(function* () {
286 const h = yield* makeHarness();
287 const threadId = ThreadId.make("changing-thread");
288 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
289 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
290 const file = h.atom("/workspace/file", ENVIRONMENT_ID, threadId);
291 h.registry.mount(file);
292 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
293 yield* h.read("/workspace/unchanged", ENVIRONMENT_ID, threadId);
294 const before = h.batches.length;
295 h.values.set("/workspace/file", { kind: "file", byteLength: 10 });
296 yield* h.atoms
297 .invalidate("/workspace/file")
298 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
299 h.atoms.refreshPath(ENVIRONMENT_ID, "/workspace/file", h.registry);
300 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
301 kind: "file",
302 byteLength: 10,
303 });
304 h.registry.refresh(h.atom("/workspace/unchanged", ENVIRONMENT_ID, threadId));
305 yield* h.read("/workspace/unchanged", ENVIRONMENT_ID, threadId);
306 expect(h.batches.slice(before)).toEqual([["/workspace/file"]]);
307 }).pipe(Effect.scoped),
308 );
309
310 it.effect("a reconnect discards retained metadata from the previous session", () =>
311 Effect.gen(function* () {
312 const h = yield* makeHarness();
313 const threadId = ThreadId.make("reconnected-thread");
314 h.registry.mount(h.atoms.retainThread({ environmentId: ENVIRONMENT_ID, threadId }));
315 h.values.set("/workspace/file", { kind: "file" });
316 yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId);
317 h.values.set("/workspace/file", { kind: "directory" });
318 yield* SubscriptionRef.set(h.session, Option.some(h.makeSession()));
319 h.registry.refresh(h.atom("/workspace/file", ENVIRONMENT_ID, threadId));
320 expect(yield* h.read("/workspace/file", ENVIRONMENT_ID, threadId)).toEqual({
321 kind: "directory",
322 });
323 expect(h.batches).toHaveLength(2);
324 }).pipe(Effect.scoped),
325 );
326
327 it.effect("keeps batches and cached paths separate across environments", () =>
328 Effect.gen(function* () {
329 const h = yield* makeHarness();
330 yield* h.seed("known", "directory");
331 expect(yield* h.read("/workspace/known", OTHER_ENVIRONMENT_ID)).toBeNull();
332 expect(h.batches).toEqual([["/workspace/known"]]);
333 yield* Effect.all(
334 [h.read("/workspace/new"), h.read("/workspace/new", OTHER_ENVIRONMENT_ID)],
335 { concurrency: "unbounded" },
336 );
337 expect(h.batches.slice(1)).toEqual([["/workspace/new"], ["/workspace/new"]]);
338 }).pipe(Effect.scoped),
339 );
340
341 it.effect("settles a failed batch and can retry after access is granted", () =>
342 Effect.gen(function* () {
343 const h = yield* makeHarness();
344 h.failure.denied = true;
345 const failures = yield* Effect.all(
346 [h.read("/workspace/a").pipe(Effect.flip), h.read("/workspace/b").pipe(Effect.flip)],
347 { concurrency: "unbounded" },
348 );
349 expect(failures.every(Schema.is(EnvironmentAuthorizationError))).toBe(true);
350 h.failure.denied = false;
351 h.values.set("/workspace/a", { kind: "directory" });
352 h.registry.refresh(h.atom("/workspace/a"));
353 expect(yield* h.read("/workspace/a")).toEqual({ kind: "directory" });
354 }).pipe(Effect.scoped),
355 );
356
357 it.effect("expires cached metadata and refreshes a changed kind", () =>
358 Effect.gen(function* () {
359 const h = yield* makeHarness();
360 h.values.set("/workspace/path", { kind: "file" });
361 yield* h.read("/workspace/path");
362 h.values.set("/workspace/path", { kind: "directory" });
363 yield* TestClock.adjust("61 seconds");
364 h.registry.refresh(h.atom("/workspace/path"));
365 expect(yield* h.read("/workspace/path")).toEqual({ kind: "directory" });
366 expect(h.batches).toHaveLength(2);
367 }).pipe(Effect.scoped),
368 );
369
370 it.effect("reuses file read sizes and normalizes Windows cache keys", () =>
371 Effect.gen(function* () {
372 const h = yield* makeHarness();
373 yield* h.atoms
374 .rememberFile("C:\\Workspace", {
375 relativePath: "file",
376 byteLength: 100,
377 })
378 .pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, h.supervisor));
379 expect(yield* h.read("c:/workspace/file/")).toEqual({ kind: "file", byteLength: 100 });
380 expect(h.batches).toHaveLength(0);
381 }).pipe(Effect.scoped),
382 );
383
384 it.effect("batches mounted paths and deduplicates repeated chips", () =>
385 Effect.gen(function* () {
386 const h = yield* makeHarness();
387 h.values.set("/workspace/file", { kind: "file", byteLength: 5 });
388 h.values.set("/workspace/folder.ts", { kind: "directory" });
389 const results = yield* Effect.all(
390 [h.read("/workspace/file"), h.read("/workspace/folder.ts"), h.read("/workspace/file")],
391 { concurrency: "unbounded" },
392 );
393 expect(h.batches).toEqual([["/workspace/file", "/workspace/folder.ts"]]);
394 expect(results).toEqual([
395 { kind: "file", byteLength: 5 },
396 { kind: "directory" },
397 { kind: "file", byteLength: 5 },
398 ]);
399 yield* h.read("/workspace/file");
400 expect(h.batches).toHaveLength(1);
401 }).pipe(Effect.scoped),
402 );
403
404 it.effect("uses search and tree kinds without another request", () =>
405 Effect.gen(function* () {
406 const h = yield* makeHarness();
407 yield* h.seed("folder.ts", "directory");
408 expect(yield* h.read("/workspace/folder.ts")).toEqual({ kind: "directory" });
409 expect(h.batches).toHaveLength(0);
410 }).pipe(Effect.scoped),
411 );
412
413 it.effect("splits large mounts into bounded batches and caches misses", () =>
414 Effect.gen(function* () {
415 const h = yield* makeHarness();
416 yield* Effect.all(
417 Array.from({ length: 130 }, (_, i) => h.read(`/workspace/${i}`)),
418 { concurrency: "unbounded" },
419 );
420 expect(h.batches.flat()).toHaveLength(130);
421 expect(h.batches.map((batch) => batch.length)).toEqual([64, 64, 2]);
422 expect(h.batches.every((batch) => batch.length <= 64)).toBe(true);
423 const count = h.batches.length;
424 yield* h.read("/workspace/0");
425 expect(h.batches).toHaveLength(count);
426 }).pipe(Effect.scoped),
427 );
428
429 it.effect("batches synchronous mounts after the runtime is warm", () =>
430 Effect.gen(function* () {
431 const h = yield* makeHarness();
432 yield* h.read("/workspace/warmup");
433 h.batches.length = 0;
434 const paths = Array.from({ length: 130 }, (_, i) => `/workspace/mounted-${i}`);
435 yield* Effect.sync(() => {
436 for (const path of paths) h.registry.get(h.atom(path));
437 });
438 yield* Effect.all(
439 paths.map((path) => h.read(path)),
440 { concurrency: "unbounded" },
441 );
442 expect(h.batches.map((batch) => batch.length)).toEqual([64, 64, 2]);
443 }).pipe(Effect.scoped),
444 );
445
446 it.effect("rechecks metadata after the environment session changes", () =>
447 Effect.gen(function* () {
448 const h = yield* makeHarness();
449 h.values.set("/workspace/path", { kind: "file" });
450 expect(yield* h.read("/workspace/path")).toEqual({ kind: "file" });
451 h.values.set("/workspace/path", { kind: "directory" });
452 yield* SubscriptionRef.set(h.session, Option.some(h.makeSession()));
453 h.registry.refresh(h.atom("/workspace/path"));
454 expect(yield* h.read("/workspace/path")).toEqual({ kind: "directory" });
455 expect(h.batches).toHaveLength(2);
456 }).pipe(Effect.scoped),
457 );
458});