1import {
2 EnvironmentId,
3 type EnvironmentId as EnvironmentIdType,
4 GitActionProgressPhase,
5 type GitActionProgressEvent,
6 type GitRunStackedActionInput,
7 type GitRunStackedActionResult,
8 GitStackedAction,
9 type ThreadId,
10 type ProjectId,
11 WS_METHODS,
12} from "@t3tools/contracts";
13import * as Cause from "effect/Cause";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Schema from "effect/Schema";
17import * as Stream from "effect/Stream";
18import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity";
19
20import type { EnvironmentRegistry } from "../connection/registry.ts";
21import * as Persistence from "../platform/persistence.ts";
22import { runStreamGuarded, RpcPermissionGuard } from "../rpc/client.ts";
23import {
24 createRuntimeCommand,
25 runStreamInEnvironment,
26 type AtomCommand,
27 type AtomCommandResult,
28} from "./runtime.ts";
29import { createCommandPermissions } from "./commandPermissions.ts";
30import { vcsCommandScheduler } from "./vcsCommandScheduler.ts";
31import { invalidateCachedVcsRefs } from "./vcsRefInvalidation.ts";
32
33export const VcsActionOperation = Schema.Literals([
34 "refresh_status",
35 "run_change_request",
36 "pull",
37 "switch_ref",
38 "create_ref",
39 "create_worktree",
40 "init",
41 "publish_repository",
42 "prepare_pull_request_thread",
43]);
44export type VcsActionOperation = typeof VcsActionOperation.Type;
45
46export interface VcsActionState {
47 readonly isRunning: boolean;
48 readonly operation: VcsActionOperation | null;
49 readonly actionId: string | null;
50 readonly action: GitStackedAction | null;
51 readonly currentLabel: string | null;
52 readonly currentPhaseLabel: string | null;
53 readonly hookName: string | null;
54 readonly lastOutputLine: string | null;
55 readonly phaseStartedAtMs: number | null;
56 readonly hookStartedAtMs: number | null;
57 readonly error: string | null;
58}
59
60export interface VcsActionTarget {
61 readonly environmentId: EnvironmentIdType | null;
62 readonly cwd: string | null;
63}
64
65export interface ResolvedVcsActionTarget {
66 readonly environmentId: EnvironmentIdType;
67 readonly cwd: string;
68}
69
70export interface BeginVcsActionInput {
71 readonly operation: VcsActionOperation;
72 readonly label: string;
73 readonly actionId?: string;
74}
75
76export interface RunVcsStackedActionInput {
77 readonly actionId: string;
78 readonly action: GitStackedAction;
79 readonly commitMessage?: string;
80 readonly featureBranch?: boolean;
81 readonly filePaths?: ReadonlyArray<string>;
82 readonly threadId?: ThreadId;
83 readonly projectId?: ProjectId;
84 readonly onProgress?: (event: GitActionProgressEvent) => void;
85}
86
87export class VcsActionUnavailableError extends Schema.TaggedError<VcsActionUnavailableError>()(
88 "VcsActionUnavailableError",
89 {
90 operation: VcsActionOperation,
91 environmentId: Schema.NullOr(EnvironmentId),
92 cwd: Schema.NullOr(Schema.String),
93 },
94) {
95 override get message(): string {
96 return `Source control operation '${this.operation.replaceAll("_", " ")}' is unavailable.`;
97 }
98}
99
100export class VcsActionRemoteFailureError extends Schema.TaggedError<VcsActionRemoteFailureError>()(
101 "VcsActionRemoteFailureError",
102 {
103 actionId: Schema.String,
104 transportActionId: Schema.String,
105 action: GitStackedAction,
106 environmentId: EnvironmentId,
107 cwd: Schema.String,
108 phase: Schema.NullOr(GitActionProgressPhase),
109 remoteMessageLength: Schema.Number,
110 },
111) {
112 override get message(): string {
113 const phase = this.phase === null ? "execution" : this.phase;
114 return `Source control action '${this.action}' failed during ${phase}.`;
115 }
116}
117
118export class VcsActionMissingTerminalEventError extends Schema.TaggedError<VcsActionMissingTerminalEventError>()(
119 "VcsActionMissingTerminalEventError",
120 {
121 actionId: Schema.String,
122 transportActionId: Schema.String,
123 action: GitStackedAction,
124 environmentId: EnvironmentId,
125 cwd: Schema.String,
126 },
127) {
128 override get message(): string {
129 return `Source control action '${this.action}' ended without a terminal result.`;
130 }
131}
132
133export class VcsActionTargetKeyParseError extends Schema.TaggedError<VcsActionTargetKeyParseError>()(
134 "VcsActionTargetKeyParseError",
135 {
136 keyLength: Schema.Number,
137 cause: Schema.Defect(),
138 },
139) {
140 override get message(): string {
141 return `Invalid source control action target key (${this.keyLength} characters).`;
142 }
143}
144
145export const VcsActionExecutionError = Schema.Union([
146 VcsActionRemoteFailureError,
147 VcsActionMissingTerminalEventError,
148]);
149export type VcsActionExecutionError = typeof VcsActionExecutionError.Type;
150
151export const EMPTY_VCS_ACTION_STATE = Object.freeze<VcsActionState>({
152 isRunning: false,
153 operation: null,
154 actionId: null,
155 action: null,
156 currentLabel: null,
157 currentPhaseLabel: null,
158 hookName: null,
159 lastOutputLine: null,
160 phaseStartedAtMs: null,
161 hookStartedAtMs: null,
162 error: null,
163});
164
165const nowMs = (): number => DateTime.toEpochMillis(DateTime.nowUnsafe());
166let nextLocalActionId = 0;
167const decodeVcsActionTargetKey = Schema.decodeUnknownSync(
168 Schema.Tuple([EnvironmentId, Schema.String]),
169);
170
171const vcsActionStateAtom = Atom.family((key: string) => {
172 return Atom.make(EMPTY_VCS_ACTION_STATE).pipe(
173 Atom.keepAlive,
174 Atom.withLabel(`vcs-action:${key}`),
175 );
176});
177
178const EMPTY_VCS_ACTION_ATOM = Atom.make(EMPTY_VCS_ACTION_STATE).pipe(
179 Atom.keepAlive,
180 Atom.withLabel("vcs-action:null"),
181);
182
183function getVcsActionTargetKey(target: VcsActionTarget): string | null {
184 if (target.environmentId === null || target.cwd === null) {
185 return null;
186 }
187 return JSON.stringify([target.environmentId, target.cwd]);
188}
189
190export function parseVcsActionTargetKey(key: string): ResolvedVcsActionTarget {
191 try {
192 const [environmentId, cwd] = decodeVcsActionTargetKey(JSON.parse(key));
193 return { environmentId, cwd };
194 } catch (cause) {
195 throw new VcsActionTargetKeyParseError({ keyLength: key.length, cause });
196 }
197}
198
199function getVcsActionStateAtom(target: VcsActionTarget) {
200 const key = getVcsActionTargetKey(target);
201 return key === null ? EMPTY_VCS_ACTION_ATOM : vcsActionStateAtom(key);
202}
203
204function createLocalActionId(): string {
205 nextLocalActionId += 1;
206 return `local-vcs-action:${nextLocalActionId}`;
207}
208
209export function beginVcsActionState(
210 input: BeginVcsActionInput,
211): VcsActionState & { readonly actionId: string } {
212 const actionId = input.actionId ?? createLocalActionId();
213 const startedAt = nowMs();
214 return {
215 ...EMPTY_VCS_ACTION_STATE,
216 isRunning: true,
217 operation: input.operation,
218 actionId,
219 currentLabel: input.label,
220 currentPhaseLabel: input.label,
221 phaseStartedAtMs: startedAt,
222 };
223}
224
225function failVcsActionState(
226 operation: VcsActionOperation,
227 actionId: string,
228 error: unknown,
229): VcsActionState {
230 return {
231 ...EMPTY_VCS_ACTION_STATE,
232 operation,
233 actionId,
234 error: error instanceof Error ? error.message : "Source control action failed.",
235 };
236}
237
238export function createVcsActionTransportId(
239 target: ResolvedVcsActionTarget,
240 actionId: string,
241): string {
242 const targetKey = JSON.stringify([target.environmentId, target.cwd]);
243 return `${targetKey.length}:${targetKey}${actionId}`;
244}
245
246export function normalizeVcsActionProgressEvent(
247 target: ResolvedVcsActionTarget,
248 transportActionId: string,
249 actionId: string,
250 event: GitActionProgressEvent,
251): GitActionProgressEvent | null {
252 if (event.actionId !== transportActionId || event.cwd !== target.cwd) {
253 return null;
254 }
255 return {
256 ...event,
257 actionId,
258 };
259}
260
261export function consumeVcsActionProgress<E, R>(
262 stream: Stream.Stream<GitActionProgressEvent, E, R>,
263 input: {
264 readonly target: ResolvedVcsActionTarget;
265 readonly transportActionId: string;
266 readonly actionId: string;
267 readonly action: GitStackedAction;
268 readonly onProgress: (event: GitActionProgressEvent) => Effect.Effect<void>;
269 },
270): Effect.Effect<GitRunStackedActionResult, E | VcsActionExecutionError, R> {
271 return Effect.suspend(() => {
272 let terminalEvent: GitActionProgressEvent | null = null;
273 return stream.pipe(
274 Stream.runForEach((event) => {
275 const normalized = normalizeVcsActionProgressEvent(
276 input.target,
277 input.transportActionId,
278 input.actionId,
279 event,
280 );
281 if (normalized === null) {
282 return Effect.void;
283 }
284 if (normalized.kind === "action_finished" || normalized.kind === "action_failed") {
285 terminalEvent = normalized;
286 }
287 return input.onProgress(normalized);
288 }),
289 Effect.flatMap(() => {
290 if (terminalEvent?.kind === "action_finished") {
291 return Effect.succeed(terminalEvent.result);
292 }
293 if (terminalEvent?.kind === "action_failed") {
294 return Effect.fail<VcsActionExecutionError>(
295 new VcsActionRemoteFailureError({
296 actionId: input.actionId,
297 transportActionId: input.transportActionId,
298 action: terminalEvent.action,
299 environmentId: input.target.environmentId,
300 cwd: input.target.cwd,
301 phase: terminalEvent.phase,
302 remoteMessageLength: terminalEvent.message.length,
303 }),
304 );
305 }
306 return Effect.fail<VcsActionExecutionError>(
307 new VcsActionMissingTerminalEventError({
308 actionId: input.actionId,
309 transportActionId: input.transportActionId,
310 action: input.action,
311 environmentId: input.target.environmentId,
312 cwd: input.target.cwd,
313 }),
314 );
315 }),
316 );
317 });
318}
319
320export function applyVcsActionProgressEvent(
321 current: VcsActionState,
322 event: GitActionProgressEvent,
323): VcsActionState {
324 if (current.actionId !== event.actionId) {
325 return current;
326 }
327 const now = nowMs();
328
329 switch (event.kind) {
330 case "action_started":
331 return {
332 ...current,
333 isRunning: true,
334 actionId: event.actionId,
335 action: event.action,
336 operation: "run_change_request",
337 phaseStartedAtMs: now,
338 hookStartedAtMs: null,
339 hookName: null,
340 lastOutputLine: null,
341 error: null,
342 };
343 case "phase_started":
344 return {
345 ...current,
346 isRunning: true,
347 actionId: event.actionId,
348 action: event.action,
349 operation: "run_change_request",
350 currentLabel: event.label,
351 currentPhaseLabel: event.label,
352 phaseStartedAtMs: now,
353 hookStartedAtMs: null,
354 hookName: null,
355 lastOutputLine: null,
356 error: null,
357 };
358 case "hook_started":
359 return {
360 ...current,
361 isRunning: true,
362 actionId: event.actionId,
363 action: event.action,
364 operation: "run_change_request",
365 currentLabel: `Running ${event.hookName}...`,
366 hookName: event.hookName,
367 hookStartedAtMs: now,
368 lastOutputLine: null,
369 error: null,
370 };
371 case "hook_output":
372 return {
373 ...current,
374 isRunning: true,
375 actionId: event.actionId,
376 action: event.action,
377 operation: "run_change_request",
378 lastOutputLine: event.text,
379 error: null,
380 };
381 case "hook_finished":
382 return {
383 ...current,
384 isRunning: true,
385 actionId: event.actionId,
386 action: event.action,
387 operation: "run_change_request",
388 currentLabel: current.currentPhaseLabel,
389 hookName: null,
390 hookStartedAtMs: null,
391 lastOutputLine: null,
392 error: null,
393 };
394 case "action_finished":
395 return {
396 ...current,
397 isRunning: true,
398 actionId: event.actionId,
399 action: event.action,
400 operation: "run_change_request",
401 error: null,
402 };
403 case "action_failed":
404 return {
405 ...current,
406 isRunning: true,
407 actionId: event.actionId,
408 action: event.action,
409 operation: "run_change_request",
410 error: event.message,
411 };
412 }
413}
414
415export function createVcsActionManager<R, E>(
416 runtime: Atom.AtomRuntime<EnvironmentRegistry | Persistence.EnvironmentCacheStore | R, E>,
417) {
418 const runStackedActionCommands = new Map<
419 string,
420 AtomCommand<RunVcsStackedActionInput, GitRunStackedActionResult, unknown>
421 >();
422 const getRunStackedActionCommand = (requestedTarget: VcsActionTarget) => {
423 const targetKey = getVcsActionTargetKey(requestedTarget);
424 const commandKey =
425 targetKey ??
426 JSON.stringify([
427 "vcs-action-target:unavailable",
428 requestedTarget.environmentId,
429 requestedTarget.cwd,
430 ]);
431 const existing = runStackedActionCommands.get(commandKey);
432 if (existing !== undefined) {
433 return existing;
434 }
435 const target = targetKey === null ? null : parseVcsActionTargetKey(targetKey);
436 const stateAtom = targetKey === null ? EMPTY_VCS_ACTION_ATOM : vcsActionStateAtom(targetKey);
437 const command = createRuntimeCommand<
438 EnvironmentRegistry | Persistence.EnvironmentCacheStore | R,
439 E,
440 RunVcsStackedActionInput,
441 GitRunStackedActionResult,
442 unknown
443 >(runtime, {
444 label: `vcs-action:run-stacked:${commandKey}`,
445 scheduler: vcsCommandScheduler,
446 concurrency: { mode: "serial", key: () => commandKey },
447 execute: Effect.fn("VcsAction.runStackedAction")(function* (
448 input: RunVcsStackedActionInput,
449 registry,
450 ) {
451 if (target === null) {
452 return yield* new VcsActionUnavailableError({
453 operation: "run_change_request",
454 environmentId: requestedTarget.environmentId,
455 cwd: requestedTarget.cwd,
456 });
457 }
458 const transportActionId = createVcsActionTransportId(target, input.actionId);
459 registry.set(
460 stateAtom,
461 beginVcsActionState({
462 operation: "run_change_request",
463 label: "Running source control action",
464 actionId: input.actionId,
465 }),
466 );
467
468 const rpcInput: GitRunStackedActionInput = {
469 actionId: transportActionId,
470 cwd: target.cwd,
471 action: input.action,
472 ...(input.commitMessage ? { commitMessage: input.commitMessage } : {}),
473 ...(input.featureBranch ? { featureBranch: true } : {}),
474 ...(input.filePaths?.length ? { filePaths: [...input.filePaths] } : {}),
475 ...(input.threadId !== undefined ? { threadId: input.threadId } : {}),
476 ...(input.projectId !== undefined ? { projectId: input.projectId } : {}),
477 };
478 const clearOwnedState = Effect.sync(() => {
479 const current = registry.get(stateAtom);
480 if (current.actionId === input.actionId) {
481 registry.set(stateAtom, EMPTY_VCS_ACTION_STATE);
482 }
483 });
484 return yield* consumeVcsActionProgress(
485 runStreamInEnvironment(
486 target.environmentId,
487 runStreamGuarded(WS_METHODS.gitRunStackedAction, rpcInput),
488 ),
489 {
490 target,
491 transportActionId,
492 actionId: input.actionId,
493 action: input.action,
494 onProgress: (event) =>
495 Effect.sync(() => {
496 const current = registry.get(stateAtom);
497 if (current.actionId !== input.actionId) {
498 return;
499 }
500 registry.set(stateAtom, applyVcsActionProgressEvent(current, event));
501 if (input.onProgress !== undefined) {
502 try {
503 input.onProgress(event);
504 } catch {}
505 }
506 }),
507 },
508 ).pipe(
509 Effect.provideService(RpcPermissionGuard, {
510 authorize: (id, method, payload) =>
511 createCommandPermissions(runtime, method).authorize(registry, id, payload),
512 }),
513 Effect.ensuring(invalidateCachedVcsRefs(registry, target)),
514 Effect.tap(() => clearOwnedState),
515 Effect.tapError((error) =>
516 Effect.sync(() => {
517 const current = registry.get(stateAtom);
518 if (current.actionId === input.actionId && current.isRunning) {
519 registry.set(
520 stateAtom,
521 failVcsActionState("run_change_request", input.actionId, error),
522 );
523 }
524 }),
525 ),
526 Effect.onInterrupt(() => clearOwnedState),
527 );
528 }),
529 });
530 runStackedActionCommands.set(commandKey, command);
531 return command;
532 };
533
534 const setState = (
535 registry: AtomRegistry.AtomRegistry,
536 target: VcsActionTarget,
537 update: (current: VcsActionState) => VcsActionState,
538 ): void => {
539 const key = getVcsActionTargetKey(target);
540 if (key === null) {
541 return;
542 }
543 const stateAtom = vcsActionStateAtom(key);
544 registry.set(stateAtom, update(registry.get(stateAtom)));
545 };
546
547 return {
548 stateAtom: getVcsActionStateAtom,
549 runStackedAction: (target: VcsActionTarget) => getRunStackedActionCommand(target),
550 track: async <A, E>(
551 registry: AtomRegistry.AtomRegistry,
552 target: VcsActionTarget,
553 input: BeginVcsActionInput,
554 action: () => Promise<AtomCommandResult<A, E>>,
555 ): Promise<AtomCommandResult<A, E | VcsActionUnavailableError>> => {
556 const key = getVcsActionTargetKey(target);
557 if (key === null) {
558 return AsyncResult.failure<never, VcsActionUnavailableError>(
559 Cause.fail(
560 new VcsActionUnavailableError({
561 operation: input.operation,
562 environmentId: target.environmentId,
563 cwd: target.cwd,
564 }),
565 ),
566 );
567 }
568 const stateAtom = vcsActionStateAtom(key);
569 const next = beginVcsActionState(input);
570 registry.set(stateAtom, next);
571 const result = await action();
572 const current = registry.get(stateAtom);
573 if (current.actionId !== next.actionId) {
574 return result;
575 }
576 if (AsyncResult.isSuccess(result) || Cause.hasInterruptsOnly(result.cause)) {
577 registry.set(stateAtom, EMPTY_VCS_ACTION_STATE);
578 } else {
579 if (registry.get(stateAtom).actionId === next.actionId) {
580 registry.set(
581 stateAtom,
582 failVcsActionState(input.operation, next.actionId, Cause.squash(result.cause)),
583 );
584 }
585 }
586 return result;
587 },
588 resetError: (
589 registry: AtomRegistry.AtomRegistry,
590 target: VcsActionTarget,
591 operation: VcsActionOperation,
592 ): void => {
593 setState(registry, target, (current) =>
594 !current.isRunning && current.operation === operation ? EMPTY_VCS_ACTION_STATE : current,
595 );
596 },
597 };
598}