1import * as FileSystem from "effect/FileSystem";
2import {
3 ChatAttachmentId,
4 getProviderAttachmentLimitError,
5 type ChatAttachment,
6} from "@t3tools/contracts";
7import * as Effect from "effect/Effect";
8import * as Schema from "effect/Schema";
9
10import {
11 parseThreadSegmentFromAttachmentId,
12 PENDING_ATTACHMENT_THREAD_SEGMENT,
13 planAttachmentClaim,
14 resolveAttachmentPath,
15} from "../attachmentStore.ts";
16import * as ServerConfig from "../config.ts";
17
18export class AttachmentClaimError extends Schema.TaggedError<AttachmentClaimError>()(
19 "AttachmentClaimError",
20 {
21 message: Schema.String,
22 cause: Schema.optional(Schema.Defect()),
23 },
24) {}
25
26export const validateAttachmentLimits = Effect.fn("AttachmentClaims.validateAttachmentLimits")(
27 function* (attachments: ReadonlyArray<ChatAttachment>) {
28 const error = getProviderAttachmentLimitError(attachments);
29 if (error) return yield* new AttachmentClaimError({ message: error });
30 },
31);
32
33export interface ClaimedAttachments {
34 readonly attachments: ReadonlyArray<ChatAttachment>;
35 readonly claimedPaths: ReadonlyArray<string>;
36}
37
38export function attachmentIsPendingUpload(attachment: ChatAttachment): boolean {
39 return parseThreadSegmentFromAttachmentId(attachment.id) === PENDING_ATTACHMENT_THREAD_SEGMENT;
40}
41
42/** Remove partial claims only before dispatch, or after proving they were not accepted. */
43export const releaseClaimedAttachments = Effect.fn("AttachmentClaims.releaseClaimedAttachments")(
44 function* (claimedPaths: ReadonlyArray<string>) {
45 if (claimedPaths.length === 0) return;
46 const fileSystem = yield* FileSystem.FileSystem;
47 yield* Effect.forEach(claimedPaths, (path) => fileSystem.remove(path).pipe(Effect.ignore), {
48 concurrency: 1,
49 discard: true,
50 }).pipe(Effect.uninterruptible);
51 },
52);
53
54/**
55 * Claims pending uploads into the target thread's attachment store before the
56 * command enters the orchestrator: verifies the staged file, copies it under a
57 * thread-scoped id, and rewrites the attachment ref. A copy, not a move — the
58 * pending file stays behind as the retry source for a failed bootstrap, and
59 * the periodic pending sweep reclaims it later. Already-claimed attachments
60 * pass through untouched.
61 */
62export const claimPendingAttachments = Effect.fn("AttachmentClaims.claimPendingAttachments")(
63 function* (input: {
64 readonly threadId: string;
65 readonly attachments: ReadonlyArray<ChatAttachment>;
66 }) {
67 yield* validateAttachmentLimits(input.attachments);
68 if (
69 new Set(input.attachments.map((attachment) => attachment.id)).size !==
70 input.attachments.length
71 ) {
72 return yield* new AttachmentClaimError({
73 message: "Duplicate attachment ids are not allowed.",
74 });
75 }
76 if (!input.attachments.some(attachmentIsPendingUpload)) {
77 return { attachments: input.attachments, claimedPaths: [] } satisfies ClaimedAttachments;
78 }
79 const serverConfig = yield* ServerConfig.ServerConfig;
80 const fileSystem = yield* FileSystem.FileSystem;
81 const claimedPaths: string[] = [];
82 const attachments = yield* Effect.forEach(
83 input.attachments,
84 (attachment) =>
85 Effect.gen(function* () {
86 if (!attachmentIsPendingUpload(attachment)) {
87 return attachment;
88 }
89 const claim = planAttachmentClaim({
90 attachmentsDir: serverConfig.attachmentsDir,
91 threadId: input.threadId,
92 attachmentId: attachment.id,
93 });
94 if (!claim.ok) {
95 return yield* new AttachmentClaimError({
96 message: `Attachment '${attachment.name}' cannot be sent: ${claim.reason}.`,
97 });
98 }
99 const info = yield* fileSystem.stat(claim.currentPath).pipe(
100 Effect.mapError(
101 (cause) =>
102 new AttachmentClaimError({
103 message: `Attachment '${attachment.name}' cannot be sent: attachment not found.`,
104 cause,
105 }),
106 ),
107 );
108 if (Number(info.size) !== attachment.sizeBytes) {
109 return yield* new AttachmentClaimError({
110 message: `Attachment '${attachment.name}' cannot be sent: stored size does not match.`,
111 });
112 }
113 const normalized: ChatAttachment = {
114 ...attachment,
115 id: ChatAttachmentId.make(claim.finalId),
116 mimeType: attachment.mimeType.toLowerCase(),
117 };
118 const expectedPath = resolveAttachmentPath({
119 attachmentsDir: serverConfig.attachmentsDir,
120 attachment: normalized,
121 });
122 if (expectedPath !== claim.finalPath) {
123 return yield* new AttachmentClaimError({
124 message: `Attachment '${attachment.name}' cannot be sent: attachment type does not match the upload.`,
125 });
126 }
127 // A copy, not a hard link: an agent editing the delivered file in
128 // place must not mutate the retry source. fs.copyFile cannot be
129 // cancelled, so the copy and its rollback registration stay in one
130 // uninterruptible region: an interrupt landing mid-copy still waits
131 // for the write to settle and records the path before cleanup runs.
132 yield* fileSystem.copyFile(claim.currentPath, claim.finalPath).pipe(
133 Effect.mapError(
134 (cause) =>
135 new AttachmentClaimError({
136 message: `Failed to claim attachment '${attachment.name}' for this thread.`,
137 cause,
138 }),
139 ),
140 Effect.andThen(Effect.sync(() => claimedPaths.push(claim.finalPath))),
141 Effect.uninterruptible,
142 );
143 return normalized;
144 }),
145 { concurrency: 1 },
146 ).pipe(Effect.onError(() => releaseClaimedAttachments(claimedPaths)));
147 return { attachments, claimedPaths } satisfies ClaimedAttachments;
148 },
149);