1import { describe, expect, it } from "@effect/vitest";
2import {
3 type AssetCreateUrlResult,
4 type AssetResource,
5 AssetWorkspaceAssetNotFoundError,
6 EnvironmentId,
7 ThreadId,
8 WS_METHODS,
9} from "@t3tools/contracts";
10import * as Cause from "effect/Cause";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Queue from "effect/Queue";
15import * as Stream from "effect/Stream";
16import * as SubscriptionRef from "effect/SubscriptionRef";
17import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
18
19import {
20 AVAILABLE_CONNECTION_STATE,
21 type PreparedConnection,
22 PrimaryConnectionTarget,
23 type SupervisorConnectionState,
24} from "../connection/model.ts";
25import * as EnvironmentRegistry from "../connection/registry.ts";
26import * as EnvironmentSupervisor from "../connection/supervisor.ts";
27import { resolveMediaSource } from "../mediaSource.ts";
28import type * as RpcSession from "../rpc/session.ts";
29import {
30 ASSET_URL_MAX_IN_FLIGHT,
31 createAssetEnvironmentAtoms,
32 createProjectFaviconUrlAtomFamily,
33 makeRevisionedAssetUrlFamily,
34} from "./assets.ts";
35
36describe("project favicon URL cache", () => {
37 it("does not reuse another environment, workspace, or selected icon's cached URL", () => {
38 const registry = AtomRegistry.make();
39 const result = Atom.make<AsyncResult.AsyncResult<AssetCreateUrlResult, unknown>>(
40 AsyncResult.success({
41 expiresAt: 4_000_000_000_000,
42 relativeUrl: "/api/assets/token/icon.svg",
43 }),
44 );
45 const favicon = createProjectFaviconUrlAtomFamily({
46 createUrl: () => result,
47 preparedConnection: () => Atom.make(Option.some({ httpBaseUrl: "https://remote.test" })),
48 });
49 const target = { environmentId: EnvironmentId.make("remote"), cwd: "/workspace" };
50 const unmount = registry.mount(favicon(target));
51 try {
52 expect(registry.get(favicon(target))).toBe("https://remote.test/api/assets/token/icon.svg");
53 registry.set(result, AsyncResult.failure(Cause.die("disconnected")));
54 expect(
55 registry.get(favicon({ ...target, environmentId: EnvironmentId.make("other") })),
56 ).toBeNull();
57 expect(registry.get(favicon({ ...target, cwd: "/other" }))).toBeNull();
58 expect(registry.get(favicon({ ...target, faviconPath: "brand.svg" }))).toBeNull();
59 expect(registry.get(favicon({ ...target, faviconPath: null }))).toBe(
60 "https://remote.test/api/assets/token/icon.svg",
61 );
62 } finally {
63 unmount();
64 registry.dispose();
65 }
66 });
67});
68
69/**
70 * Environments whose `assets.createUrl` never answers until the test releases a call.
71 * `started` receives the environment of each call as it begins, and `inFlight` counts
72 * calls that have begun and not yet been released or interrupted.
73 */
74const makeAssetUrlHarness = Effect.fn("TestAssetUrl.makeHarness")(function* (
75 environmentIds: ReadonlyArray<EnvironmentId>,
76) {
77 const started = yield* Queue.unbounded<EnvironmentId>();
78 const inFlight = new Map<EnvironmentId, number>();
79 const maxInFlight = new Map<EnvironmentId, number>();
80 const releases = new Map<EnvironmentId, Queue.Queue<void>>();
81 const supervisors = new Map<
82 EnvironmentId,
83 EnvironmentSupervisor.EnvironmentSupervisor["Service"]
84 >();
85 const sessions = new Map<
86 EnvironmentId,
87 SubscriptionRef.SubscriptionRef<Option.Option<RpcSession.RpcSession>>
88 >();
89 const makeSession = (environmentId: EnvironmentId) =>
90 ({
91 client: {
92 [WS_METHODS.assetsCreateUrl]: () =>
93 Effect.gen(function* () {
94 const now = (inFlight.get(environmentId) ?? 0) + 1;
95 inFlight.set(environmentId, now);
96 maxInFlight.set(environmentId, Math.max(maxInFlight.get(environmentId) ?? 0, now));
97 yield* Queue.offer(started, environmentId);
98 yield* Queue.take(releases.get(environmentId)!);
99 return { relativeUrl: "/api/assets/token/image.png", expiresAt: 4_000_000_000_000 };
100 }).pipe(
101 Effect.ensuring(
102 Effect.sync(() =>
103 inFlight.set(environmentId, (inFlight.get(environmentId) ?? 1) - 1),
104 ),
105 ),
106 ),
107 },
108 }) as unknown as RpcSession.RpcSession;
109 for (const environmentId of environmentIds) {
110 releases.set(environmentId, yield* Queue.unbounded<void>());
111 const session = yield* SubscriptionRef.make(Option.some(makeSession(environmentId)));
112 sessions.set(environmentId, session);
113 supervisors.set(
114 environmentId,
115 EnvironmentSupervisor.EnvironmentSupervisor.of({
116 target: new PrimaryConnectionTarget({
117 environmentId,
118 label: environmentId,
119 httpBaseUrl: "https://assets.example.test",
120 wsBaseUrl: "wss://assets.example.test",
121 }),
122 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
123 ...AVAILABLE_CONNECTION_STATE,
124 desired: true,
125 network: "online",
126 phase: "connected",
127 attempt: 1,
128 generation: 1,
129 }),
130 session,
131 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()),
132 connect: Effect.void,
133 disconnect: Effect.void,
134 retryNow: Effect.void,
135 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]),
136 );
137 }
138 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (environmentId, effect) =>
139 Effect.provideService(
140 effect,
141 EnvironmentSupervisor.EnvironmentSupervisor,
142 supervisors.get(environmentId)!,
143 );
144 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
145 environmentId,
146 stream,
147 ) =>
148 Stream.provideService(
149 stream,
150 EnvironmentSupervisor.EnvironmentSupervisor,
151 supervisors.get(environmentId)!,
152 );
153 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
154 run,
155 followStream,
156 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
157 const registry = AtomRegistry.make();
158 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
159 const atoms = createAssetEnvironmentAtoms(
160 Atom.runtime(Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry)),
161 );
162 return {
163 started,
164 inFlight: (environmentId: EnvironmentId) => inFlight.get(environmentId) ?? 0,
165 maxInFlight: (environmentId: EnvironmentId) => maxInFlight.get(environmentId) ?? 0,
166 release: (environmentId: EnvironmentId) => Queue.offer(releases.get(environmentId)!, undefined),
167 replaceSession: (environmentId: EnvironmentId) =>
168 SubscriptionRef.set(sessions.get(environmentId)!, Option.some(makeSession(environmentId))),
169 mountImages: (environmentId: EnvironmentId, count: number) => {
170 for (let index = 0; index < count; index += 1) {
171 registry.mount(
172 atoms.createUrl({
173 environmentId,
174 input: { resource: { _tag: "attachment", attachmentId: `image-${index}` } },
175 }),
176 );
177 }
178 },
179 };
180});
181
182describe("asset URL requests", () => {
183 const remote = EnvironmentId.make("asset-remote");
184 const other = EnvironmentId.make("asset-other");
185
186 it.effect("keeps a thread of images from flooding one environment with requests", () =>
187 Effect.gen(function* () {
188 const harness = yield* makeAssetUrlHarness([remote]);
189 const total = ASSET_URL_MAX_IN_FLIGHT * 3;
190 harness.mountImages(remote, total);
191
192 for (let call = 0; call < ASSET_URL_MAX_IN_FLIGHT; call += 1) {
193 yield* Queue.take(harness.started);
194 }
195 expect(harness.inFlight(remote)).toBe(ASSET_URL_MAX_IN_FLIGHT);
196
197 // Each answer admits the next waiting request.
198 for (let call = 0; call < total; call += 1) {
199 yield* harness.release(remote);
200 if (call < total - ASSET_URL_MAX_IN_FLIGHT) yield* Queue.take(harness.started);
201 }
202 expect(harness.maxInFlight(remote)).toBe(ASSET_URL_MAX_IN_FLIGHT);
203 }),
204 );
205
206 it.effect("limits each environment on its own", () =>
207 Effect.gen(function* () {
208 const harness = yield* makeAssetUrlHarness([remote, other]);
209 harness.mountImages(remote, ASSET_URL_MAX_IN_FLIGHT + 2);
210 for (let call = 0; call < ASSET_URL_MAX_IN_FLIGHT; call += 1) {
211 yield* Queue.take(harness.started);
212 }
213 harness.mountImages(other, 1);
214 expect(yield* Queue.take(harness.started)).toBe(other);
215 expect(harness.inFlight(remote)).toBe(ASSET_URL_MAX_IN_FLIGHT);
216 }),
217 );
218
219 it.effect("frees the limit when the session behind the requests is replaced", () =>
220 Effect.gen(function* () {
221 const harness = yield* makeAssetUrlHarness([remote]);
222 harness.mountImages(remote, ASSET_URL_MAX_IN_FLIGHT * 2);
223 for (let call = 0; call < ASSET_URL_MAX_IN_FLIGHT; call += 1) {
224 yield* Queue.take(harness.started);
225 }
226
227 yield* harness.replaceSession(remote);
228 for (let call = 0; call < ASSET_URL_MAX_IN_FLIGHT; call += 1) {
229 yield* Queue.take(harness.started);
230 }
231 expect(harness.maxInFlight(remote)).toBe(ASSET_URL_MAX_IN_FLIGHT);
232 }),
233 );
234});
235
236describe("desktop-local media fallback", () => {
237 const remote = EnvironmentId.make("media-remote");
238 const local = EnvironmentId.make("media-local");
239 const threadId = ThreadId.make("remote-thread");
240
241 // The remote is missing every file; the desktop has one at every absolute path.
242 const makeFallback = Effect.fn("TestAssetUrl.makeFallback")(function* () {
243 const calls: Array<EnvironmentId> = [];
244 const supervisor = (environmentId: EnvironmentId) =>
245 Effect.gen(function* () {
246 const session = {
247 client: {
248 [WS_METHODS.assetsCreateUrl]: ({ resource }: { resource: AssetResource }) =>
249 Effect.suspend(() => {
250 calls.push(environmentId);
251 return environmentId === remote
252 ? Effect.fail(new AssetWorkspaceAssetNotFoundError({ resource }))
253 : Effect.succeed({ relativeUrl: "/api/assets/t/local.png", expiresAt: 4e12 });
254 }),
255 },
256 } as unknown as RpcSession.RpcSession;
257 return EnvironmentSupervisor.EnvironmentSupervisor.of({
258 target: new PrimaryConnectionTarget({
259 environmentId,
260 label: environmentId,
261 httpBaseUrl: `https://${environmentId}.test`,
262 wsBaseUrl: `wss://${environmentId}.test`,
263 }),
264 state: yield* SubscriptionRef.make<SupervisorConnectionState>({
265 ...AVAILABLE_CONNECTION_STATE,
266 desired: true,
267 network: "online",
268 phase: "connected",
269 attempt: 1,
270 generation: 1,
271 }),
272 session: yield* SubscriptionRef.make(Option.some(session)),
273 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()),
274 connect: Effect.void,
275 disconnect: Effect.void,
276 retryNow: Effect.void,
277 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
278 });
279 const supervisors = new Map([
280 [remote, yield* supervisor(remote)],
281 [local, yield* supervisor(local)],
282 ]);
283 type Registry = EnvironmentRegistry.EnvironmentRegistry["Service"];
284 const run: Registry["run"] = (environmentId, effect) =>
285 Effect.provideService(
286 effect,
287 EnvironmentSupervisor.EnvironmentSupervisor,
288 supervisors.get(environmentId)!,
289 );
290 const followStream: Registry["followStream"] = (environmentId, stream) =>
291 Stream.provideService(
292 stream,
293 EnvironmentSupervisor.EnvironmentSupervisor,
294 supervisors.get(environmentId)!,
295 );
296 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
297 run,
298 followStream,
299 } as unknown as Registry);
300 const registry = AtomRegistry.make();
301 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
302 const atoms = createAssetEnvironmentAtoms(
303 Atom.runtime(Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry)),
304 Atom.make({ environmentId: local, httpBaseUrl: "https://media-local.test" }),
305 );
306 const load = (source: string) => {
307 const media = resolveMediaSource(source, { threadId, workspaceRoot: "/tmp/repo" });
308 if (media?.access !== "environment") throw new Error(`${source} is not host media`);
309 calls.length = 0;
310 return AtomRegistry.getResult(
311 registry,
312 atoms.createUrl({ environmentId: remote, input: { resource: media.resource } }),
313 ).pipe(
314 Effect.result,
315 Effect.map((result) => ({ loaded: result._tag === "Success", calls: [...calls] })),
316 );
317 };
318 return { load };
319 });
320
321 it.effect("keeps a relative Markdown image on the thread's environment", () =>
322 Effect.gen(function* () {
323 const { load } = yield* makeFallback();
324 expect(yield* load("docs/report.png")).toEqual({ loaded: false, calls: [remote] });
325 // The same resolved path written absolute still falls back, and shares no cached result.
326 expect(yield* load("/tmp/repo/docs/report.png")).toEqual({
327 loaded: true,
328 calls: [remote, local],
329 });
330 }),
331 );
332});
333
334const revisionedEnvironmentId = EnvironmentId.make("environment-1");
335const revisionedResource: AssetResource = {
336 _tag: "workspace-file",
337 threadId: ThreadId.make("thread-1"),
338 path: "docs/diagram.png",
339};
340
341type MintResult = AsyncResult.AsyncResult<{
342 relativeUrl: string;
343 expiresAt: number;
344 fileRevision?: string;
345}>;
346
347const minted = (n: number, fileRevision?: string): MintResult =>
348 AsyncResult.success({
349 relativeUrl: `/a/${n}`,
350 expiresAt: 60 * 60_000,
351 ...(fileRevision === undefined ? {} : { fileRevision }),
352 });
353
354// A stand-in for the asset URL query: a refresh starts a mint (counted, shown as
355// waiting) and `settle` completes the latest one.
356function setup(initial: MintResult) {
357 let mints = 0;
358 // Node removal runs on the registry's scheduler; `unmount` flushes it so the
359 // next mount of the same revision builds a fresh node.
360 const scheduled: Array<() => void> = [];
361 const registry = AtomRegistry.make({
362 scheduleTask: (task) => {
363 scheduled.push(task);
364 return () => {};
365 },
366 });
367 const server = Atom.make(initial).pipe(Atom.keepAlive);
368 const query = Atom.readable(
369 (get) => get(server),
370 () => {
371 mints += 1;
372 registry.set(server, AsyncResult.waiting(registry.get(server)));
373 },
374 ).pipe(Atom.keepAlive);
375 const revisioned = makeRevisionedAssetUrlFamily(
376 () => query,
377 () => 0,
378 );
379 const atom = (revision: string | null) =>
380 revisioned({
381 environmentId: revisionedEnvironmentId,
382 input: { resource: revisionedResource },
383 revision,
384 });
385 const mount = (revision: string | null) => {
386 const release = registry.mount(atom(revision));
387 return () => {
388 release();
389 for (const task of scheduled.splice(0)) task();
390 };
391 };
392 const settle = (fileRevision?: string) => registry.set(server, minted(mints + 1, fileRevision));
393 const url = (revision: string | null) => {
394 const result = registry.get(atom(revision));
395 return result._tag === "Success" ? result.value.relativeUrl : null;
396 };
397 return { mount, settle, url, mints: () => mints };
398}
399
400describe("makeRevisionedAssetUrlFamily", () => {
401 it("re-mints a settled URL on first read and on each new revision, not on remount", () => {
402 const { mount, settle, mints } = setup(minted(0));
403
404 mount("mutation-1")();
405 expect(mints()).toBe(1);
406 settle();
407
408 mount("mutation-1")();
409 expect(mints()).toBe(1);
410
411 mount("mutation-2")();
412 expect(mints()).toBe(2);
413 });
414
415 it("re-mints when a revert returns to an earlier revision", () => {
416 const { mount, settle, mints } = setup(minted(0));
417
418 mount("mutation-a")();
419 settle();
420 mount("mutation-b")();
421 settle();
422 expect(mints()).toBe(2);
423
424 const reverted = mount("mutation-a");
425 expect(mints()).toBe(3);
426 reverted();
427 });
428
429 it("shares a mint in flight on the first read", () => {
430 const { mount, mints } = setup(AsyncResult.initial(true));
431
432 mount("mutation-1")();
433 expect(mints()).toBe(0);
434 });
435
436 it("re-mints after a mint that was in flight when the revision changed", () => {
437 const { mount, settle, mints } = setup(minted(0));
438
439 const before = mount("mutation-1");
440 settle();
441 const edited = mount("mutation-2");
442 expect(mints()).toBe(2);
443 before();
444
445 const stillMinting = mount("mutation-3");
446 edited();
447 expect(mints()).toBe(2);
448
449 settle();
450 expect(mints()).toBe(3);
451 settle();
452 expect(mints()).toBe(3);
453 stillMinting();
454 });
455
456 it("keeps the URL while the file is unchanged and swaps it once the file changes", () => {
457 const { mount, settle, url } = setup(minted(0, "file-a"));
458
459 const first = mount("mutation-1");
460 settle("file-a");
461 first();
462
463 const unchanged = mount("mutation-2");
464 settle("file-a");
465 expect(url("mutation-2")).toBe("/a/0");
466 unchanged();
467
468 // The query now holds /a/3; a later revision must still keep the URL on screen.
469 const stillUnchanged = mount("mutation-3");
470 expect(url("mutation-3")).toBe("/a/0");
471 settle("file-a");
472 expect(url("mutation-3")).toBe("/a/0");
473 stillUnchanged();
474
475 const edited = mount("mutation-4");
476 settle("file-b");
477 expect(url("mutation-4")).toBe("/a/5");
478 edited();
479 });
480});