1import { describe, expect, it } from "@effect/vitest";
2import {
3 AuthFilesystemWriteScope,
4 EnvironmentId,
5 ProjectId,
6 ThreadId,
7 type AuthSessionState,
8 type ScopedProjectRef,
9 WS_METHODS,
10} from "@t3tools/contracts";
11import { vi } from "vite-plus/test";
12import * as Crypto from "effect/Crypto";
13import * as Effect from "effect/Effect";
14import * as Deferred from "effect/Deferred";
15import * as Fiber from "effect/Fiber";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import * as SubscriptionRef from "effect/SubscriptionRef";
19import * as Stream from "effect/Stream";
20import { Atom, AtomRegistry, AsyncResult } from "effect/reactivity";
21import { AVAILABLE_CONNECTION_STATE } from "../connection/model.ts";
22
23import * as EnvironmentRegistry from "../connection/registry.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import type { RpcSession } from "../rpc/session.ts";
26import type { EnvironmentProject } from "./models.ts";
27import { createProjectEnvironmentAtoms } from "./projectCommands.ts";
28
29const ENVIRONMENT_ID = EnvironmentId.make("environment-1");
30vi.mock("./session.ts", () => ({
31 createEnvironmentSessionAtoms: () => ({ sessionStateAtom: sessions }),
32}));
33const sessions = Atom.family((_id: EnvironmentId) =>
34 Atom.make<AsyncResult.AsyncResult<AuthSessionState>>(AsyncResult.initial()),
35);
36const PROJECT_ID = ProjectId.make("scratch");
37const PROJECT = {
38 id: PROJECT_ID,
39 environmentId: ENVIRONMENT_ID,
40 workspaceRoot: "/scratch",
41} as EnvironmentProject;
42
43const makeHarness = Effect.fn("TestProjectCommands.makeHarness")(function* () {
44 const metadataRequests: string[][] = [];
45 const file = { byteLength: 5 };
46 const fileChanges = yield* SubscriptionRef.make(0);
47 const readResponse = { wait: Effect.void as Effect.Effect<void> };
48 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
49 target: { environmentId: ENVIRONMENT_ID },
50 session: yield* SubscriptionRef.make(
51 Option.some({
52 client: {
53 [WS_METHODS.projectsEnsureScratch]: () => Effect.succeed({ projectId: PROJECT_ID }),
54 [WS_METHODS.projectsWatch]: () => SubscriptionRef.changes(fileChanges),
55 [WS_METHODS.projectsReadFile]: () =>
56 Effect.gen(function* () {
57 const byteLength = file.byteLength;
58 yield* readResponse.wait;
59 return { relativePath: "foo", byteLength, contents: "old", truncated: false };
60 }),
61 [WS_METHODS.projectsWriteFile]: () =>
62 Effect.sync(() => {
63 file.byteLength = 10;
64 return { relativePath: "foo" };
65 }),
66 [WS_METHODS.filesystemGetMetadata]: (input: { paths: string[] }) =>
67 Effect.sync(() => {
68 metadataRequests.push(input.paths);
69 return {
70 entries: input.paths.map(() => ({
71 kind: "file" as const,
72 byteLength: file.byteLength,
73 })),
74 };
75 }),
76 },
77 } as unknown as RpcSession),
78 ),
79 state: yield* SubscriptionRef.make({
80 ...AVAILABLE_CONNECTION_STATE,
81 phase: "connected" as const,
82 }),
83 } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
84 const runtime = Atom.runtime(
85 Layer.mergeAll(
86 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, {
87 run: (_environmentId, effect) =>
88 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
89 followStream: (_environmentId, stream) =>
90 Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
91 } as EnvironmentRegistry.EnvironmentRegistry["Service"]),
92 Layer.succeed(
93 Crypto.Crypto,
94 Crypto.make({
95 randomBytes: (size) => new Uint8Array(size),
96 digest: (_algorithm, data) => Effect.succeed(data),
97 }),
98 ),
99 ),
100 );
101 const storedProject = Atom.make<EnvironmentProject | null>(null);
102 const commands = createProjectEnvironmentAtoms(runtime, {
103 projectAtom: (ref: ScopedProjectRef) =>
104 ref.environmentId === ENVIRONMENT_ID && ref.projectId === PROJECT_ID
105 ? storedProject
106 : Atom.make(null),
107 });
108 const registry = AtomRegistry.make();
109 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
110 registry.set(
111 sessions(ENVIRONMENT_ID),
112 AsyncResult.success({
113 authenticated: true,
114 auth: {
115 policy: "remote-reachable" as const,
116 bootstrapMethods: [],
117 sessionMethods: [],
118 sessionCookieName: "test",
119 },
120 scopes: [AuthFilesystemWriteScope],
121 permissions: [AuthFilesystemWriteScope],
122 }),
123 );
124 const openScratch = Effect.promise(() =>
125 commands.openScratch.run(registry, { environmentId: ENVIRONMENT_ID, input: {} }),
126 );
127 return {
128 registry,
129 storedProject,
130 openScratch,
131 commands,
132 metadataRequests,
133 readResponse,
134 file,
135 fileChanges,
136 };
137});
138
139describe("watched file reads", () => {
140 it.effect("refreshes a mounted file query when its host reports a disk change", () =>
141 Effect.gen(function* () {
142 const h = yield* makeHarness();
143 const watch = h.commands.watch({
144 environmentId: ENVIRONMENT_ID,
145 input: { cwd: "/repo", relativePath: "foo", directory: false },
146 });
147 const unwatch = h.registry.mount(watch);
148 yield* Effect.addFinalizer(() => Effect.sync(unwatch));
149 yield* AtomRegistry.getResult(h.registry, watch);
150 const query = h.commands.readFile({
151 environmentId: ENVIRONMENT_ID,
152 input: { cwd: "/repo", relativePath: "foo" },
153 });
154 const unmount = h.registry.mount(query);
155 yield* Effect.addFinalizer(() => Effect.sync(unmount));
156 expect(
157 (yield* AtomRegistry.getResult(h.registry, query, { suspendOnWaiting: true })).byteLength,
158 ).toBe(5);
159 const updated = yield* AtomRegistry.toStream(h.registry, query).pipe(
160 Stream.filter((result) => AsyncResult.isSuccess(result) && result.value.byteLength === 12),
161 Stream.runHead,
162 Effect.forkChild,
163 );
164 h.file.byteLength = 12;
165 yield* SubscriptionRef.set(h.fileChanges, 1);
166 const result = yield* Fiber.join(updated);
167 expect(Option.isSome(result)).toBe(true);
168 }).pipe(Effect.scoped),
169 );
170});
171
172describe("writeFile metadata", () => {
173 it.effect("reuses a current file read without another metadata request", () =>
174 Effect.gen(function* () {
175 const h = yield* makeHarness();
176 yield* AtomRegistry.getResult(
177 h.registry,
178 h.commands.readFile({
179 environmentId: ENVIRONMENT_ID,
180 input: { cwd: "/repo", relativePath: "foo" },
181 }),
182 { suspendOnWaiting: true },
183 );
184 const metadata = h.commands.fileMetadata({
185 environmentId: ENVIRONMENT_ID,
186 input: { path: "/repo/foo" },
187 });
188 expect(
189 yield* AtomRegistry.getResult(h.registry, metadata, { suspendOnWaiting: true }),
190 ).toEqual({ kind: "file", byteLength: 5 });
191 expect(h.metadataRequests).toHaveLength(0);
192 }).pipe(Effect.scoped),
193 );
194
195 it.effect("does not restore old metadata from a read completed after a write", () =>
196 Effect.gen(function* () {
197 const h = yield* makeHarness();
198 const threadId = ThreadId.make("stale-read-thread");
199 h.registry.mount(h.commands.fileMetadataThread({ environmentId: ENVIRONMENT_ID, threadId }));
200 const metadata = h.commands.fileMetadata({
201 environmentId: ENVIRONMENT_ID,
202 input: { path: "/repo/foo", threadId },
203 });
204 h.registry.mount(metadata);
205 const readMetadata = () =>
206 AtomRegistry.getResult(h.registry, metadata, { suspendOnWaiting: true });
207 yield* readMetadata();
208 const started = yield* Deferred.make<void>();
209 const release = yield* Deferred.make<void>();
210 h.readResponse.wait = Deferred.succeed(started, undefined).pipe(
211 Effect.andThen(Deferred.await(release)),
212 );
213 const fileRead = yield* AtomRegistry.getResult(
214 h.registry,
215 h.commands.readFile({
216 environmentId: ENVIRONMENT_ID,
217 input: { cwd: "/repo", relativePath: "foo" },
218 }),
219 { suspendOnWaiting: true },
220 ).pipe(Effect.forkChild);
221 yield* Deferred.await(started);
222 const write = yield* Effect.promise(() =>
223 h.commands.writeFile.run(h.registry, {
224 environmentId: ENVIRONMENT_ID,
225 input: { cwd: "/repo", relativePath: "foo", contents: "new" },
226 }),
227 );
228 expect(write._tag).toBe("Success");
229 expect(yield* readMetadata()).toEqual({ kind: "file", byteLength: 10 });
230 yield* Deferred.succeed(release, undefined);
231 expect((yield* Fiber.join(fileRead)).byteLength).toBe(5);
232 h.registry.refresh(metadata);
233 expect(yield* readMetadata()).toEqual({ kind: "file", byteLength: 10 });
234 expect(h.metadataRequests).toHaveLength(2);
235 }).pipe(Effect.scoped),
236 );
237
238 it.effect("refreshes retained canonical and original paths after an aliased write", () =>
239 Effect.gen(function* () {
240 const h = yield* makeHarness();
241 const threadId = ThreadId.make("write-metadata-thread");
242 h.registry.mount(h.commands.fileMetadataThread({ environmentId: ENVIRONMENT_ID, threadId }));
243 const atoms = ["/repo/foo", "/repo/./foo"].map((path) =>
244 h.commands.fileMetadata({ environmentId: ENVIRONMENT_ID, input: { path, threadId } }),
245 );
246 for (const atom of atoms) h.registry.mount(atom);
247 const read = () =>
248 Effect.forEach(
249 atoms,
250 (atom) => AtomRegistry.getResult(h.registry, atom, { suspendOnWaiting: true }),
251 { concurrency: "unbounded" },
252 );
253 expect(yield* read()).toEqual([
254 { kind: "file", byteLength: 5 },
255 { kind: "file", byteLength: 5 },
256 ]);
257 const result = yield* Effect.promise(() =>
258 h.commands.writeFile.run(h.registry, {
259 environmentId: ENVIRONMENT_ID,
260 input: { cwd: "/repo", relativePath: "./foo", contents: "new" },
261 }),
262 );
263 expect(result).toMatchObject({ _tag: "Success", value: { relativePath: "foo" } });
264 expect(yield* read()).toEqual([
265 { kind: "file", byteLength: 10 },
266 { kind: "file", byteLength: 10 },
267 ]);
268 expect(h.metadataRequests.flat()).toEqual([
269 "/repo/foo",
270 "/repo/./foo",
271 "/repo/foo",
272 "/repo/./foo",
273 ]);
274 }).pipe(Effect.scoped),
275 );
276});
277
278describe("openScratch", () => {
279 it.effect("resolves once the created project reaches the client store", () =>
280 Effect.gen(function* () {
281 const { registry, storedProject, openScratch } = yield* makeHarness();
282 const opening = yield* Effect.forkChild(openScratch);
283 yield* Effect.yieldNow;
284 registry.set(storedProject, PROJECT);
285 const result = yield* Fiber.join(opening);
286 expect(result).toMatchObject({ _tag: "Success", value: PROJECT });
287 }).pipe(Effect.scoped),
288 );
289});