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 /** The thread the action runs beside; the server links a pull request it creates to it. */
83 readonly threadId?: ThreadId;
84 readonly projectId?: ProjectId;
85 readonly onProgress?: (event: GitActionProgressEvent) => void;
86}
87
88export class VcsActionUnavailableError extends Schema.TaggedError<VcsActionUnavailableError>()(
89 "VcsActionUnavailableError",
90 {
91 operation: VcsActionOperation,
92 environmentId: Schema.NullOr(EnvironmentId),
93 cwd: Schema.NullOr(Schema.String),
94 },
95) {
96 override get message(): string {
97 return `Source control operation '${this.operation.replaceAll("_", " ")}' is unavailable.`;
98 }
99}
100
101export class VcsActionRemoteFailureError extends Schema.TaggedError<VcsActionRemoteFailureError>()(
102 "VcsActionRemoteFailureError",
103 {
104 actionId: Schema.String,
105 transportActionId: Schema.String,
106 action: GitStackedAction,
107 environmentId: EnvironmentId,
108 cwd: Schema.String,
109 phase: Schema.NullOr(GitActionProgressPhase),
110 remoteMessageLength: Schema.Number,
111 },
112) {
113 override get message(): string {
114 const phase = this.phase === null ? "execution" : this.phase;
115 return `Source control action '${this.action}' failed during ${phase}.`;
116 }
117}
118
119export class VcsActionMissingTerminalEventError extends Schema.TaggedError<VcsActionMissingTerminalEventError>()(
120 "VcsActionMissingTerminalEventError",
121 {
122 actionId: Schema.String,
123 transportActionId: Schema.String,
124 action: GitStackedAction,
125 environmentId: EnvironmentId,
126 cwd: Schema.String,
127 },
128) {
129 override get message(): string {
130 return `Source control action '${this.action}' ended without a terminal result.`;
131 }
132}
133
134export class VcsActionTargetKeyParseError extends Schema.TaggedError<VcsActionTargetKeyParseError>()(
135 "VcsActionTargetKeyParseError",
136 {
137 keyLength: Schema.Number,
138 cause: Schema.Defect(),
139 },
140) {
141 override get message(): string {
142 return `Invalid source control action target key (${this.keyLength} characters).`;
143 }
144}
145
146export const VcsActionExecutionError = Schema.Union([
147 VcsActionRemoteFailureError,
148 VcsActionMissingTerminalEventError,
149]);
150export type VcsActionExecutionError = typeof VcsActionExecutionError.Type;
151
152export const EMPTY_VCS_ACTION_STATE = Object.freeze<VcsActionState>({
153 isRunning: false,
154 operation: null,
155 actionId: null,
156 action: null,
157 currentLabel: null,
158 currentPhaseLabel: null,
159 hookName: null,
160 lastOutputLine: null,
161 phaseStartedAtMs: null,
162 hookStartedAtMs: null,
163 error: null,
164});
165
166const nowMs = (): number => DateTime.toEpochMillis(DateTime.nowUnsafe());
167let nextLocalActionId = 0;
168const decodeVcsActionTargetKey = Schema.decodeUnknownSync(
169 Schema.Tuple([EnvironmentId, Schema.String]),
170);
171
172const vcsActionStateAtom = Atom.family((key: string) => {
173 return Atom.make(EMPTY_VCS_ACTION_STATE).pipe(
174 Atom.keepAlive,
175 Atom.withLabel(`vcs-action:${key}`),
176 );
177});
178
179const EMPTY_VCS_ACTION_ATOM = Atom.make(EMPTY_VCS_ACTION_STATE).pipe(
180 Atom.keepAlive,
181 Atom.withLabel("vcs-action:null"),
182);
183
184export function getVcsActionTargetKey(target: VcsActionTarget): string | null {
185 if (target.environmentId === null || target.cwd === null) {
186 return null;
187 }
188 return JSON.stringify([target.environmentId, target.cwd]);
189}
190
191export function parseVcsActionTargetKey(key: string): ResolvedVcsActionTarget {
192 try {
193 const [environmentId, cwd] = decodeVcsActionTargetKey(JSON.parse(key));
194 return { environmentId, cwd };
195 } catch (cause) {
196 throw new VcsActionTargetKeyParseError({ keyLength: key.length, cause });
197 }
198}
199
200function getVcsActionStateAtom(target: VcsActionTarget) {
201 const key = getVcsActionTargetKey(target);
202 return key === null ? EMPTY_VCS_ACTION_ATOM : vcsActionStateAtom(key);
203}
204
205function createLocalActionId(): string {
206 nextLocalActionId += 1;
207 return `local-vcs-action:${nextLocalActionId}`;
208}
209
210export function beginVcsActionState(
211 input: BeginVcsActionInput,
212): VcsActionState & { readonly actionId: string } {
213 const actionId = input.actionId ?? createLocalActionId();
214 const startedAt = nowMs();
215 return {
216 ...EMPTY_VCS_ACTION_STATE,
217 isRunning: true,
218 operation: input.operation,
219 actionId,
220 currentLabel: input.label,
221 currentPhaseLabel: input.label,
222 phaseStartedAtMs: startedAt,
223 };
224}
225
226function failVcsActionState(
227 operation: VcsActionOperation,
228 actionId: string,
229 error: unknown,
230): VcsActionState {
231 return {
232 ...EMPTY_VCS_ACTION_STATE,
233 operation,
234 actionId,
235 error: error instanceof Error ? error.message : "Source control action failed.",
236 };
237}
238
239export function createVcsActionTransportId(
240 target: ResolvedVcsActionTarget,
241 actionId: string,
242): string {
243 const targetKey = JSON.stringify([target.environmentId, target.cwd]);
244 return `${targetKey.length}:${targetKey}${actionId}`;
245}
246
247export function normalizeVcsActionProgressEvent(
248 target: ResolvedVcsActionTarget,
249 transportActionId: string,
250 actionId: string,
251 event: GitActionProgressEvent,
252): GitActionProgressEvent | null {
253 if (event.actionId !== transportActionId || event.cwd !== target.cwd) {
254 return null;
255 }
256 return {
257 ...event,
258 actionId,
259 };
260}
261
262export function consumeVcsActionProgress<E, R>(
263 stream: Stream.Stream<GitActionProgressEvent, E, R>,
264 input: {
265 readonly target: ResolvedVcsActionTarget;
266 readonly transportActionId: string;
267 readonly actionId: string;
268 readonly action: GitStackedAction;
269 readonly onProgress: (event: GitActionProgressEvent) => Effect.Effect<void>;
270 },
271): Effect.Effect<GitRunStackedActionResult, E | VcsActionExecutionError, R> {
272 return Effect.suspend(() => {
273 let terminalEvent: GitActionProgressEvent | null = null;
274 return stream.pipe(
275 Stream.runForEach((event) => {
276 const normalized = normalizeVcsActionProgressEvent(
277 input.target,
278 input.transportActionId,
279 input.actionId,
280 event,
281 );
282 if (normalized === null) {
283 return Effect.void;
284 }
285 if (normalized.kind === "action_finished" || normalized.kind === "action_failed") {
286 terminalEvent = normalized;
287 }
288 return input.onProgress(normalized);
289 }),
290 Effect.flatMap(() => {
291 if (terminalEvent?.kind === "action_finished") {
292 return Effect.succeed(terminalEvent.result);
293 }
294 if (terminalEvent?.kind === "action_failed") {
295 return Effect.fail<VcsActionExecutionError>(
296 new VcsActionRemoteFailureError({
297 actionId: input.actionId,
298 transportActionId: input.transportActionId,
299 action: terminalEvent.action,
300 environmentId: input.target.environmentId,
301 cwd: input.target.cwd,
302 phase: terminalEvent.phase,
303 remoteMessageLength: terminalEvent.message.length,
304 }),
305 );
306 }
307 return Effect.fail<VcsActionExecutionError>(
308 new VcsActionMissingTerminalEventError({
309 actionId: input.actionId,
310 transportActionId: input.transportActionId,
311 action: input.action,
312 environmentId: input.target.environmentId,
313 cwd: input.target.cwd,
314 }),
315 );
316 }),
317 );
318 });
319}
320
321export function applyVcsActionProgressEvent(
322 current: VcsActionState,
323 event: GitActionProgressEvent,
324): VcsActionState {
325 if (current.actionId !== event.actionId) {
326 return current;
327 }
328 const now = nowMs();
329
330 switch (event.kind) {
331 case "action_started":
332 return {
333 ...current,
334 isRunning: true,
335 actionId: event.actionId,
336 action: event.action,
337 operation: "run_change_request",
338 phaseStartedAtMs: now,
339 hookStartedAtMs: null,
340 hookName: null,
341 lastOutputLine: null,
342 error: null,
343 };
344 case "phase_started":
345 return {
346 ...current,
347 isRunning: true,
348 actionId: event.actionId,
349 action: event.action,
350 operation: "run_change_request",
351 currentLabel: event.label,
352 currentPhaseLabel: event.label,
353 phaseStartedAtMs: now,
354 hookStartedAtMs: null,
355 hookName: null,
356 lastOutputLine: null,
357 error: null,
358 };
359 case "hook_started":
360 return {
361 ...current,
362 isRunning: true,
363 actionId: event.actionId,
364 action: event.action,
365 operation: "run_change_request",
366 currentLabel: `Running ${event.hookName}...`,
367 hookName: event.hookName,
368 hookStartedAtMs: now,
369 lastOutputLine: null,
370 error: null,
371 };
372 case "hook_output":
373 return {
374 ...current,
375 isRunning: true,
376 actionId: event.actionId,
377 action: event.action,
378 operation: "run_change_request",
379 lastOutputLine: event.text,
380 error: null,
381 };
382 case "hook_finished":
383 return {
384 ...current,
385 isRunning: true,
386 actionId: event.actionId,
387 action: event.action,
388 operation: "run_change_request",
389 currentLabel: current.currentPhaseLabel,
390 hookName: null,
391 hookStartedAtMs: null,
392 lastOutputLine: null,
393 error: null,
394 };
395 case "action_finished":
396 return {
397 ...current,
398 isRunning: true,
399 actionId: event.actionId,
400 action: event.action,
401 operation: "run_change_request",
402 error: null,
403 };
404 case "action_failed":
405 return {
406 ...current,
407 isRunning: true,
408 actionId: event.actionId,
409 action: event.action,
410 operation: "run_change_request",
411 error: event.message,
412 };
413 }
414}
415
416export function createVcsActionManager<R, E>(
417 runtime: Atom.AtomRuntime<EnvironmentRegistry | Persistence.EnvironmentCacheStore | R, E>,
418) {
419 const runStackedActionCommands = new Map<
420 string,
421 AtomCommand<RunVcsStackedActionInput, GitRunStackedActionResult, unknown>
422 >();
423 const getRunStackedActionCommand = (requestedTarget: VcsActionTarget) => {
424 const targetKey = getVcsActionTargetKey(requestedTarget);
425 const commandKey =
426 targetKey ??
427 JSON.stringify([
428 "vcs-action-target:unavailable",
429 requestedTarget.environmentId,
430 requestedTarget.cwd,
431 ]);
432 const existing = runStackedActionCommands.get(commandKey);
433 if (existing !== undefined) {
434 return existing;
435 }
436 const target = targetKey === null ? null : parseVcsActionTargetKey(targetKey);
437 const stateAtom = targetKey === null ? EMPTY_VCS_ACTION_ATOM : vcsActionStateAtom(targetKey);
438 const command = createRuntimeCommand<
439 EnvironmentRegistry | Persistence.EnvironmentCacheStore | R,
440 E,
441 RunVcsStackedActionInput,
442 GitRunStackedActionResult,
443 unknown
444 >(runtime, {
445 label: `vcs-action:run-stacked:${commandKey}`,
446 scheduler: vcsCommandScheduler,
447 concurrency: { mode: "serial", key: () => commandKey },
448 execute: (input: RunVcsStackedActionInput, registry) => {
449 if (target === null) {
450 return Effect.fail(
451 new VcsActionUnavailableError({
452 operation: "run_change_request",
453 environmentId: requestedTarget.environmentId,
454 cwd: requestedTarget.cwd,
455 }),
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 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 // Presentation callbacks must not fail the source-control operation.
506 }
507 }
508 }),
509 },
510 ).pipe(
511 Effect.provideService(RpcPermissionGuard, {
512 authorize: (id, method, payload) =>
513 createCommandPermissions(runtime, method).authorize(registry, id, payload),
514 }),
515 Effect.ensuring(invalidateCachedVcsRefs(registry, target)),
516 Effect.tap(() => clearOwnedState),
517 Effect.tapError((error) =>
518 Effect.sync(() => {
519 const current = registry.get(stateAtom);
520 if (current.actionId === input.actionId && current.isRunning) {
521 registry.set(
522 stateAtom,
523 failVcsActionState("run_change_request", input.actionId, error),
524 );
525 }
526 }),
527 ),
528 Effect.onInterrupt(() => clearOwnedState),
529 );
530 },
531 });
532 runStackedActionCommands.set(commandKey, command);
533 return command;
534 };
535
536 const setState = (
537 registry: AtomRegistry.AtomRegistry,
538 target: VcsActionTarget,
539 update: (current: VcsActionState) => VcsActionState,
540 ): void => {
541 const key = getVcsActionTargetKey(target);
542 if (key === null) {
543 return;
544 }
545 const stateAtom = vcsActionStateAtom(key);
546 registry.set(stateAtom, update(registry.get(stateAtom)));
547 };
548
549 return {
550 stateAtom: getVcsActionStateAtom,
551 runStackedAction: (target: VcsActionTarget) => getRunStackedActionCommand(target),
552 track: async <A, E>(
553 registry: AtomRegistry.AtomRegistry,
554 target: VcsActionTarget,
555 input: BeginVcsActionInput,
556 action: () => Promise<AtomCommandResult<A, E>>,
557 ): Promise<AtomCommandResult<A, E | VcsActionUnavailableError>> => {
558 const key = getVcsActionTargetKey(target);
559 if (key === null) {
560 return AsyncResult.failure<never, VcsActionUnavailableError>(
561 Cause.fail(
562 new VcsActionUnavailableError({
563 operation: input.operation,
564 environmentId: target.environmentId,
565 cwd: target.cwd,
566 }),
567 ),
568 );
569 }
570 const stateAtom = vcsActionStateAtom(key);
571 const next = beginVcsActionState(input);
572 registry.set(stateAtom, next);
573 const result = await action();
574 const current = registry.get(stateAtom);
575 if (current.actionId !== next.actionId) {
576 return result;
577 }
578 if (AsyncResult.isSuccess(result) || Cause.hasInterruptsOnly(result.cause)) {
579 registry.set(stateAtom, EMPTY_VCS_ACTION_STATE);
580 } else {
581 if (registry.get(stateAtom).actionId === next.actionId) {
582 registry.set(
583 stateAtom,
584 failVcsActionState(input.operation, next.actionId, Cause.squash(result.cause)),
585 );
586 }
587 }
588 return result;
589 },
590 resetError: (
591 registry: AtomRegistry.AtomRegistry,
592 target: VcsActionTarget,
593 operation: VcsActionOperation,
594 ): void => {
595 setState(registry, target, (current) =>
596 !current.isRunning && current.operation === operation ? EMPTY_VCS_ACTION_STATE : current,
597 );
598 },
599 };
600}