1import {
2 CommandId,
3 type OrchestrationProjectShell,
4 ProjectId,
5 type Project,
6 type ProjectCreatePayload,
7 type ProjectUpdatePayload,
8 type ProjectSnapshot,
9 type ThreadId,
10} from "@t3tools/contracts";
11import * as KeyedLock from "@t3tools/shared/KeyedLock";
12import * as Context from "effect/Context";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Result from "effect/Result";
18import * as Schema from "effect/Schema";
19
20import * as EventSink from "../orchestration-v2/EventSink.ts";
21import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
22import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts";
23import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
24import {
25 decodeProjectCommandRejection,
26 encodeProjectCommandRejection,
27 planProjectCommand,
28 type ProjectCommand,
29} from "../orchestration-v2/ProjectCommands.ts";
30import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
31import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts";
32import { planThreadDeletion } from "../orchestration-v2/ThreadDeletion.ts";
33import * as ServerSettings from "../serverSettings.ts";
34import * as ProjectEnrichmentService from "./ProjectEnrichmentService.ts";
35import * as WorkspacePaths from "../workspace/WorkspacePaths.ts";
36
37export interface ProjectCreateInput extends ProjectCreatePayload {
38 readonly commandId: CommandId;
39 readonly projectId: ProjectId;
40}
41
42export interface ProjectUpdateInput extends ProjectUpdatePayload {
43 readonly commandId: CommandId;
44 readonly projectId: ProjectId;
45}
46
47export interface ProjectBootstrapInput extends ProjectCreateInput {}
48
49export interface ProjectDeleteInput {
50 readonly commandId: CommandId;
51 readonly projectId: ProjectId;
52 readonly force?: boolean;
53}
54
55export class ProjectNotFoundError extends Schema.TaggedError<ProjectNotFoundError>()(
56 "ProjectNotFoundError",
57 { projectId: ProjectId },
58) {
59 override get message(): string {
60 return `Project ${this.projectId} was not found.`;
61 }
62}
63
64export class ProjectConflictError extends Schema.TaggedError<ProjectConflictError>()(
65 "ProjectConflictError",
66 {
67 projectId: ProjectId,
68 workspaceRoot: Schema.String,
69 conflictingProjectId: ProjectId,
70 },
71) {
72 override get message(): string {
73 return `Workspace ${this.workspaceRoot} already belongs to project ${this.conflictingProjectId}.`;
74 }
75}
76
77export class ProjectNotEmptyError extends Schema.TaggedError<ProjectNotEmptyError>()(
78 "ProjectNotEmptyError",
79 { projectId: ProjectId },
80) {
81 override get message(): string {
82 return `Project ${this.projectId} is not empty.`;
83 }
84}
85
86export class ProjectOperationError extends Schema.TaggedError<ProjectOperationError>()(
87 "ProjectOperationError",
88 {
89 operation: Schema.Literals([
90 "normalize-workspace",
91 "read-project",
92 "list-projects",
93 "list-threads",
94 "delete-thread",
95 "dispatch-project-command",
96 ]),
97 projectId: Schema.optional(ProjectId),
98 workspaceRoot: Schema.optional(Schema.String),
99 cause: Schema.Defect(),
100 },
101) {
102 override get message(): string {
103 return `Project operation '${this.operation}' failed${this.projectId === undefined ? "" : ` for ${this.projectId}`}.`;
104 }
105}
106
107export type ProjectServiceError =
108 | ProjectNotFoundError
109 | ProjectConflictError
110 | ProjectNotEmptyError
111 | ProjectOperationError;
112
113export class ProjectService extends Context.Service<
114 ProjectService,
115 {
116 readonly create: (input: ProjectCreateInput) => Effect.Effect<Project, ProjectServiceError>;
117 readonly bootstrap: (
118 input: ProjectBootstrapInput,
119 ) => Effect.Effect<
120 { readonly project: Project; readonly created: boolean },
121 ProjectServiceError
122 >;
123 readonly update: (input: ProjectUpdateInput) => Effect.Effect<Project, ProjectServiceError>;
124 readonly delete: (input: ProjectDeleteInput) => Effect.Effect<Project, ProjectServiceError>;
125 readonly getById: (
126 projectId: ProjectId,
127 options?: { readonly includeDeleted?: boolean },
128 ) => Effect.Effect<Option.Option<Project>, ProjectOperationError>;
129 readonly getByWorkspaceRoot: (
130 workspaceRoot: string,
131 options?: { readonly includeDeleted?: boolean },
132 ) => Effect.Effect<Option.Option<Project>, ProjectOperationError>;
133 readonly snapshot: Effect.Effect<ProjectSnapshot, ProjectOperationError>;
134 /**
135 * An active project's shell with its immediately available repository
136 * identity; missing identity resolves in the background.
137 */
138 readonly getShell: (
139 projectId: ProjectId,
140 ) => Effect.Effect<Option.Option<OrchestrationProjectShell>, ProjectOperationError>;
141 /** Active project shells, enriched like `getShell`, in creation order. */
142 readonly listShells: (options?: {
143 readonly projectIds?: ReadonlyArray<ProjectId>;
144 }) => Effect.Effect<ReadonlyArray<OrchestrationProjectShell>, ProjectOperationError>;
145 }
146>()("t3/project/ProjectService") {}
147
148export const make = Effect.gen(function* () {
149 const projects = yield* ProjectStore.ProjectStoreV2;
150 const projectEnrichment = yield* ProjectEnrichmentService.ProjectEnrichmentService;
151 const workspacePaths = yield* WorkspacePaths.WorkspacePaths;
152 const threadProjections = yield* ProjectionStore.ProjectionStoreV2;
153 const eventSink = yield* EventSink.EventSinkV2;
154 const idAllocator = yield* IdAllocator.IdAllocatorV2;
155 const legacyImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter;
156 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
157 const serverSettings = yield* ServerSettings.ServerSettingsService;
158 // Commands for one project run in order. Commands that claim a workspace root
159 // also hold that root, so two projects cannot both claim it.
160 const projectLocks = yield* KeyedLock.make<ProjectId>();
161 const workspaceLocks = yield* KeyedLock.make<string>();
162
163 const toProject = (
164 row: ProjectStore.ProjectRow,
165 enrichment: ProjectEnrichmentService.ProjectEnrichment | null,
166 ): Project => ({
167 id: row.projectId,
168 title: row.title,
169 workspaceRoot: row.workspaceRoot,
170 repositoryIdentity: enrichment?.repositoryIdentity ?? null,
171 faviconPath: row.faviconPath ?? enrichment?.faviconPath ?? null,
172 defaultModelSelection: row.defaultModelSelection,
173 defaultThreadEnvMode: row.defaultThreadEnvMode,
174 autoPull: row.autoPull,
175 projectIcon: row.projectIcon,
176 scripts: row.scripts,
177 createdAt: row.createdAt,
178 updatedAt: row.updatedAt,
179 deletedAt: row.deletedAt,
180 });
181
182 const hydrate = Effect.fn("ProjectService.hydrate")(function* (row: ProjectStore.ProjectRow) {
183 const enrichment =
184 row.deletedAt === null
185 ? yield* projectEnrichment.getAvailable(row.workspaceRoot)
186 : yield* projectEnrichment.peek(row.workspaceRoot);
187 return toProject(row, enrichment);
188 });
189
190 const readRow = (projectId: ProjectId, options?: { readonly includeDeleted?: boolean }) =>
191 projects
192 .get(projectId, options)
193 .pipe(
194 Effect.mapError(
195 (cause) => new ProjectOperationError({ operation: "read-project", projectId, cause }),
196 ),
197 );
198
199 const normalizeWorkspaceRoot = (input: {
200 readonly projectId?: ProjectId;
201 readonly workspaceRoot: string;
202 readonly createIfMissing?: boolean;
203 }) =>
204 workspacePaths
205 .normalizeWorkspaceRoot(input.workspaceRoot, {
206 createIfMissing: input.createIfMissing ?? false,
207 })
208 .pipe(
209 Effect.mapError(
210 (cause) =>
211 new ProjectOperationError({
212 operation: "normalize-workspace",
213 ...(input.projectId === undefined ? {} : { projectId: input.projectId }),
214 workspaceRoot: input.workspaceRoot,
215 cause,
216 }),
217 ),
218 );
219
220 /**
221 * Plan one command against rows read under its locks, then commit its event or
222 * its rejection. A reused command id resolves to the receipt it already has.
223 */
224 const commit = Effect.fn("ProjectService.commit")(function* (command: ProjectCommand) {
225 const { projectId } = command;
226 const dispatchError = (cause: unknown) =>
227 new ProjectOperationError({ operation: "dispatch-project-command", projectId, cause });
228 const workspaceRoot = command.type === "project.delete" ? undefined : command.workspaceRoot;
229 const planAndCommit = Effect.gen(function* () {
230 const project = Option.getOrUndefined(yield* readRow(projectId, { includeDeleted: true }));
231 const workspaceOwner =
232 workspaceRoot === undefined
233 ? undefined
234 : Option.getOrUndefined(
235 yield* projects
236 .findActiveByWorkspaceRoot(workspaceRoot)
237 .pipe(Effect.mapError(dispatchError)),
238 );
239 const now = yield* DateTime.now;
240 const eventId = yield* idAllocator.allocate
241 .event({ commandId: command.commandId })
242 .pipe(Effect.mapError(dispatchError));
243 const planned = planProjectCommand({
244 command,
245 state: { project, workspaceOwner },
246 eventId,
247 now,
248 });
249 if (Result.isSuccess(planned)) {
250 const { receipt } = yield* eventSink.commitProjectCommand({
251 commandId: command.commandId,
252 projectId,
253 commandType: command.type,
254 acceptedAt: now,
255 event: planned.success,
256 });
257 return receipt;
258 }
259 return yield* eventSink.commitRejectedProjectCommand({
260 commandId: command.commandId,
261 projectId,
262 commandType: command.type,
263 rejectedAt: now,
264 error: encodeProjectCommandRejection(planned.failure),
265 });
266 });
267 const receipt = yield* projectLocks
268 .withLock(
269 projectId,
270 workspaceRoot === undefined
271 ? planAndCommit
272 : workspaceLocks.withLock(workspaceRoot, planAndCommit),
273 )
274 .pipe(Effect.mapError(dispatchError));
275 if (receipt.projectId !== projectId || receipt.commandType !== command.type) {
276 return yield* dispatchError(
277 `Command ${command.commandId} was already used by ${receipt.commandType} for ${receipt.projectId}.`,
278 );
279 }
280 // A retried command re-plans against the state it already produced, so its
281 // first receipt, not the new plan, decides the outcome.
282 if (receipt.status === "accepted") return;
283 const rejection = Option.getOrUndefined(decodeProjectCommandRejection(receipt.error));
284 switch (rejection?._tag) {
285 case "ProjectWorkspaceConflictError":
286 return yield* new ProjectConflictError({
287 projectId,
288 workspaceRoot: rejection.workspaceRoot,
289 conflictingProjectId: rejection.conflictingProjectId,
290 });
291 case "ProjectCommandMissingProjectError":
292 return yield* new ProjectNotFoundError({ projectId });
293 default:
294 return yield* dispatchError(
295 rejection ?? receipt.error ?? "The command was previously rejected.",
296 );
297 }
298 });
299
300 const readCommitted = Effect.fn("ProjectService.readCommitted")(function* (projectId: ProjectId) {
301 const row = yield* readRow(projectId, { includeDeleted: true });
302 if (Option.isNone(row)) {
303 return yield* new ProjectOperationError({
304 operation: "read-project",
305 projectId,
306 cause: "The accepted project command did not produce a project row.",
307 });
308 }
309 return yield* hydrate(row.value);
310 });
311
312 const getById: ProjectService["Service"]["getById"] = Effect.fn("ProjectService.getById")(
313 function* (projectId, options) {
314 const row = yield* readRow(projectId, options);
315 return Option.isNone(row) ? Option.none() : Option.some(yield* hydrate(row.value));
316 },
317 );
318
319 const getByWorkspaceRoot: ProjectService["Service"]["getByWorkspaceRoot"] = Effect.fn(
320 "ProjectService.getByWorkspaceRoot",
321 )(function* (workspaceRoot, options) {
322 const normalized = yield* normalizeWorkspaceRoot({ workspaceRoot });
323 const row = yield* (
324 options?.includeDeleted === true
325 ? projects
326 .list({ includeDeleted: true })
327 .pipe(
328 Effect.map((rows) =>
329 Option.fromUndefinedOr(rows.find((row) => row.workspaceRoot === normalized)),
330 ),
331 )
332 : projects.findActiveByWorkspaceRoot(normalized)
333 ).pipe(
334 Effect.mapError((cause) => new ProjectOperationError({ operation: "list-projects", cause })),
335 );
336 return Option.isNone(row) ? Option.none() : Option.some(yield* hydrate(row.value));
337 });
338
339 const create: ProjectService["Service"]["create"] = Effect.fn("ProjectService.create")(
340 function* (input) {
341 const workspaceRoot = yield* normalizeWorkspaceRoot({
342 projectId: input.projectId,
343 workspaceRoot: input.workspaceRoot,
344 createIfMissing: input.createWorkspaceRootIfMissing ?? false,
345 });
346 yield* commit({
347 type: "project.create",
348 commandId: input.commandId,
349 projectId: input.projectId,
350 title: input.title,
351 workspaceRoot,
352 ...(input.scripts === undefined ? {} : { scripts: input.scripts }),
353 });
354 yield* projectEnrichment.invalidate([workspaceRoot]);
355 return yield* readCommitted(input.projectId);
356 },
357 );
358
359 const update: ProjectService["Service"]["update"] = Effect.fn("ProjectService.update")(
360 function* (input) {
361 const existing = yield* readRow(input.projectId);
362 if (Option.isNone(existing)) {
363 return yield* new ProjectNotFoundError({ projectId: input.projectId });
364 }
365 const previousRoot = existing.value.workspaceRoot;
366 const workspaceRoot =
367 input.workspaceRoot === undefined
368 ? previousRoot
369 : yield* normalizeWorkspaceRoot({
370 projectId: input.projectId,
371 workspaceRoot: input.workspaceRoot,
372 });
373 yield* commit({
374 type: "project.meta.update",
375 commandId: input.commandId,
376 projectId: input.projectId,
377 ...(input.title === undefined ? {} : { title: input.title }),
378 ...(workspaceRoot === previousRoot ? {} : { workspaceRoot }),
379 ...(input.defaultModelSelection === undefined
380 ? {}
381 : { defaultModelSelection: input.defaultModelSelection }),
382 ...(input.autoPull === undefined ? {} : { autoPull: input.autoPull }),
383 ...(input.projectIcon === undefined ? {} : { projectIcon: input.projectIcon }),
384 ...(input.faviconPath === undefined ? {} : { faviconPath: input.faviconPath }),
385 ...(input.defaultThreadEnvMode === undefined
386 ? {}
387 : { defaultThreadEnvMode: input.defaultThreadEnvMode }),
388 ...(input.scripts === undefined ? {} : { scripts: input.scripts }),
389 });
390 if (workspaceRoot !== previousRoot) {
391 yield* projectEnrichment.invalidate([previousRoot, workspaceRoot]);
392 }
393 return yield* readCommitted(input.projectId);
394 },
395 );
396
397 const bootstrap: ProjectService["Service"]["bootstrap"] = Effect.fn("ProjectService.bootstrap")(
398 function* (input) {
399 const existing = yield* getByWorkspaceRoot(input.workspaceRoot);
400 if (Option.isSome(existing)) return { project: existing.value, created: false };
401 return { project: yield* create(input), created: true };
402 },
403 );
404
405 /** Delete one child thread durably; a stable command id makes a retry resume the cascade. */
406 const deleteChildThread = Effect.fn("ProjectService.deleteChildThread")(function* (
407 input: ProjectDeleteInput,
408 threadId: ThreadId,
409 ) {
410 yield* legacyImporter.ensureTranscript(threadId);
411 const projection = yield* threadProjections.getThreadRecords(threadId, [
412 "runs",
413 "attempts",
414 "nodes",
415 "runtimeRequests",
416 "subagents",
417 "providerSessions",
418 ]);
419 if (projection.thread.deletedAt !== null || projection.thread.projectId !== input.projectId) {
420 return;
421 }
422 const command = {
423 type: "thread.delete" as const,
424 commandId: CommandId.make(`${input.commandId}:delete-thread:${threadId}`),
425 threadId,
426 };
427 const now = yield* DateTime.now;
428 const plan = yield* planThreadDeletion({
429 command,
430 projection,
431 attachmentIds: yield* threadProjections.getThreadAttachmentIds(threadId),
432 now,
433 idAllocator,
434 });
435 const committed = yield* eventSink.commitCommand({
436 commandId: command.commandId,
437 commandType: command.type,
438 threadId,
439 acceptedAt: now,
440 events: plan.events,
441 effects: plan.effects,
442 });
443 if (
444 committed.receipt.threadId !== command.threadId ||
445 committed.receipt.commandType !== command.type
446 ) {
447 return yield* Effect.fail("The thread deletion command ID belongs to a different command.");
448 }
449 if (committed.receipt.status === "rejected") {
450 return yield* Effect.fail(
451 committed.receipt.error ?? "Thread deletion was previously rejected.",
452 );
453 }
454 });
455
456 /** Refuse a non-empty project without force, else delete its live threads first. */
457 const deleteChildThreads = Effect.fn("ProjectService.deleteChildThreads")(function* (
458 input: ProjectDeleteInput,
459 ) {
460 const { projectId } = input;
461 // The V2 shell is the only record of which threads are live.
462 const snapshot = yield* threadProjections
463 .getShellSnapshot()
464 .pipe(
465 Effect.mapError(
466 (cause) => new ProjectOperationError({ operation: "list-threads", projectId, cause }),
467 ),
468 );
469 const projectThreads = [...snapshot.threads, ...snapshot.archivedThreads].filter(
470 (thread) => thread.projectId === projectId,
471 );
472 if (projectThreads.length > 0 && input.force !== true) {
473 return yield* new ProjectNotEmptyError({ projectId });
474 }
475 // Delete children durably before the project so a failed cascade can be retried.
476 yield* Effect.forEach(
477 projectThreads,
478 (thread) =>
479 threadCommands
480 .withLock(thread.id, deleteChildThread(input, thread.id))
481 .pipe(
482 Effect.mapError(
483 (cause) =>
484 new ProjectOperationError({ operation: "delete-thread", projectId, cause }),
485 ),
486 ),
487 { concurrency: 1, discard: true },
488 );
489 });
490
491 // A machine-disabled provider instance that a deleted project's override
492 // once turned on must not stay probed and alive forever. Nothing else
493 // removes the entry, so deletion is where it goes; best-effort because a
494 // settings write failure should not block the deletion it is cleaning up
495 // after, and a server crash before a replay never revisits this project.
496 const clearProjectSettingsOverrides = (projectId: ProjectId) =>
497 serverSettings.updateSettings({ projectSettingsOverrides: { [projectId]: null } }).pipe(
498 Effect.asVoid,
499 Effect.catchCause((cause) =>
500 Effect.logWarning(
501 `Failed to clear settings overrides for deleted project ${projectId}.`,
502 cause,
503 ),
504 ),
505 );
506
507 const deleteProject: ProjectService["Service"]["delete"] = Effect.fn("ProjectService.delete")(
508 function* (input) {
509 const { projectId } = input;
510 // A deleted row still reaches commit, so a retried command id replays its
511 // receipt and any other command id is rejected as not found.
512 const existing = yield* readRow(projectId, { includeDeleted: true });
513 if (Option.isNone(existing)) {
514 return yield* new ProjectNotFoundError({ projectId });
515 }
516
517 const isFirstDelete = existing.value.deletedAt === null;
518 if (isFirstDelete) {
519 yield* deleteChildThreads(input);
520 }
521 yield* commit({ type: "project.delete", commandId: input.commandId, projectId });
522 if (isFirstDelete) {
523 yield* clearProjectSettingsOverrides(projectId);
524 }
525 yield* projectEnrichment.invalidate([existing.value.workspaceRoot]);
526 return yield* readCommitted(projectId);
527 },
528 );
529
530 const enrichShell = (shell: OrchestrationProjectShell) =>
531 projectEnrichment.getAvailable(shell.workspaceRoot).pipe(
532 Effect.map((enrichment) => ({
533 ...shell,
534 repositoryIdentity: enrichment.repositoryIdentity,
535 })),
536 );
537
538 const getShell: ProjectService["Service"]["getShell"] = Effect.fn("ProjectService.getShell")(
539 function* (projectId) {
540 const shell = yield* projects
541 .getShell(projectId)
542 .pipe(
543 Effect.mapError(
544 (cause) => new ProjectOperationError({ operation: "read-project", projectId, cause }),
545 ),
546 );
547 return Option.isNone(shell) ? shell : Option.some(yield* enrichShell(shell.value));
548 },
549 );
550
551 const listShells: ProjectService["Service"]["listShells"] = Effect.fn(
552 "ProjectService.listShells",
553 )(function* (options) {
554 const shells = yield* projects
555 .listShells(options)
556 .pipe(
557 Effect.mapError(
558 (cause) => new ProjectOperationError({ operation: "list-projects", cause }),
559 ),
560 );
561 return yield* Effect.forEach(shells, enrichShell, { concurrency: 16 });
562 });
563
564 const snapshot = Effect.gen(function* () {
565 const rows = yield* projects
566 .list()
567 .pipe(
568 Effect.mapError(
569 (cause) => new ProjectOperationError({ operation: "list-projects", cause }),
570 ),
571 );
572 const hydrated = yield* Effect.forEach(rows, hydrate, { concurrency: 8 });
573 return {
574 projects: hydrated,
575 updatedAt: DateTime.formatIso(yield* DateTime.now),
576 } satisfies ProjectSnapshot;
577 });
578
579 return ProjectService.of({
580 create,
581 bootstrap,
582 update,
583 delete: deleteProject,
584 getById,
585 getByWorkspaceRoot,
586 snapshot,
587 getShell,
588 listShells,
589 });
590});
591
592export const layer = Layer.effect(ProjectService, make).pipe(
593 Layer.provide(ThreadCommandExecutor.layer),
594);