1import {
2 FILESYSTEM_METADATA_BATCH_LIMIT,
3 type EnvironmentId,
4 type FilesystemEntryMetadata,
5 type ProjectEntry,
6 type ScopedThreadRef,
7 type ThreadId,
8 WS_METHODS,
9} from "@t3tools/contracts";
10import { normalizeProjectPathForComparison, resolveWorkspaceFilePath } from "@t3tools/shared/path";
11import * as Clock from "effect/Clock";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import * as Request from "effect/Request";
15import * as RequestResolver from "effect/RequestResolver";
16import * as SubscriptionRef from "effect/SubscriptionRef";
17import { Atom, AtomRegistry } from "effect/reactivity";
18
19import * as EnvironmentSupervisor from "../connection/supervisor.ts";
20import type { EnvironmentRegistry } from "../connection/registry.ts";
21import {
22 request,
23 type EnvironmentRpcFailure,
24 EnvironmentRpcUnavailableError,
25} from "../rpc/client.ts";
26import type { RpcSession } from "../rpc/session.ts";
27import { createEnvironmentQueryAtomFamily } from "./runtime.ts";
28
29const METADATA_STALE_TIME_MS = 60_000;
30const METADATA_CACHE_LIMIT = 1024;
31
32interface CachedMetadata {
33 value: FilesystemEntryMetadata | null;
34 expires: number;
35}
36
37type RetainedMetadata = Map<string, CachedMetadata>;
38
39interface FileMetadataInput {
40 path: string;
41 threadId?: ThreadId;
42}
43
44class FileMetadataRequest extends Request.Class<
45 {
46 path: string;
47 session: RpcSession;
48 supervisor: EnvironmentSupervisor.EnvironmentSupervisor["Service"];
49 retained?: RetainedMetadata;
50 },
51 FilesystemEntryMetadata | null,
52 EnvironmentRpcFailure<typeof WS_METHODS.filesystemGetMetadata> | EnvironmentRpcUnavailableError
53> {}
54
55/** Shared by file chips, search results, and the file tree in one client. Cache
56 * entries belong to an RPC session so reconnects and grant changes recheck them. */
57export function createFileMetadataAtoms<R, E>(
58 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
59) {
60 const cache = new WeakMap<RpcSession, Map<string, CachedMetadata>>();
61 const invalidationVersions = new WeakMap<RpcSession, number>();
62 const inFlight = new WeakMap<RpcSession, Map<string, Set<{ valid: boolean }>>>();
63 const retainedThreads = new Map<
64 string,
65 { readers: number; sessions: WeakMap<RpcSession, RetainedMetadata> }
66 >();
67 const threadKey = (ref: ScopedThreadRef) => JSON.stringify([ref.environmentId, ref.threadId]);
68 const retainThreadFamily = Atom.family((key: string) =>
69 Atom.make((get) => {
70 let retained = retainedThreads.get(key);
71 if (!retained) {
72 retained = { readers: 0, sessions: new WeakMap() };
73 retainedThreads.set(key, retained);
74 }
75 retained.readers++;
76 get.addFinalizer(() => {
77 if (--retained.readers === 0) retainedThreads.delete(key);
78 });
79 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`file-metadata-thread:${key}`)),
80 );
81 const revisions = Atom.family((key: string) =>
82 Atom.make(0).pipe(Atom.withLabel(`file-metadata-revision:${key}`)),
83 );
84 const revision = (environmentId: EnvironmentId, path: string) =>
85 revisions(JSON.stringify([environmentId, normalizeProjectPathForComparison(path)]));
86 const knownEntry = (session: RpcSession, path: string, now: number) => {
87 const key = normalizeProjectPathForComparison(path);
88 const cached = cache.get(session)?.get(key);
89 return cached && cached.expires > now ? cached : undefined;
90 };
91 const remember = (
92 session: RpcSession,
93 path: string,
94 value: FilesystemEntryMetadata | null,
95 now: number,
96 partial = false,
97 ) => {
98 let entries = cache.get(session);
99 if (!entries) {
100 entries = new Map();
101 cache.set(session, entries);
102 }
103 const key = normalizeProjectPathForComparison(path);
104 entries.delete(key);
105 const entry = { value, expires: now + METADATA_STALE_TIME_MS };
106 entries.set(key, entry);
107 for (const thread of retainedThreads.values()) {
108 const retained = thread.sessions.get(session);
109 const previous = retained?.get(key);
110 if (retained && previous)
111 retained.set(key, {
112 ...entry,
113 value:
114 partial && value && previous.value?.kind === value.kind
115 ? { ...previous.value, ...value }
116 : value,
117 });
118 }
119 if (entries.size > METADATA_CACHE_LIMIT) {
120 const oldest = entries.keys().next().value;
121 if (oldest !== undefined) entries.delete(oldest);
122 }
123 };
124 const resolver = RequestResolver.makeWith<FileMetadataRequest>({
125 batchKey: (entry) => entry.request.session,
126 delay: Effect.yieldNow,
127 collectWhile: (entries) => entries.size < FILESYSTEM_METADATA_BATCH_LIMIT,
128 runAll: Effect.fnUntraced(function* (entries) {
129 const session = entries[0].request.session;
130 let pending = [...entries];
131 while (pending.length > 0) {
132 const paths = [...new Set(pending.map((entry) => entry.request.path))];
133 let requests = inFlight.get(session);
134 if (!requests) {
135 requests = new Map();
136 inFlight.set(session, requests);
137 }
138 const tokens = new Map(paths.map((path) => [path, { valid: true }]));
139 for (const [path, token] of tokens) {
140 let active = requests.get(path);
141 if (!active) requests.set(path, (active = new Set()));
142 active.add(token);
143 }
144 const activeRequests = requests;
145 pending = yield* Effect.gen(function* () {
146 const result = yield* request(WS_METHODS.filesystemGetMetadata, { paths }).pipe(
147 Effect.provideService(
148 EnvironmentSupervisor.EnvironmentSupervisor,
149 entries[0].request.supervisor,
150 ),
151 );
152 const now = yield* Clock.currentTimeMillis;
153 const byPath = new Map(paths.map((path, index) => [path, result.entries[index] ?? null]));
154 const retry: typeof pending = [];
155 for (const entry of pending) {
156 // Retry only paths invalidated during this request. Never retain the
157 // old response or complete a waiting chip with pre-write metadata.
158 if (!tokens.get(entry.request.path)!.valid) {
159 retry.push(entry);
160 continue;
161 }
162 const value = byPath.get(entry.request.path) ?? null;
163 remember(session, entry.request.path, value, now);
164 entry.request.retained?.set(entry.request.path, {
165 value,
166 expires: now + METADATA_STALE_TIME_MS,
167 });
168 entry.completeUnsafe(Exit.succeed(value));
169 }
170 return retry;
171 }).pipe(
172 Effect.ensuring(
173 Effect.sync(() => {
174 for (const [path, token] of tokens) {
175 const active = activeRequests.get(path);
176 active?.delete(token);
177 if (active?.size === 0) activeRequests.delete(path);
178 }
179 }),
180 ),
181 );
182 }
183 }),
184 });
185
186 const metadataFamily = createEnvironmentQueryAtomFamily(runtime, {
187 label: "environment-data:filesystem:metadata",
188 staleTimeMs: METADATA_STALE_TIME_MS,
189 idleTtlMs: METADATA_STALE_TIME_MS,
190 execute: Effect.fnUntraced(function* (input: FileMetadataInput) {
191 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
192 const session = yield* SubscriptionRef.get(supervisor.session);
193 if (session._tag === "None") return null;
194 const thread = input.threadId
195 ? retainedThreads.get(
196 threadKey({ environmentId: supervisor.target.environmentId, threadId: input.threadId }),
197 )
198 : undefined;
199 let retained = thread?.sessions.get(session.value);
200 if (thread && !retained) {
201 retained = new Map();
202 thread.sessions.set(session.value, retained);
203 }
204 const retainedEntry = retained?.get(input.path);
205 if (retainedEntry) return retainedEntry.value;
206 const now = yield* Clock.currentTimeMillis;
207 const cached = knownEntry(session.value, input.path, now);
208 if (cached) {
209 retained?.set(input.path, cached);
210 return cached.value;
211 }
212 return yield* Effect.request(
213 new FileMetadataRequest({
214 path: input.path,
215 session: session.value,
216 supervisor,
217 ...(retained ? { retained } : {}),
218 }),
219 resolver,
220 );
221 }),
222 refreshTrigger: ({
223 environmentId,
224 input,
225 }: {
226 environmentId: EnvironmentId;
227 input: FileMetadataInput;
228 }) => revision(environmentId, input.path),
229 });
230 const metadata = (target: {
231 environmentId: EnvironmentId;
232 input: { path: string; threadId?: ThreadId };
233 }) =>
234 metadataFamily({
235 ...target,
236 input: { ...target.input, path: normalizeProjectPathForComparison(target.input.path) },
237 });
238
239 const rememberEntries = Effect.fnUntraced(function* (
240 cwd: string,
241 entries: ReadonlyArray<ProjectEntry>,
242 isCurrent?: (session: RpcSession) => boolean,
243 ) {
244 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
245 const session = yield* SubscriptionRef.get(supervisor.session);
246 if (session._tag === "None") return;
247 const now = yield* Clock.currentTimeMillis;
248 if (isCurrent && !isCurrent(session.value)) return;
249 for (const entry of entries) {
250 const path = resolveWorkspaceFilePath(entry.path, cwd);
251 const previous = knownEntry(session.value, path, now);
252 if (previous?.value?.kind === entry.kind) continue;
253 // Listing kinds avoid another request; size and MIME hints are optional.
254 remember(session.value, path, { kind: entry.kind }, now, true);
255 }
256 });
257
258 const invalidate = Effect.fnUntraced(function* (path: string) {
259 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
260 const session = yield* SubscriptionRef.get(supervisor.session);
261 if (session._tag === "Some") {
262 invalidationVersions.set(session.value, (invalidationVersions.get(session.value) ?? 0) + 1);
263 const key = normalizeProjectPathForComparison(path);
264 for (const token of inFlight.get(session.value)?.get(key) ?? []) token.valid = false;
265 cache.get(session.value)?.delete(key);
266 for (const thread of retainedThreads.values())
267 thread.sessions.get(session.value)?.delete(normalizeProjectPathForComparison(path));
268 }
269 });
270
271 const rememberFile = Effect.fnUntraced(function* (
272 cwd: string,
273 file: { relativePath: string; byteLength: number },
274 isCurrent?: (session: RpcSession) => boolean,
275 ) {
276 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
277 const session = yield* SubscriptionRef.get(supervisor.session);
278 if (session._tag === "None") return;
279 const path = resolveWorkspaceFilePath(file.relativePath, cwd);
280 const now = yield* Clock.currentTimeMillis;
281 if (isCurrent && !isCurrent(session.value)) return;
282 const previous = knownEntry(session.value, path, now)?.value;
283 remember(
284 session.value,
285 path,
286 {
287 ...previous,
288 kind: "file",
289 byteLength: file.byteLength,
290 },
291 now,
292 true,
293 );
294 });
295
296 const seedAfterRead = <A, ReadError, ReadServices>(
297 read: Effect.Effect<A, ReadError, ReadServices>,
298 seed: (
299 result: A,
300 isCurrent: (session: RpcSession) => boolean,
301 ) => Effect.Effect<void, never, EnvironmentSupervisor.EnvironmentSupervisor>,
302 ) =>
303 Effect.gen(function* () {
304 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
305 const session = yield* SubscriptionRef.get(supervisor.session);
306 const version = session._tag === "Some" ? (invalidationVersions.get(session.value) ?? 0) : 0;
307 const result = yield* read;
308 // Seeding is optional. Any intervening invalidation or reconnect makes
309 // the old response unsuitable for the current metadata cache.
310 yield* seed(
311 result,
312 (current) =>
313 session._tag === "Some" &&
314 current === session.value &&
315 (invalidationVersions.get(current) ?? 0) === version,
316 );
317 return result;
318 });
319
320 const refreshPath = (
321 environmentId: EnvironmentId,
322 path: string,
323 registry: AtomRegistry.AtomRegistry,
324 ) => {
325 const signal = revision(environmentId, path);
326 registry.set(signal, registry.get(signal) + 1);
327 };
328
329 return {
330 metadata,
331 retainThread: (ref: ScopedThreadRef) => retainThreadFamily(threadKey(ref)),
332 rememberEntries,
333 rememberFile,
334 seedAfterRead,
335 invalidate,
336 refreshPath,
337 };
338}