1import {
2 type EnvironmentId,
3 type VcsListRefsInput,
4 type VcsListRefsResult,
5 type VcsStatusResult,
6 WS_METHODS,
7} from "@t3tools/contracts";
8import { applyGitStatusStreamEvent } from "@t3tools/shared/git";
9import * as Duration from "effect/Duration";
10import * as Effect from "effect/Effect";
11import * as Option from "effect/Option";
12import * as Result from "effect/Result";
13import * as Schedule from "effect/Schedule";
14import * as Stream from "effect/Stream";
15import * as SubscriptionRef from "effect/SubscriptionRef";
16import { Atom, AtomRegistry } from "effect/reactivity";
17
18import {
19 createEnvironmentRpcCommand,
20 createEnvironmentRpcSubscriptionAtomFamily,
21 createEnvironmentSubscriptionAtomFamily,
22} from "./runtime.ts";
23import type { EnvironmentRegistry } from "../connection/registry.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import { safeErrorLogAttributes } from "../errors/safeLog.ts";
26import * as Persistence from "../platform/persistence.ts";
27import {
28 request,
29 subscribe,
30 subscribeVcsBranch,
31 refreshVcsBranch,
32 type EnvironmentRpcInput,
33} from "../rpc/client.ts";
34import { followStreamInEnvironment } from "./runtime.ts";
35import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
36import {
37 invalidateCachedVcsRefs,
38 vcsRefsCacheStateAtom,
39 withVcsRefsPersistenceLock,
40} from "./vcsRefInvalidation.ts";
41
42// Share freshness across search and pagination keys, but keep it local to this registry.
43const serverRefsRevisionAtom = Atom.family((_key: string) => Atom.make(0).pipe(Atom.keepAlive));
44
45const OFFLINE_BRANCH_LIST_LIMIT = 100;
46const VCS_REFS_IDLE_TTL_MS = 30_000;
47const VCS_STATUS_IDLE_TTL_MS = 10_000;
48const VCS_REFS_RETRY_SCHEDULE = Schedule.exponential("1 second").pipe(
49 Schedule.modifyDelay(({ duration }) =>
50 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
51 ),
52);
53
54function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
55 return (
56 input.query === undefined &&
57 input.cursor === undefined &&
58 input.includeMatchingRemoteRefs === undefined &&
59 input.refKind === undefined &&
60 input.limit === OFFLINE_BRANCH_LIST_LIMIT
61 );
62}
63
64export const commitVcsRefsRefresh = Effect.fn("CachedVcsRefsState.commitRefresh")(function* (
65 registry: AtomRegistry.AtomRegistry,
66 cache: Persistence.EnvironmentCacheStore["Service"],
67 input: {
68 readonly environmentId: EnvironmentId;
69 readonly cwd: string;
70 readonly refs: VcsListRefsResult;
71 readonly expectedRevision: number;
72 readonly persist: boolean;
73 },
74) {
75 return yield* withVcsRefsPersistenceLock(
76 input.environmentId,
77 Effect.gen(function* () {
78 const stateAtom = vcsRefsCacheStateAtom({ environmentId: input.environmentId });
79 const state = registry.get(stateAtom);
80 if (state.revision !== input.expectedRevision) {
81 return false;
82 }
83 let persistedCacheReadable = state.persistedCacheReadable;
84 if (input.persist) {
85 if (!persistedCacheReadable) {
86 persistedCacheReadable = yield* cache.clearVcsRefs(input.environmentId).pipe(
87 Effect.as(true),
88 Effect.catch((error) =>
89 Effect.logWarning("Could not recover invalidated cached Git refs.").pipe(
90 Effect.annotateLogs({
91 environmentId: input.environmentId,
92 cwd: input.cwd,
93 ...safeErrorLogAttributes(error),
94 }),
95 Effect.as(false),
96 ),
97 ),
98 );
99 }
100 yield* cache.saveVcsRefs(input.environmentId, input.cwd, input.refs).pipe(
101 Effect.catch((error) =>
102 Effect.logWarning("Could not persist cached Git refs.").pipe(
103 Effect.annotateLogs({
104 environmentId: input.environmentId,
105 cwd: input.cwd,
106 ...safeErrorLogAttributes(error),
107 }),
108 ),
109 ),
110 );
111 if (persistedCacheReadable !== state.persistedCacheReadable) {
112 registry.update(stateAtom, (current) =>
113 current.revision === input.expectedRevision
114 ? { ...current, persistedCacheReadable }
115 : current,
116 );
117 }
118 }
119 return true;
120 }),
121 );
122});
123
124/**
125 * Filtered or paginated lists intentionally stay live-only: treating a
126 * partial result as a complete offline list would make branch selection
127 * misleading.
128 */
129export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChanges")(function* (
130 input: VcsListRefsInput,
131 expectedRevision?: number,
132 registry?: AtomRegistry.AtomRegistry,
133 persistedCacheReadable = true,
134) {
135 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
136 const cache = yield* Persistence.EnvironmentCacheStore;
137 const environmentId = supervisor.target.environmentId;
138 const useCache = canUseVcsRefsCache(input);
139 const cached =
140 useCache && persistedCacheReadable
141 ? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe(
142 Effect.catch((error) =>
143 Effect.logWarning("Could not load cached Git refs.").pipe(
144 Effect.annotateLogs({
145 environmentId,
146 cwd: input.cwd,
147 ...safeErrorLogAttributes(error),
148 }),
149 Effect.as(Option.none<VcsListRefsResult>()),
150 ),
151 ),
152 )
153 : Option.none<VcsListRefsResult>();
154 const refresh = Effect.fn("CachedVcsRefsState.refresh")(function* () {
155 const revisionAtom = serverRefsRevisionAtom(
156 `${environmentId.length}:${environmentId}${input.cwd}`,
157 );
158 const bypassServerSnapshot =
159 expectedRevision !== undefined &&
160 registry !== undefined &&
161 expectedRevision > registry.get(revisionAtom);
162 const refs = yield* request(
163 WS_METHODS.vcsListRefs,
164 bypassServerSnapshot ? { ...input, refresh: true } : input,
165 ).pipe(Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor));
166 const persist = cache.saveVcsRefs(environmentId, input.cwd, refs).pipe(
167 Effect.catch((error) =>
168 Effect.logWarning("Could not persist cached Git refs.").pipe(
169 Effect.annotateLogs({
170 environmentId,
171 cwd: input.cwd,
172 ...safeErrorLogAttributes(error),
173 }),
174 ),
175 ),
176 );
177 if (expectedRevision === undefined || registry === undefined) {
178 if (useCache) yield* persist;
179 return Option.some(refs);
180 }
181 const committed = yield* commitVcsRefsRefresh(registry, cache, {
182 environmentId,
183 cwd: input.cwd,
184 refs,
185 expectedRevision,
186 persist: useCache,
187 });
188 if (committed && bypassServerSnapshot) registry.set(revisionAtom, expectedRevision);
189 return committed ? Option.some(refs) : Option.none<VcsListRefsResult>();
190 });
191
192 const cachedRefs = Stream.fromEffect(Effect.succeed(cached)).pipe(
193 Stream.filterMap((refs) =>
194 Option.match(refs, {
195 onNone: () => Result.failVoid,
196 onSome: Result.succeed,
197 }),
198 ),
199 );
200 const refreshedRefs = SubscriptionRef.changes(supervisor.session).pipe(
201 Stream.map(Option.getOrNull),
202 Stream.changes,
203 Stream.switchMap((session) =>
204 session === null
205 ? Stream.empty
206 : Stream.fromEffect(
207 refresh().pipe(
208 // Session removal can precede switchMap cancelling this refresh.
209 Effect.catchTags({ EnvironmentRpcUnavailableError: () => Effect.never }),
210 Effect.tapError((error) =>
211 Effect.logWarning("Could not refresh Git refs.").pipe(
212 Effect.annotateLogs({
213 environmentId,
214 cwd: input.cwd,
215 ...safeErrorLogAttributes(error),
216 }),
217 ),
218 ),
219 ),
220 ).pipe(
221 Stream.retry(VCS_REFS_RETRY_SCHEDULE),
222 Stream.filterMap((refs) =>
223 Option.match(refs, {
224 onNone: () => Result.failVoid,
225 onSome: Result.succeed,
226 }),
227 ),
228 ),
229 ),
230 );
231
232 return Stream.concat(cachedRefs, refreshedRefs);
233});
234
235function cachedVcsRefsChanges(
236 environmentId: EnvironmentId,
237 input: VcsListRefsInput,
238 expectedRevision: number,
239 persistedCacheReadable: boolean,
240) {
241 return followStreamInEnvironment(
242 environmentId,
243 Stream.unwrap(
244 Effect.gen(function* () {
245 const registry = yield* AtomRegistry.AtomRegistry;
246 return yield* makeCachedVcsRefsChanges(
247 input,
248 expectedRevision,
249 registry,
250 persistedCacheReadable,
251 );
252 }),
253 ),
254 );
255}
256
257export function createVcsEnvironmentAtoms<R, E>(
258 runtime: Atom.AtomRuntime<EnvironmentRegistry | Persistence.EnvironmentCacheStore | R, E>,
259) {
260 /**
261 * One flat family on purpose: families hold entries via WeakRef, so a nested
262 * per-environment family can be collected between lookups, dropping every
263 * cached page atom and collapsing paginated ref lists mid-scroll.
264 */
265 const listRefsFamily = Atom.family((key: string) => {
266 const [environmentId, input] = JSON.parse(key) as [EnvironmentId, VcsListRefsInput];
267 return runtime
268 .atom((get) => {
269 const state = get(vcsRefsCacheStateAtom({ environmentId }));
270 return cachedVcsRefsChanges(
271 environmentId,
272 input,
273 state.revision,
274 state.persistedCacheReadable,
275 );
276 })
277 .pipe(
278 Atom.setIdleTTL(VCS_REFS_IDLE_TTL_MS),
279 Atom.withLabel(`environment-data:vcs:list-refs:${key}`),
280 );
281 });
282 const listRefs = (target: {
283 readonly environmentId: EnvironmentId;
284 readonly input: VcsListRefsInput;
285 }) => listRefsFamily(JSON.stringify([target.environmentId, target.input]));
286 const invalidateRefs = (
287 target: { readonly environmentId: EnvironmentId; readonly input: { readonly cwd: string } },
288 registry: AtomRegistry.AtomRegistry,
289 ) =>
290 invalidateCachedVcsRefs(registry, {
291 environmentId: target.environmentId,
292 cwd: target.input.cwd,
293 });
294
295 return {
296 listRefs,
297 // An uncached read for decisions that must match the repository now, not
298 // the persisted list a picker renders from.
299 readRefs: createEnvironmentRpcCommand(runtime, {
300 label: "environment-data:vcs:read-refs",
301 tag: WS_METHODS.vcsListRefs,
302 }),
303 branch: createEnvironmentSubscriptionAtomFamily(runtime, {
304 label: "environment-data:vcs:branch",
305 idleTtlMs: VCS_STATUS_IDLE_TTL_MS,
306 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsBranch>) =>
307 subscribeVcsBranch(input),
308 }),
309 refreshBranch: createEnvironmentRpcCommand(runtime, {
310 label: "environment-data:vcs:refresh-branch",
311 tag: WS_METHODS.vcsRefreshBranch,
312 execute: refreshVcsBranch,
313 scheduler: vcsCommandScheduler,
314 concurrency: vcsCommandConcurrency,
315 }),
316 status: createEnvironmentSubscriptionAtomFamily(runtime, {
317 label: "environment-data:vcs:status",
318 idleTtlMs: VCS_STATUS_IDLE_TTL_MS,
319 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>) =>
320 subscribe(WS_METHODS.subscribeVcsStatus, input).pipe(
321 Stream.mapAccum(
322 () => null as VcsStatusResult | null,
323 (current, event) => {
324 const next = applyGitStatusStreamEvent(current, event);
325 return [next, [next]] as const;
326 },
327 ),
328 ),
329 }),
330 pull: createEnvironmentRpcCommand(runtime, {
331 label: "environment-data:vcs:pull",
332 tag: WS_METHODS.vcsPull,
333 scheduler: vcsCommandScheduler,
334 concurrency: vcsCommandConcurrency,
335 onSettled: invalidateRefs,
336 }),
337 refreshStatus: createEnvironmentRpcCommand(runtime, {
338 label: "environment-data:vcs:refresh-status",
339 tag: WS_METHODS.vcsRefreshStatus,
340 scheduler: vcsCommandScheduler,
341 concurrency: vcsCommandConcurrency,
342 onSettled: invalidateRefs,
343 }),
344 // Outside the serial queue: it only reads status, and a checkout picked right after opening
345 // the picker must not wait behind that read.
346 refreshLocalStatusForPicker: createEnvironmentRpcCommand(runtime, {
347 label: "environment-data:vcs:refresh-local-status-for-picker",
348 tag: WS_METHODS.vcsRefreshStatus,
349 onSettled: invalidateRefs,
350 }),
351 createWorktree: createEnvironmentRpcCommand(runtime, {
352 label: "environment-data:vcs:create-worktree",
353 tag: WS_METHODS.vcsCreateWorktree,
354 scheduler: vcsCommandScheduler,
355 concurrency: vcsCommandConcurrency,
356 onSettled: invalidateRefs,
357 }),
358 worktreeSetup: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
359 label: "environment-data:vcs:worktree-setup",
360 tag: WS_METHODS.subscribeWorktreeSetup,
361 idleTtlMs: VCS_STATUS_IDLE_TTL_MS,
362 }),
363 cancelWorktreeSetup: createEnvironmentRpcCommand(runtime, {
364 label: "environment-data:vcs:cancel-worktree-setup",
365 tag: WS_METHODS.worktreeSetupCancel,
366 }),
367 removeWorktree: createEnvironmentRpcCommand(runtime, {
368 label: "environment-data:vcs:remove-worktree",
369 tag: WS_METHODS.vcsRemoveWorktree,
370 scheduler: vcsCommandScheduler,
371 concurrency: vcsCommandConcurrency,
372 onSettled: invalidateRefs,
373 }),
374 createRef: createEnvironmentRpcCommand(runtime, {
375 label: "environment-data:vcs:create-ref",
376 tag: WS_METHODS.vcsCreateRef,
377 scheduler: vcsCommandScheduler,
378 concurrency: vcsCommandConcurrency,
379 onSettled: invalidateRefs,
380 }),
381 switchRef: createEnvironmentRpcCommand(runtime, {
382 label: "environment-data:vcs:switch-ref",
383 tag: WS_METHODS.vcsSwitchRef,
384 scheduler: vcsCommandScheduler,
385 concurrency: vcsCommandConcurrency,
386 onSettled: invalidateRefs,
387 }),
388 init: createEnvironmentRpcCommand(runtime, {
389 label: "environment-data:vcs:init",
390 tag: WS_METHODS.vcsInit,
391 scheduler: vcsCommandScheduler,
392 concurrency: vcsCommandConcurrency,
393 onSettled: invalidateRefs,
394 }),
395 };
396}
397
398export * from "./gitActions.ts";
399export * from "./vcsAction.ts";
400export * from "./vcsRef.ts";
401export * from "./vcsStatusRevalidation.ts";