1import { SourceControlProviderKind } from "@t3tools/contracts";
2import type {
3 ProjectCloneSnapshot,
4 ProjectCloneStage,
5 ProjectCloneStartInput,
6 ProjectCloneStartResult,
7 ProjectId,
8 SourceControlRepositoryInfo,
9} from "@t3tools/contracts";
10import {
11 OrchestrationDispatchCommandError,
12 PROJECT_CLONE_DETAIL_MAX_LENGTH,
13 PROJECT_CLONE_ERROR_MAX_LENGTH,
14 SourceControlRepositoryError,
15} from "@t3tools/contracts";
16import * as Cause from "effect/Cause";
17import * as Context from "effect/Context";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Exit from "effect/Exit";
21import * as Fiber from "effect/Fiber";
22import * as Layer from "effect/Layer";
23import * as PubSub from "effect/PubSub";
24import * as Queue from "effect/Queue";
25import * as Ref from "effect/Ref";
26import * as Schema from "effect/Schema";
27import * as Semaphore from "effect/Semaphore";
28import * as Scope from "effect/Scope";
29import * as Stream from "effect/Stream";
30
31import * as SourceControlRepositoryService from "../sourceControl/SourceControlRepositoryService.ts";
32
33/**
34 * Runs repository clones that back newly added projects and tracks their
35 * progress so clients can show it anywhere, not just in the surface that
36 * started the clone.
37 *
38 * The clone is detached from the request that started it: the palette closes
39 * immediately, the project already exists (pointing at the empty destination),
40 * and the composer can hold a draft for it. Snapshots are memory only. A done
41 * clone is dropped after a short grace window; a failed one stays until it is
42 * retried or the server restarts, since the empty project is the durable
43 * record the user can act on.
44 */
45export class ProjectCloneTracker extends Context.Service<
46 ProjectCloneTracker,
47 {
48 /**
49 * Resolves the remote, creates the project, and starts the clone in the
50 * background. Fails before creating anything when the destination or
51 * repository is unusable so the caller can report it inline.
52 */
53 readonly start: (
54 input: ProjectCloneStartInput,
55 hooks: ProjectCloneHooks,
56 ) => Effect.Effect<
57 ProjectCloneStartResult,
58 SourceControlRepositoryError | OrchestrationDispatchCommandError
59 >;
60 /** Interrupts a running clone and deletes the partial checkout. */
61 readonly cancel: (projectId: ProjectId) => Effect.Effect<boolean>;
62 /** Restarts a failed or cancelled clone into the same destination. */
63 readonly retry: (projectId: ProjectId) => Effect.Effect<boolean, SourceControlRepositoryError>;
64 /**
65 * Stops tracking a project's clone, interrupting it if it still runs and
66 * removing an unfinished checkout. Called when the project is deleted.
67 */
68 readonly discard: (projectId: ProjectId) => Effect.Effect<void>;
69 readonly get: (projectId: ProjectId) => Effect.Effect<ProjectCloneSnapshot | null>;
70 /** Emits every tracked clone first, then the full list after each change. */
71 readonly stream: Stream.Stream<ReadonlyArray<ProjectCloneSnapshot>>;
72 }
73>()("t3/project/ProjectCloneTracker") {}
74
75/**
76 * Orchestration side effects the caller owns. The tracker never depends on the
77 * engine directly: the WebSocket handler dispatches with the client's origin,
78 * and the tracker only needs the outcome.
79 */
80export interface ProjectCloneHooks {
81 readonly createProject: (input: {
82 readonly projectId: ProjectId;
83 readonly title: string;
84 readonly workspaceRoot: string;
85 readonly createdAt: string;
86 }) => Effect.Effect<void, OrchestrationDispatchCommandError>;
87 /** Runs after a successful clone so cached repository identity and git status refresh. */
88 readonly onCloned: (input: {
89 readonly projectId: ProjectId;
90 readonly workspaceRoot: string;
91 }) => Effect.Effect<void>;
92}
93
94/** Finished snapshots stay visible this long so a late subscriber sees the outcome. */
95const DONE_RETENTION = "30 seconds";
96
97function clampText(text: string, maxLength: number): string {
98 return text.length <= maxLength ? text : `${text.slice(0, maxLength - 1)}…`;
99}
100
101const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
102
103interface TrackedClone {
104 readonly snapshot: ProjectCloneSnapshot;
105 readonly fiber: Fiber.Fiber<unknown, unknown> | null;
106 readonly hooks: ProjectCloneHooks;
107 readonly input: {
108 /** What git is given; may carry credentials and never leaves the server. */
109 readonly cloneUrl: string;
110 readonly destinationPath: string;
111 readonly repository: SourceControlRepositoryInfo | null;
112 };
113}
114
115/** @public Service construction is part of the canonical Effect module API. */
116export const make = Effect.gen(function* () {
117 const repositories = yield* SourceControlRepositoryService.SourceControlRepositoryService;
118 const clones = yield* Ref.make(new Map<ProjectId, TrackedClone>());
119 const changes = yield* PubSub.unbounded<ReadonlyArray<ProjectCloneSnapshot>>();
120 const retentionFibers = new Map<ProjectId, Fiber.Fiber<unknown, never>>();
121 let sequence = 0;
122 // Clone fibers outlive the RPC that started them but not the server.
123 const cloneScope = yield* Scope.make("parallel");
124 yield* Effect.addFinalizer(() => Scope.close(cloneScope, Exit.void));
125 // start/cancel/retry/discard mutate the same entry and the same directory;
126 // one at a time keeps a double-clicked Retry from racing two clones into it.
127 const actionLock = yield* Semaphore.make(1);
128 const locked = <A, E, R>(effect: Effect.Effect<A, E, R>) => actionLock.withPermits(1)(effect);
129
130 const list = Ref.get(clones).pipe(
131 Effect.map((current) => Array.from(current.values(), (tracked) => tracked.snapshot)),
132 );
133 const publish = list.pipe(Effect.flatMap((snapshots) => PubSub.publish(changes, snapshots)));
134
135 const modify = (
136 projectId: ProjectId,
137 mutate: (tracked: TrackedClone) => TrackedClone,
138 ): Effect.Effect<ProjectCloneSnapshot | null> =>
139 Ref.modify(clones, (current) => {
140 const existing = current.get(projectId);
141 if (!existing) return [null, current] as const;
142 const nextTracked = mutate(existing);
143 const nextSnapshot = { ...nextTracked.snapshot, sequence: ++sequence };
144 const next = new Map(current);
145 next.set(projectId, { ...nextTracked, snapshot: nextSnapshot });
146 return [nextSnapshot, next] as const;
147 }).pipe(Effect.tap((snapshot) => (snapshot ? publish : Effect.void)));
148
149 const clearRetention = (projectId: ProjectId) => {
150 const fiber = retentionFibers.get(projectId);
151 retentionFibers.delete(projectId);
152 return fiber ? Fiber.interrupt(fiber).pipe(Effect.ignore) : Effect.void;
153 };
154
155 const remove = (projectId: ProjectId) =>
156 Ref.update(clones, (current) => {
157 if (!current.has(projectId)) return current;
158 const next = new Map(current);
159 next.delete(projectId);
160 return next;
161 }).pipe(Effect.andThen(publish));
162
163 const scheduleRemoval = (projectId: ProjectId) =>
164 Effect.gen(function* () {
165 yield* clearRetention(projectId);
166 const fiber = yield* remove(projectId).pipe(
167 Effect.delay(DONE_RETENTION),
168 Effect.ensuring(
169 Effect.sync(() => {
170 if (retentionFibers.get(projectId) === fiber) retentionFibers.delete(projectId);
171 }),
172 ),
173 Effect.forkDetach,
174 );
175 retentionFibers.set(projectId, fiber);
176 });
177
178 const progress = (
179 projectId: ProjectId,
180 update: {
181 readonly stage: ProjectCloneStage;
182 readonly percent: number | null;
183 readonly detail: string | null;
184 },
185 ) =>
186 modify(projectId, (tracked) => ({
187 ...tracked,
188 snapshot: {
189 ...tracked.snapshot,
190 stage: update.stage,
191 percent: update.percent,
192 detail:
193 update.detail === null ? null : clampText(update.detail, PROJECT_CLONE_DETAIL_MAX_LENGTH),
194 },
195 })).pipe(Effect.asVoid);
196
197 const finish = (
198 projectId: ProjectId,
199 phase: "done" | "failed" | "cancelled",
200 error: string | null,
201 ) =>
202 Effect.gen(function* () {
203 const endedAt = yield* nowIso;
204 yield* modify(projectId, (tracked) => ({
205 ...tracked,
206 fiber: null,
207 snapshot: {
208 ...tracked.snapshot,
209 phase,
210 endedAt,
211 percent: phase === "done" ? 100 : tracked.snapshot.percent,
212 error: error === null ? null : clampText(error, PROJECT_CLONE_ERROR_MAX_LENGTH),
213 },
214 }));
215 if (phase === "done") yield* scheduleRemoval(projectId);
216 });
217
218 /**
219 * The clone body. Runs in its own fiber; the tracker records the outcome
220 * through `onExit`, which still runs when the fiber is interrupted (a
221 * `matchCause` handler would be skipped). Once git has finished the clone
222 * is marked done before the post-clone hook runs, so a late Cancel cannot
223 * tear down a complete checkout.
224 */
225 const runClone = (projectId: ProjectId, tracked: TrackedClone) =>
226 repositories
227 .cloneRepository(
228 { remoteUrl: tracked.input.cloneUrl, destinationPath: tracked.input.destinationPath },
229 { onProgress: (update) => progress(projectId, update), timeoutMs: null },
230 )
231 .pipe(
232 Effect.onExit((exit) =>
233 Exit.isSuccess(exit)
234 ? finish(projectId, "done", null)
235 : Cause.hasInterruptsOnly(exit.cause)
236 ? finish(projectId, "cancelled", null)
237 : finish(projectId, "failed", describeCloneFailure(exit.cause)),
238 ),
239 Effect.flatMap(() =>
240 tracked.hooks
241 .onCloned({ projectId, workspaceRoot: tracked.input.destinationPath })
242 .pipe(Effect.ignoreCause({ log: true })),
243 ),
244 Effect.ignoreCause(),
245 );
246
247 const launch = (projectId: ProjectId) =>
248 Effect.gen(function* () {
249 const current = yield* Ref.get(clones);
250 const tracked = current.get(projectId);
251 if (!tracked) return;
252 const fiber = yield* runClone(projectId, tracked).pipe(Effect.forkIn(cloneScope));
253 yield* Ref.update(clones, (map) => {
254 const existing = map.get(projectId);
255 if (!existing) return map;
256 const next = new Map(map);
257 next.set(projectId, { ...existing, fiber });
258 return next;
259 });
260 });
261
262 const start: ProjectCloneTracker["Service"]["start"] = Effect.fn("ProjectCloneTracker.start")(
263 function* (input, hooks) {
264 const prepared = yield* repositories.prepareClone(input);
265 const startedAt = yield* nowIso;
266 const snapshot: ProjectCloneSnapshot = {
267 projectId: input.projectId,
268 remoteUrl: prepared.remoteUrl,
269 destinationPath: prepared.destinationPath,
270 repository: prepared.repository,
271 phase: "running",
272 stage: "connecting",
273 percent: null,
274 detail: null,
275 error: null,
276 startedAt,
277 endedAt: null,
278 sequence: ++sequence,
279 };
280 const claimed = yield* Ref.modify(clones, (current) => {
281 // A second start for the same project, or for a destination another
282 // clone already owns, must not race two gits into one directory.
283 const conflict = Array.from(current.values()).some(
284 (tracked) =>
285 tracked.snapshot.projectId === input.projectId ||
286 tracked.input.destinationPath === prepared.destinationPath,
287 );
288 if (conflict) return [false, current] as const;
289 const next = new Map(current);
290 next.set(input.projectId, {
291 snapshot,
292 fiber: null,
293 hooks,
294 input: {
295 cloneUrl: prepared.cloneUrl,
296 destinationPath: prepared.destinationPath,
297 repository: prepared.repository,
298 },
299 });
300 return [true, next] as const;
301 });
302 if (!claimed) {
303 return yield* new SourceControlRepositoryError({
304 operation: "cloneRepository",
305 provider: input.provider ?? SourceControlProviderKind.make("unknown"),
306 detail: "A clone into this destination is already in progress.",
307 });
308 }
309 // Everything after the claim runs to completion even if the requesting
310 // connection drops: a claimed entry with no fiber could neither be
311 // cancelled nor retried. Any failure in here releases the claim.
312 yield* Effect.uninterruptible(
313 Effect.gen(function* () {
314 yield* clearRetention(input.projectId);
315 // The entry is registered before the project exists so a
316 // thread.create or project.delete racing this call already sees it.
317 yield* hooks.createProject({
318 projectId: input.projectId,
319 title: input.title,
320 workspaceRoot: prepared.destinationPath,
321 createdAt: input.createdAt,
322 });
323 yield* publish;
324 yield* launch(input.projectId);
325 }).pipe(Effect.tapError(() => remove(input.projectId))),
326 );
327 return {
328 projectId: input.projectId,
329 cwd: prepared.destinationPath,
330 remoteUrl: prepared.remoteUrl,
331 repository: prepared.repository,
332 };
333 },
334 );
335
336 const get: ProjectCloneTracker["Service"]["get"] = (projectId) =>
337 Ref.get(clones).pipe(Effect.map((current) => current.get(projectId)?.snapshot ?? null));
338
339 const cancel: ProjectCloneTracker["Service"]["cancel"] = (projectId) =>
340 Effect.gen(function* () {
341 const current = yield* Ref.get(clones);
342 const tracked = current.get(projectId);
343 if (!tracked || tracked.snapshot.phase !== "running" || !tracked.fiber) return false;
344 // Uninterruptible past this point: a client that disconnects mid-cancel
345 // must not leave a dead fiber behind a "running" snapshot.
346 yield* Effect.uninterruptible(
347 Effect.gen(function* () {
348 yield* Fiber.interrupt(tracked.fiber!);
349 const after = yield* get(projectId);
350 // Git finished in the window before the interrupt landed: keep it.
351 if (after?.phase === "done") return;
352 if (after?.phase === "running") yield* finish(projectId, "cancelled", null);
353 // The partial checkout goes so a retry starts from an empty destination.
354 yield* repositories.discardClone(tracked.input.destinationPath).pipe(Effect.ignore);
355 }),
356 );
357 return true;
358 });
359
360 const retry: ProjectCloneTracker["Service"]["retry"] = (projectId) =>
361 Effect.gen(function* () {
362 const current = yield* Ref.get(clones);
363 const tracked = current.get(projectId);
364 if (
365 !tracked ||
366 (tracked.snapshot.phase !== "failed" && tracked.snapshot.phase !== "cancelled")
367 ) {
368 return false;
369 }
370 // A retry into leftover files would fail on the non-empty destination
371 // with a less useful message, so a cleanup failure is the error here.
372 yield* repositories.discardClone(tracked.input.destinationPath);
373 const startedAt = yield* nowIso;
374 yield* modify(projectId, (entry) => ({
375 ...entry,
376 snapshot: {
377 ...entry.snapshot,
378 phase: "running",
379 stage: "connecting",
380 percent: null,
381 detail: null,
382 error: null,
383 startedAt,
384 endedAt: null,
385 },
386 }));
387 yield* launch(projectId);
388 return true;
389 });
390
391 const discard: ProjectCloneTracker["Service"]["discard"] = (projectId) =>
392 Effect.gen(function* () {
393 const current = yield* Ref.get(clones);
394 const tracked = current.get(projectId);
395 if (!tracked) return;
396 if (tracked.fiber) yield* Fiber.interrupt(tracked.fiber);
397 // Git may have finished while the interrupt was landing; the project
398 // is going away either way, but a complete checkout is the user's.
399 const after = yield* get(projectId);
400 if (after?.phase !== "done") {
401 yield* repositories.discardClone(tracked.input.destinationPath).pipe(Effect.ignore);
402 }
403 yield* clearRetention(projectId);
404 yield* remove(projectId);
405 });
406
407 // One-slot sliding mailbox per subscriber: a slow socket only ever holds the
408 // newest list, and lists are whole states so skipping intermediates is safe.
409 const stream: ProjectCloneTracker["Service"]["stream"] = Stream.callback<
410 ReadonlyArray<ProjectCloneSnapshot>
411 >(
412 (mailbox) =>
413 Effect.gen(function* () {
414 const subscription = yield* PubSub.subscribe(changes);
415 Queue.offerUnsafe(mailbox, yield* list);
416 yield* Stream.fromSubscription(subscription).pipe(
417 Stream.runForEach((snapshots) =>
418 Effect.sync(() => Queue.offerUnsafe(mailbox, snapshots)),
419 ),
420 Effect.forkScoped,
421 );
422 }),
423 { bufferSize: 1, strategy: "sliding" },
424 );
425
426 return ProjectCloneTracker.of({
427 start: (input, hooks) => locked(start(input, hooks)),
428 cancel: (projectId) => locked(cancel(projectId)),
429 retry: (projectId) => locked(retry(projectId)),
430 discard: (projectId) => locked(discard(projectId)),
431 get,
432 stream,
433 });
434});
435
436const isSourceControlRepositoryError = Schema.is(SourceControlRepositoryError);
437
438/**
439 * A project whose clone has not landed has no files to work in. Every
440 * dispatch transport (WebSocket, HTTP) runs this before normalizing so no
441 * client can start a thread on an empty tree, and no attachment copies are
442 * made for a command that is about to be refused.
443 */
444export const rejectCommandsDuringClone = (
445 tracker: ProjectCloneTracker["Service"],
446 command: { readonly type: string; readonly projectId?: ProjectId; readonly bootstrap?: unknown },
447): Effect.Effect<void, OrchestrationDispatchCommandError> =>
448 Effect.gen(function* () {
449 const projectId =
450 command.type === "thread.create"
451 ? (command.projectId ?? null)
452 : command.type === "thread.turn.start"
453 ? bootstrapProjectId(command.bootstrap)
454 : null;
455 if (projectId === null) return;
456 const clone = yield* tracker.get(projectId);
457 if (clone === null || clone.phase === "done") return;
458 return yield* new OrchestrationDispatchCommandError({
459 message:
460 clone.phase === "running"
461 ? "The repository is still being cloned."
462 : "The repository was not cloned. Retry the clone first.",
463 });
464 });
465
466function bootstrapProjectId(bootstrap: unknown): ProjectId | null {
467 if (typeof bootstrap !== "object" || bootstrap === null) return null;
468 const createThread = (bootstrap as { createThread?: { projectId?: ProjectId } }).createThread;
469 return createThread?.projectId ?? null;
470}
471
472function describeCloneFailure(cause: Cause.Cause<unknown>): string {
473 const error = Cause.squash(cause);
474 if (isSourceControlRepositoryError(error)) return error.detail;
475 return error instanceof Error && error.message.trim().length > 0
476 ? error.message
477 : "The repository could not be cloned.";
478}
479
480export const layer = Layer.effect(ProjectCloneTracker, make);