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 { request, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
28import { followStreamInEnvironment } from "./runtime.ts";
29import { vcsCommandConcurrency, vcsCommandScheduler } from "./vcsCommandScheduler.ts";
30import {
31 invalidateCachedVcsRefs,
32 vcsRefsCacheStateAtom,
33 withVcsRefsPersistenceLock,
34} from "./vcsRefInvalidation.ts";
35
36const OFFLINE_BRANCH_LIST_LIMIT = 100;
37const VCS_REFS_IDLE_TTL_MS = 30_000;
38// Rows keep the last status they rendered, so the live stream only needs a
39// short grace period when virtualization or scrolling releases its consumer.
40const VCS_STATUS_IDLE_TTL_MS = 10_000;
41const VCS_REFS_RETRY_SCHEDULE = Schedule.exponential("1 second").pipe(
42 Schedule.modifyDelay(({ duration }) =>
43 Effect.succeed(Duration.min(duration, Duration.seconds(30))),
44 ),
45);
46
47function canUseVcsRefsCache(input: VcsListRefsInput): boolean {
48 return (
49 input.query === undefined &&
50 input.cursor === undefined &&
51 input.includeMatchingRemoteRefs === undefined &&
52 input.refKind === undefined &&
53 input.limit === OFFLINE_BRANCH_LIST_LIMIT
54 );
55}
56
57export const commitVcsRefsRefresh = Effect.fn("CachedVcsRefsState.commitRefresh")(function* (
58 registry: AtomRegistry.AtomRegistry,
59 cache: Persistence.EnvironmentCacheStore["Service"],
60 input: {
61 readonly environmentId: EnvironmentId;
62 readonly cwd: string;
63 readonly refs: VcsListRefsResult;
64 readonly expectedRevision: number;
65 readonly persist: boolean;
66 },
67) {
68 return yield* withVcsRefsPersistenceLock(
69 input.environmentId,
70 Effect.gen(function* () {
71 const stateAtom = vcsRefsCacheStateAtom({ environmentId: input.environmentId });
72 const state = registry.get(stateAtom);
73 if (state.revision !== input.expectedRevision) {
74 return false;
75 }
76 let persistedCacheReadable = state.persistedCacheReadable;
77 if (input.persist) {
78 if (!persistedCacheReadable) {
79 persistedCacheReadable = yield* cache.clearVcsRefs(input.environmentId).pipe(
80 Effect.as(true),
81 Effect.catch((error) =>
82 Effect.logWarning("Could not recover invalidated cached Git refs.").pipe(
83 Effect.annotateLogs({
84 environmentId: input.environmentId,
85 cwd: input.cwd,
86 ...safeErrorLogAttributes(error),
87 }),
88 Effect.as(false),
89 ),
90 ),
91 );
92 }
93 yield* cache.saveVcsRefs(input.environmentId, input.cwd, input.refs).pipe(
94 Effect.catch((error) =>
95 Effect.logWarning("Could not persist cached Git refs.").pipe(
96 Effect.annotateLogs({
97 environmentId: input.environmentId,
98 cwd: input.cwd,
99 ...safeErrorLogAttributes(error),
100 }),
101 ),
102 ),
103 );
104 if (persistedCacheReadable !== state.persistedCacheReadable) {
105 registry.update(stateAtom, (current) =>
106 current.revision === input.expectedRevision
107 ? { ...current, persistedCacheReadable }
108 : current,
109 );
110 }
111 }
112 return true;
113 }),
114 );
115});
116
117/**
118 * Retains the last unfiltered branch-list response for the new-task picker.
119 * Filtered or paginated lists intentionally stay live-only: treating a
120 * partial result as a complete offline list would make branch selection
121 * misleading.
122 */
123export const makeCachedVcsRefsChanges = Effect.fn("CachedVcsRefsState.makeChanges")(function* (
124 input: VcsListRefsInput,
125 expectedRevision?: number,
126 registry?: AtomRegistry.AtomRegistry,
127 persistedCacheReadable = true,
128) {
129 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
130 const cache = yield* Persistence.EnvironmentCacheStore;
131 const environmentId = supervisor.target.environmentId;
132 const useCache = canUseVcsRefsCache(input);
133 const cached =
134 useCache && persistedCacheReadable
135 ? yield* cache.loadVcsRefs(environmentId, input.cwd).pipe(
136 Effect.catch((error) =>
137 Effect.logWarning("Could not load cached Git refs.").pipe(
138 Effect.annotateLogs({
139 environmentId,
140 cwd: input.cwd,
141 ...safeErrorLogAttributes(error),
142 }),
143 Effect.as(Option.none<VcsListRefsResult>()),
144 ),
145 ),
146 )
147 : Option.none<VcsListRefsResult>();
148 const refresh = Effect.fn("CachedVcsRefsState.refresh")(function* () {
149 const refs = yield* request(WS_METHODS.vcsListRefs, input).pipe(
150 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
151 );
152 const persist = cache.saveVcsRefs(environmentId, input.cwd, refs).pipe(
153 Effect.catch((error) =>
154 Effect.logWarning("Could not persist cached Git refs.").pipe(
155 Effect.annotateLogs({
156 environmentId,
157 cwd: input.cwd,
158 ...safeErrorLogAttributes(error),
159 }),
160 ),
161 ),
162 );
163 if (expectedRevision === undefined || registry === undefined) {
164 if (useCache) yield* persist;
165 return Option.some(refs);
166 }
167 const committed = yield* commitVcsRefsRefresh(registry, cache, {
168 environmentId,
169 cwd: input.cwd,
170 refs,
171 expectedRevision,
172 persist: useCache,
173 });
174 return committed ? Option.some(refs) : Option.none<VcsListRefsResult>();
175 });
176
177 const cachedRefs = Stream.fromEffect(Effect.succeed(cached)).pipe(
178 Stream.filterMap((refs) =>
179 Option.match(refs, {
180 onNone: () => Result.failVoid,
181 onSome: Result.succeed,
182 }),
183 ),
184 );
185 const refreshedRefs = Stream.concat(
186 Stream.fromEffect(SubscriptionRef.get(supervisor.state)),
187 SubscriptionRef.changes(supervisor.state),
188 ).pipe(
189 Stream.map((connection) => (connection.phase === "connected" ? connection.generation : null)),
190 Stream.changes,
191 Stream.switchMap((generation) =>
192 generation === null
193 ? Stream.empty
194 : Stream.fromEffect(
195 refresh().pipe(
196 Effect.tapError((error) =>
197 Effect.logWarning("Could not refresh Git refs.").pipe(
198 Effect.annotateLogs({
199 environmentId,
200 cwd: input.cwd,
201 ...safeErrorLogAttributes(error),
202 }),
203 ),
204 ),
205 ),
206 ).pipe(
207 Stream.retry(VCS_REFS_RETRY_SCHEDULE),
208 Stream.filterMap((refs) =>
209 Option.match(refs, {
210 onNone: () => Result.failVoid,
211 onSome: Result.succeed,
212 }),
213 ),
214 ),
215 ),
216 );
217
218 return Stream.concat(cachedRefs, refreshedRefs);
219});
220
221function cachedVcsRefsChanges(
222 environmentId: EnvironmentId,
223 input: VcsListRefsInput,
224 expectedRevision: number,
225 persistedCacheReadable: boolean,
226) {
227 return followStreamInEnvironment(
228 environmentId,
229 Stream.unwrap(
230 Effect.gen(function* () {
231 const registry = yield* AtomRegistry.AtomRegistry;
232 return yield* makeCachedVcsRefsChanges(
233 input,
234 expectedRevision,
235 registry,
236 persistedCacheReadable,
237 );
238 }),
239 ),
240 );
241}
242
243export function createVcsEnvironmentAtoms<R, E>(
244 runtime: Atom.AtomRuntime<EnvironmentRegistry | Persistence.EnvironmentCacheStore | R, E>,
245) {
246 /**
247 * One flat family on purpose: families hold entries via WeakRef, so a nested
248 * per-environment family can be collected between lookups, dropping every
249 * cached page atom and collapsing paginated ref lists mid-scroll.
250 */
251 const listRefsFamily = Atom.family((key: string) => {
252 const [environmentId, input] = JSON.parse(key) as [EnvironmentId, VcsListRefsInput];
253 return runtime
254 .atom((get) => {
255 const state = get(vcsRefsCacheStateAtom({ environmentId }));
256 return cachedVcsRefsChanges(
257 environmentId,
258 input,
259 state.revision,
260 state.persistedCacheReadable,
261 );
262 })
263 .pipe(
264 Atom.setIdleTTL(VCS_REFS_IDLE_TTL_MS),
265 Atom.withLabel(`environment-data:vcs:list-refs:${key}`),
266 );
267 });
268 const listRefs = (target: {
269 readonly environmentId: EnvironmentId;
270 readonly input: VcsListRefsInput;
271 }) => listRefsFamily(JSON.stringify([target.environmentId, target.input]));
272 const invalidateRefs = (
273 target: { readonly environmentId: EnvironmentId; readonly input: { readonly cwd: string } },
274 registry: AtomRegistry.AtomRegistry,
275 ) =>
276 invalidateCachedVcsRefs(registry, {
277 environmentId: target.environmentId,
278 cwd: target.input.cwd,
279 });
280
281 return {
282 listRefs,
283 status: createEnvironmentSubscriptionAtomFamily(runtime, {
284 label: "environment-data:vcs:status",
285 idleTtlMs: VCS_STATUS_IDLE_TTL_MS,
286 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.subscribeVcsStatus>) =>
287 subscribe(WS_METHODS.subscribeVcsStatus, input).pipe(
288 Stream.mapAccum(
289 () => null as VcsStatusResult | null,
290 (current, event) => {
291 const next = applyGitStatusStreamEvent(current, event);
292 return [next, [next]] as const;
293 },
294 ),
295 ),
296 }),
297 pull: createEnvironmentRpcCommand(runtime, {
298 label: "environment-data:vcs:pull",
299 tag: WS_METHODS.vcsPull,
300 scheduler: vcsCommandScheduler,
301 concurrency: vcsCommandConcurrency,
302 onSettled: invalidateRefs,
303 }),
304 refreshStatus: createEnvironmentRpcCommand(runtime, {
305 label: "environment-data:vcs:refresh-status",
306 tag: WS_METHODS.vcsRefreshStatus,
307 scheduler: vcsCommandScheduler,
308 concurrency: vcsCommandConcurrency,
309 onSettled: invalidateRefs,
310 }),
311 createWorktree: createEnvironmentRpcCommand(runtime, {
312 label: "environment-data:vcs:create-worktree",
313 tag: WS_METHODS.vcsCreateWorktree,
314 scheduler: vcsCommandScheduler,
315 concurrency: vcsCommandConcurrency,
316 onSettled: invalidateRefs,
317 }),
318 // Live stages of a bootstrap worktree setup. Null until the server begins
319 // tracking, then a snapshot per change, then null again after the setup
320 // is dropped. Short TTL so a closed thread releases its subscription.
321 worktreeSetup: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
322 label: "environment-data:vcs:worktree-setup",
323 tag: WS_METHODS.subscribeWorktreeSetup,
324 idleTtlMs: VCS_STATUS_IDLE_TTL_MS,
325 }),
326 cancelWorktreeSetup: createEnvironmentRpcCommand(runtime, {
327 label: "environment-data:vcs:cancel-worktree-setup",
328 tag: WS_METHODS.worktreeSetupCancel,
329 }),
330 removeWorktree: createEnvironmentRpcCommand(runtime, {
331 label: "environment-data:vcs:remove-worktree",
332 tag: WS_METHODS.vcsRemoveWorktree,
333 scheduler: vcsCommandScheduler,
334 concurrency: vcsCommandConcurrency,
335 onSettled: invalidateRefs,
336 }),
337 createRef: createEnvironmentRpcCommand(runtime, {
338 label: "environment-data:vcs:create-ref",
339 tag: WS_METHODS.vcsCreateRef,
340 scheduler: vcsCommandScheduler,
341 concurrency: vcsCommandConcurrency,
342 onSettled: invalidateRefs,
343 }),
344 switchRef: createEnvironmentRpcCommand(runtime, {
345 label: "environment-data:vcs:switch-ref",
346 tag: WS_METHODS.vcsSwitchRef,
347 scheduler: vcsCommandScheduler,
348 concurrency: vcsCommandConcurrency,
349 onSettled: invalidateRefs,
350 }),
351 init: createEnvironmentRpcCommand(runtime, {
352 label: "environment-data:vcs:init",
353 tag: WS_METHODS.vcsInit,
354 scheduler: vcsCommandScheduler,
355 concurrency: vcsCommandConcurrency,
356 onSettled: invalidateRefs,
357 }),
358 };
359}
360
361export * from "./gitActions.ts";
362export * from "./vcsAction.ts";
363export * from "./vcsRef.ts";