1import { SourceControlProviderKind } from "@t3tools/contracts";
2import type {
3 ProjectCloneSnapshot,
4 ProjectCloneStage,
5 ProjectCloneStartInput,
6 ProjectCloneStartResult,
7 ProjectId,
8 SourceControlCloneProtocol,
9 SourceControlRepositoryInfo,
10} from "@t3tools/contracts";
11import {
12 OrchestrationDispatchCommandError,
13 PROJECT_CLONE_DETAIL_MAX_LENGTH,
14 PROJECT_CLONE_ERROR_MAX_LENGTH,
15 SourceControlRepositoryError,
16} from "@t3tools/contracts";
17import * as Cause from "effect/Cause";
18import * as Context from "effect/Context";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as Exit from "effect/Exit";
22import * as Fiber from "effect/Fiber";
23import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";
25import * as PubSub from "effect/PubSub";
26import * as Queue from "effect/Queue";
27import * as Ref from "effect/Ref";
28import * as Schema from "effect/Schema";
29import * as Semaphore from "effect/Semaphore";
30import * as Scope from "effect/Scope";
31import * as Stream from "effect/Stream";
32
33import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts";
34import type { ProjectConflictError } from "./ProjectService.ts";
35
36export class ProjectCloneTracker extends Context.Service<
37 ProjectCloneTracker,
38 {
39 readonly start: (
40 input: ProjectCloneStartInput,
41 hooks: ProjectCloneHooks,
42 ) => Effect.Effect<
43 ProjectCloneStartResult,
44 SourceControlRepositoryError | OrchestrationDispatchCommandError
45 >;
46 /** Interrupts a running clone and deletes the partial checkout. */
47 readonly cancel: (projectId: ProjectId) => Effect.Effect<boolean>;
48 readonly retry: (projectId: ProjectId) => Effect.Effect<boolean, SourceControlRepositoryError>;
49 /** Stops tracking, interrupts any running clone and removes unfinished checkouts. */
50 readonly discard: (projectId: ProjectId) => Effect.Effect<void>;
51 readonly get: (projectId: ProjectId) => Effect.Effect<ProjectCloneSnapshot | null>;
52 readonly stream: Stream.Stream<ReadonlyArray<ProjectCloneSnapshot>>;
53 }
54>()("t3/project/ProjectCloneTracker") {}
55
56export interface ProjectCloneHooks {
57 /** Fails with `ProjectConflictError` when another project already owns the destination. */
58 readonly createProject: (input: {
59 readonly projectId: ProjectId;
60 readonly title: string;
61 readonly workspaceRoot: string;
62 readonly createdAt: string;
63 }) => Effect.Effect<void, OrchestrationDispatchCommandError | ProjectConflictError>;
64 readonly onCloned: (input: {
65 readonly projectId: ProjectId;
66 readonly workspaceRoot: string;
67 }) => Effect.Effect<void>;
68}
69
70const DONE_RETENTION = "30 seconds";
71// Git prints nothing while a transfer waits on the network, and the kernel can
72// take 15+ minutes to give up on a dead connection. Without this the banner
73// keeps showing the last transfer speed as if data were still arriving.
74const STALL_AFTER = "20 seconds";
75
76function clampText(text: string, maxLength: number): string {
77 return text.length <= maxLength ? text : `${text.slice(0, maxLength - 1)}…`;
78}
79
80const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
81
82interface TrackedClone {
83 readonly snapshot: ProjectCloneSnapshot;
84 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
85 readonly hooks: ProjectCloneHooks;
86 readonly input: {
87 /** What git is given; may carry credentials and never leaves the server. */
88 readonly cloneUrl: string;
89 readonly destinationPath: string;
90 readonly repository: SourceControlRepositoryInfo | null;
91 readonly protocol: SourceControlCloneProtocol;
92 };
93}
94
95/** @public */
96export const make = Effect.gen(function* () {
97 const repositories = yield* SourceControlRepositoryService.SourceControlRepositoryService;
98 const clones = yield* Ref.make(new Map<ProjectId, TrackedClone>());
99 const changes = yield* PubSub.unbounded<ReadonlyArray<ProjectCloneSnapshot>>();
100 const retentionFibers = new Map<ProjectId, Fiber.Fiber<unknown, never>>();
101 let sequence = 0;
102 const cloneScope = yield* Scope.make("parallel");
103 yield* Effect.addFinalizer(() => Scope.close(cloneScope, Exit.void));
104 // Serialize actions on the shared entry and directory so retries cannot race clones.
105 const actionLock = yield* Semaphore.make(1);
106 const locked = <A, E, R>(effect: Effect.Effect<A, E, R>) => actionLock.withPermits(1)(effect);
107
108 const list = Ref.get(clones).pipe(
109 Effect.map((current) => Array.from(current.values(), (tracked) => tracked.snapshot)),
110 );
111 const publish = list.pipe(Effect.flatMap((snapshots) => PubSub.publish(changes, snapshots)));
112
113 const modify = (
114 projectId: ProjectId,
115 mutate: (tracked: TrackedClone) => TrackedClone,
116 ): Effect.Effect<ProjectCloneSnapshot | null> =>
117 Ref.modify(clones, (current) => {
118 const existing = current.get(projectId);
119 if (!existing) return [null, current] as const;
120 const nextTracked = mutate(existing);
121 const nextSnapshot = { ...nextTracked.snapshot, sequence: ++sequence };
122 const next = new Map(current);
123 next.set(projectId, { ...nextTracked, snapshot: nextSnapshot });
124 return [nextSnapshot, next] as const;
125 }).pipe(Effect.tap((snapshot) => (snapshot ? publish : Effect.void)));
126
127 const clearRetention = (projectId: ProjectId) => {
128 const fiber = retentionFibers.get(projectId);
129 retentionFibers.delete(projectId);
130 return fiber ? Fiber.interrupt(fiber).pipe(Effect.ignore) : Effect.void;
131 };
132
133 const remove = (projectId: ProjectId) =>
134 Ref.update(clones, (current) => {
135 if (!current.has(projectId)) return current;
136 const next = new Map(current);
137 next.delete(projectId);
138 return next;
139 }).pipe(Effect.andThen(publish));
140
141 const scheduleRemoval = Effect.fn("ProjectCloneTracker.scheduleRemoval")(function* (
142 projectId: ProjectId,
143 ) {
144 yield* clearRetention(projectId);
145 const fiber = yield* remove(projectId).pipe(
146 Effect.delay(DONE_RETENTION),
147 Effect.ensuring(
148 Effect.sync(() => {
149 if (retentionFibers.get(projectId) === fiber) retentionFibers.delete(projectId);
150 }),
151 ),
152 Effect.forkDetach,
153 );
154 retentionFibers.set(projectId, fiber);
155 });
156
157 const progress = (
158 projectId: ProjectId,
159 update: {
160 readonly stage: ProjectCloneStage;
161 readonly percent: number | null;
162 readonly detail: string | null;
163 },
164 ) =>
165 modify(projectId, (tracked) => ({
166 ...tracked,
167 snapshot: {
168 ...tracked.snapshot,
169 stage: update.stage,
170 percent: update.percent,
171 detail:
172 update.detail === null ? null : clampText(update.detail, PROJECT_CLONE_DETAIL_MAX_LENGTH),
173 stalled: false,
174 },
175 })).pipe(Effect.asVoid);
176
177 const markStalled = (projectId: ProjectId) =>
178 modify(projectId, (tracked) => ({
179 ...tracked,
180 snapshot: { ...tracked.snapshot, stalled: true },
181 })).pipe(Effect.asVoid);
182
183 /** Marks the clone stalled once, then waits for git to speak again. */
184 const watchForStall = (projectId: ProjectId, pulses: Queue.Dequeue<void>) =>
185 Queue.take(pulses).pipe(
186 Effect.timeoutOption(STALL_AFTER),
187 Effect.flatMap((pulse) =>
188 Option.isSome(pulse)
189 ? Effect.void
190 : markStalled(projectId).pipe(Effect.andThen(Queue.take(pulses))),
191 ),
192 Effect.forever,
193 );
194
195 const finish = Effect.fn("ProjectCloneTracker.finish")(function* (
196 projectId: ProjectId,
197 phase: "done" | "failed" | "cancelled",
198 error: string | null,
199 ) {
200 const endedAt = yield* nowIso;
201 yield* modify(projectId, (tracked) => ({
202 ...tracked,
203 fiber: null,
204 snapshot: {
205 ...tracked.snapshot,
206 phase,
207 endedAt,
208 percent: phase === "done" ? 100 : tracked.snapshot.percent,
209 stalled: false,
210 error: error === null ? null : clampText(error, PROJECT_CLONE_ERROR_MAX_LENGTH),
211 },
212 }));
213 if (phase === "done") yield* scheduleRemoval(projectId);
214 });
215
216 const cloneWatchingForStall = Effect.fn("ProjectCloneTracker.cloneWatchingForStall")(function* (
217 projectId: ProjectId,
218 tracked: TrackedClone,
219 ) {
220 const pulses = yield* Queue.sliding<void>(1);
221 yield* watchForStall(projectId, pulses).pipe(Effect.forkScoped);
222 return yield* repositories.cloneRepository(
223 {
224 remoteUrl: tracked.input.cloneUrl,
225 destinationPath: tracked.input.destinationPath,
226 protocol: tracked.input.protocol,
227 },
228 {
229 onProgress: (update) =>
230 Queue.offer(pulses, undefined).pipe(Effect.andThen(progress(projectId, update))),
231 timeoutMs: null,
232 },
233 );
234 }, Effect.scoped);
235
236 /**
237 * `onExit` records outcomes even on interruption; `matchCause` would be skipped.
238 * Mark the clone done before the post-clone hook so a late Cancel cannot remove it.
239 */
240 const runClone = (projectId: ProjectId, tracked: TrackedClone) =>
241 cloneWatchingForStall(projectId, tracked).pipe(
242 Effect.onExit((exit) =>
243 Exit.isSuccess(exit)
244 ? finish(projectId, "done", null)
245 : Cause.hasInterruptsOnly(exit.cause)
246 ? finish(projectId, "cancelled", null)
247 : finish(projectId, "failed", describeCloneFailure(exit.cause)),
248 ),
249 Effect.flatMap(() =>
250 tracked.hooks
251 .onCloned({ projectId, workspaceRoot: tracked.input.destinationPath })
252 .pipe(Effect.ignoreCause({ log: true })),
253 ),
254 Effect.ignoreCause(),
255 );
256
257 const launch = Effect.fn("ProjectCloneTracker.launch")(function* (projectId: ProjectId) {
258 const current = yield* Ref.get(clones);
259 const tracked = current.get(projectId);
260 if (!tracked) return;
261 const fiber = yield* runClone(projectId, tracked).pipe(Effect.forkIn(cloneScope));
262 yield* Ref.update(clones, (map) => {
263 const existing = map.get(projectId);
264 if (!existing) return map;
265 const next = new Map(map);
266 next.set(projectId, { ...existing, fiber });
267 return next;
268 });
269 });
270
271 const moveClaim = (from: ProjectId, to: ProjectId) =>
272 Ref.modify(clones, (current) => {
273 const claim = current.get(from);
274 if (!claim || current.has(to)) return [false, current] as const;
275 const next = new Map(current);
276 next.delete(from);
277 next.set(to, { ...claim, snapshot: { ...claim.snapshot, projectId: to } });
278 return [true, next] as const;
279 });
280
281 const start: ProjectCloneTracker["Service"]["start"] = Effect.fn("ProjectCloneTracker.start")(
282 function* (input, hooks) {
283 const prepared = yield* repositories.prepareClone(input);
284 const startedAt = yield* nowIso;
285 const snapshot: ProjectCloneSnapshot = {
286 projectId: input.projectId,
287 remoteUrl: prepared.remoteUrl,
288 destinationPath: prepared.destinationPath,
289 repository: prepared.repository,
290 phase: "running",
291 stage: "connecting",
292 percent: null,
293 detail: null,
294 stalled: false,
295 error: null,
296 startedAt,
297 endedAt: null,
298 sequence: ++sequence,
299 };
300 const claimed = yield* Ref.modify(clones, (current) => {
301 const conflict = Array.from(current.values()).some(
302 (tracked) =>
303 tracked.snapshot.projectId === input.projectId ||
304 tracked.input.destinationPath === prepared.destinationPath,
305 );
306 if (conflict) return [false, current] as const;
307 const next = new Map(current);
308 next.set(input.projectId, {
309 snapshot,
310 fiber: null,
311 hooks,
312 input: {
313 cloneUrl: prepared.cloneUrl,
314 destinationPath: prepared.destinationPath,
315 repository: prepared.repository,
316 protocol: input.protocol ?? "auto",
317 },
318 });
319 return [true, next] as const;
320 });
321 const alreadyInProgress = new SourceControlRepositoryError({
322 operation: "cloneRepository",
323 provider: input.provider ?? SourceControlProviderKind.make("unknown"),
324 detail: "A clone into this destination is already in progress.",
325 });
326 if (!claimed) return yield* alreadyInProgress;
327 // Finish or release the claim despite disconnects; a claim without a fiber cannot be cancelled or retried.
328 const projectId = yield* Effect.uninterruptible(
329 Effect.gen(function* () {
330 yield* clearRetention(input.projectId);
331 // Register before creation so racing thread.create and project.delete see the clone.
332 const projectId = yield* hooks
333 .createProject({
334 projectId: input.projectId,
335 title: input.title,
336 workspaceRoot: prepared.destinationPath,
337 createdAt: input.createdAt,
338 })
339 .pipe(
340 Effect.as(input.projectId),
341 // A failed clone's project still owns its empty folder after a restart
342 // forgets the clone; clone for that project instead.
343 Effect.catchTags({
344 ProjectConflictError: (conflict) =>
345 moveClaim(input.projectId, conflict.conflictingProjectId).pipe(
346 Effect.flatMap((moved) =>
347 moved ? Effect.succeed(conflict.conflictingProjectId) : alreadyInProgress,
348 ),
349 ),
350 }),
351 );
352 yield* publish;
353 yield* launch(projectId);
354 return projectId;
355 }).pipe(Effect.tapError(() => remove(input.projectId))),
356 );
357 return {
358 projectId,
359 cwd: prepared.destinationPath,
360 remoteUrl: prepared.remoteUrl,
361 repository: prepared.repository,
362 };
363 },
364 );
365
366 const get: ProjectCloneTracker["Service"]["get"] = (projectId) =>
367 Ref.get(clones).pipe(Effect.map((current) => current.get(projectId)?.snapshot ?? null));
368
369 const cancel: ProjectCloneTracker["Service"]["cancel"] = Effect.fn("ProjectCloneTracker.cancel")(
370 function* (projectId) {
371 const current = yield* Ref.get(clones);
372 const tracked = current.get(projectId);
373 if (!tracked || tracked.snapshot.phase !== "running" || !tracked.fiber) return false;
374 // Finish cancellation despite disconnects to avoid dead fibers behind running snapshots.
375 yield* Effect.uninterruptible(
376 Effect.gen(function* () {
377 yield* Fiber.interrupt(tracked.fiber!);
378 const after = yield* get(projectId);
379 if (after?.phase === "done") return;
380 if (after?.phase === "running") yield* finish(projectId, "cancelled", null);
381 yield* repositories.discardClone(tracked.input.destinationPath).pipe(Effect.ignore);
382 }),
383 );
384 return true;
385 },
386 );
387
388 const retry: ProjectCloneTracker["Service"]["retry"] = Effect.fn("ProjectCloneTracker.retry")(
389 function* (projectId) {
390 const current = yield* Ref.get(clones);
391 const tracked = current.get(projectId);
392 if (
393 !tracked ||
394 (tracked.snapshot.phase !== "failed" && tracked.snapshot.phase !== "cancelled")
395 ) {
396 return false;
397 }
398 yield* repositories.discardClone(tracked.input.destinationPath);
399 const startedAt = yield* nowIso;
400 yield* modify(projectId, (entry) => ({
401 ...entry,
402 snapshot: {
403 ...entry.snapshot,
404 phase: "running",
405 stage: "connecting",
406 percent: null,
407 detail: null,
408 stalled: false,
409 error: null,
410 startedAt,
411 endedAt: null,
412 },
413 }));
414 yield* launch(projectId);
415 return true;
416 },
417 );
418
419 const discard: ProjectCloneTracker["Service"]["discard"] = Effect.fn(
420 "ProjectCloneTracker.discard",
421 )(function* (projectId) {
422 const current = yield* Ref.get(clones);
423 const tracked = current.get(projectId);
424 if (!tracked) return;
425 if (tracked.fiber) yield* Fiber.interrupt(tracked.fiber);
426 // Git may have finished while the interrupt was landing; the project
427 // is going away either way, but a complete checkout is the user's.
428 const after = yield* get(projectId);
429 if (after?.phase !== "done") {
430 yield* repositories.discardClone(tracked.input.destinationPath).pipe(Effect.ignore);
431 }
432 yield* clearRetention(projectId);
433 yield* remove(projectId);
434 });
435
436 const stream: ProjectCloneTracker["Service"]["stream"] = Stream.callback<
437 ReadonlyArray<ProjectCloneSnapshot>
438 >(
439 (mailbox) =>
440 Effect.gen(function* () {
441 const subscription = yield* PubSub.subscribe(changes);
442 Queue.offerUnsafe(mailbox, yield* list);
443 yield* Stream.fromSubscription(subscription).pipe(
444 Stream.runForEach((snapshots) =>
445 Effect.sync(() => Queue.offerUnsafe(mailbox, snapshots)),
446 ),
447 Effect.forkScoped,
448 );
449 }),
450 { bufferSize: 1, strategy: "sliding" },
451 );
452
453 // Forwarders retain the underlying operation spans.
454 return ProjectCloneTracker.of({
455 start: Effect.fnUntraced(function* (input, hooks) {
456 return yield* locked(start(input, hooks));
457 }),
458 cancel: Effect.fnUntraced(function* (projectId) {
459 return yield* locked(cancel(projectId));
460 }),
461 retry: Effect.fnUntraced(function* (projectId) {
462 return yield* locked(retry(projectId));
463 }),
464 discard: Effect.fnUntraced(function* (projectId) {
465 return yield* locked(discard(projectId));
466 }),
467 get,
468 stream,
469 });
470});
471
472const isSourceControlRepositoryError = Schema.is(SourceControlRepositoryError);
473
474/**
475 * Every dispatch transport (WebSocket, HTTP) runs this before normalizing so no
476 * client can start a thread on an empty tree, and no attachment copies are made
477 * for a command that is about to be refused.
478 */
479export const rejectCommandsDuringClone = Effect.fn("ProjectCloneTracker.rejectCommandsDuringClone")(
480 function* (
481 tracker: ProjectCloneTracker["Service"],
482 command: {
483 readonly type: string;
484 readonly projectId?: ProjectId;
485 readonly bootstrap?: unknown;
486 },
487 ): Effect.fn.Return<void, OrchestrationDispatchCommandError> {
488 const projectId =
489 command.type === "thread.create"
490 ? (command.projectId ?? null)
491 : command.type === "thread.turn.start"
492 ? bootstrapProjectId(command.bootstrap)
493 : null;
494 if (projectId === null) return;
495 const clone = yield* tracker.get(projectId);
496 if (clone === null || clone.phase === "done") return;
497 return yield* new OrchestrationDispatchCommandError({
498 message:
499 clone.phase === "running"
500 ? "The repository is still being cloned."
501 : "The repository was not cloned. Retry the clone first.",
502 });
503 },
504);
505
506function bootstrapProjectId(bootstrap: unknown): ProjectId | null {
507 if (typeof bootstrap !== "object" || bootstrap === null) return null;
508 const createThread = (bootstrap as { createThread?: { projectId?: ProjectId } }).createThread;
509 return createThread?.projectId ?? null;
510}
511
512function describeCloneFailure(cause: Cause.Cause<unknown>): string {
513 const error = Cause.squash(cause);
514 if (isSourceControlRepositoryError(error)) return error.detail;
515 return error instanceof Error && error.message.trim().length > 0
516 ? error.message
517 : "The repository could not be cloned.";
518}
519
520export const layer = Layer.effect(ProjectCloneTracker, make);