1import { CommandId } from "@t3tools/contracts";
2import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid";
3import * as Effect from "effect/Effect";
4import * as FileSystem from "effect/FileSystem";
5import * as Layer from "effect/Layer";
6import * as Stream from "effect/Stream";
7
8import * as GitWorkflowService from "../git/GitWorkflowService.ts";
9import { forkParked } from "../serverActivation.ts";
10import * as VcsStatusBroadcaster from "../vcs/VcsStatusBroadcaster.ts";
11import * as ProjectionStore from "./ProjectionStore.ts";
12import * as ThreadManagement from "./ThreadManagementService.ts";
13
14const make = Effect.gen(function* () {
15 const threads = yield* ThreadManagement.ThreadManagementService;
16 const git = yield* GitWorkflowService.GitWorkflowService;
17 const projections = yield* ProjectionStore.ProjectionStoreV2;
18 const fs = yield* FileSystem.FileSystem;
19 return Effect.fn("WorktreeBranchSync.reconcile")(function* (input: {
20 readonly cwd: string;
21 readonly refName: string | null;
22 }) {
23 if (input.refName === null) return;
24 const paths = yield* projections.getThreadWorktreePaths({ includeArchived: true });
25 const matchingPaths = new Set<string>();
26 for (const worktreePath of new Set(paths.map((thread) => thread.worktreePath))) {
27 const realPath =
28 worktreePath === input.cwd
29 ? worktreePath
30 : yield* fs.realPath(worktreePath).pipe(Effect.orElseSucceed(() => null));
31 if (realPath === input.cwd) matchingPaths.add(worktreePath);
32 }
33 const matchingThreads = new Map(
34 paths
35 .filter((thread) => matchingPaths.has(thread.worktreePath))
36 .map((thread) => [thread.threadId, thread.worktreePath]),
37 );
38 if (matchingThreads.size === 0) return;
39 const snapshot = yield* threads.getShellSnapshot({ threadIds: [...matchingThreads.keys()] });
40 const sharing = [...snapshot.threads, ...snapshot.archivedThreads].filter(
41 (thread) =>
42 thread.worktreePath === matchingThreads.get(thread.id) && thread.branch !== input.refName,
43 );
44 if (sharing.length === 0) return;
45 // Status events may queue behind a generated rename or another checkout.
46 if ((yield* git.currentBranch(input.cwd)) !== input.refName) return;
47 for (const thread of sharing) {
48 yield* threads
49 .dispatch({
50 type: "thread.metadata.update",
51 commandId: CommandId.make(yield* randomUuidV4),
52 threadId: thread.id,
53 branch: input.refName,
54 expectedBranch: thread.branch,
55 expectedWorktreePath: thread.worktreePath,
56 })
57 .pipe(
58 Effect.catchTags({
59 OrchestratorDispatchError: (cause) =>
60 Effect.logWarning("Worktree branch update was rejected", {
61 threadId: thread.id,
62 cause,
63 }),
64 }),
65 );
66 }
67 });
68});
69
70export const layer = Layer.effectDiscard(
71 Effect.gen(function* () {
72 const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
73 const reconcile = yield* make;
74 yield* forkParked(
75 Stream.runForEach(broadcaster.streamBranchChanges, (input) =>
76 reconcile(input).pipe(
77 Effect.catchCause((cause) =>
78 Effect.logWarning("Failed to reconcile worktree branch", { cwd: input.cwd, cause }),
79 ),
80 ),
81 ),
82 );
83 }),
84);