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";
9import { isSvgAttachment } from "@t3tools/shared/image";
10
11import {
12 parseThreadSegmentFromAttachmentId,
13 PENDING_ATTACHMENT_THREAD_SEGMENT,
14 planAttachmentClaim,
15 resolveAttachmentPath,
16} from "../attachmentStore.ts";
17import * as ServerConfig from "../config.ts";
18import * as DocumentText from "../provider/DocumentText.ts";
19
20export class AttachmentClaimError extends Schema.TaggedError<AttachmentClaimError>()(
21 "AttachmentClaimError",
22 {
23 message: Schema.String,
24 cause: Schema.optional(Schema.Defect()),
25 },
26) {}
27
28export const validateAttachmentLimits = Effect.fn("AttachmentClaims.validateAttachmentLimits")(
29 function* (attachments: ReadonlyArray<ChatAttachment>) {
30 const error = getProviderAttachmentLimitError(attachments);
31 if (error) return yield* new AttachmentClaimError({ message: error });
32 },
33);
34
35export interface ClaimedAttachments {
36 readonly attachments: ReadonlyArray<ChatAttachment>;
37 readonly claimedPaths: ReadonlyArray<string>;
38}
39
40export function attachmentIsPendingUpload(attachment: ChatAttachment): boolean {
41 return parseThreadSegmentFromAttachmentId(attachment.id) === PENDING_ATTACHMENT_THREAD_SEGMENT;
42}
43
44/** Remove partial claims only before dispatch, or after proving they were not accepted. */
45export const releaseClaimedAttachments = Effect.fn("AttachmentClaims.releaseClaimedAttachments")(
46 function* (claimedPaths: ReadonlyArray<string>) {
47 if (claimedPaths.length === 0) return;
48 const fileSystem = yield* FileSystem.FileSystem;
49 yield* Effect.forEach(
50 claimedPaths,
51 (path) =>
52 fileSystem
53 .remove(path)
54 .pipe(
55 Effect.ignore,
56 Effect.andThen(fileSystem.remove(`${path}.txt`, { force: true }).pipe(Effect.ignore)),
57 ),
58 {
59 concurrency: 1,
60 discard: true,
61 },
62 ).pipe(Effect.uninterruptible);
63 },
64);
65
66/** Copies preserve the pending retry source after failed bootstrap; the pending sweep reclaims it. */
67export const claimPendingAttachments = Effect.fn("AttachmentClaims.claimPendingAttachments")(
68 function* (input: {
69 readonly threadId: string;
70 readonly attachments: ReadonlyArray<ChatAttachment>;
71 }) {
72 yield* validateAttachmentLimits(input.attachments);
73 if (
74 new Set(input.attachments.map((attachment) => attachment.id)).size !==
75 input.attachments.length
76 ) {
77 return yield* new AttachmentClaimError({
78 message: "Duplicate attachment ids are not allowed.",
79 });
80 }
81 if (!input.attachments.some(attachmentIsPendingUpload)) {
82 return { attachments: input.attachments, claimedPaths: [] } satisfies ClaimedAttachments;
83 }
84 const serverConfig = yield* ServerConfig.ServerConfig;
85 const fileSystem = yield* FileSystem.FileSystem;
86 const claimedPaths: string[] = [];
87 const documents = yield* DocumentText.DocumentText;
88 const attachments = yield* Effect.forEach(
89 input.attachments,
90 (attachment) =>
91 Effect.gen(function* () {
92 if (!attachmentIsPendingUpload(attachment)) {
93 return attachment;
94 }
95 const claim = planAttachmentClaim({
96 attachmentsDir: serverConfig.attachmentsDir,
97 threadId: input.threadId,
98 attachmentId: attachment.id,
99 });
100 if (!claim.ok) {
101 return yield* new AttachmentClaimError({
102 message: `Attachment '${attachment.name}' cannot be sent: ${claim.reason}.`,
103 });
104 }
105 const info = yield* fileSystem.stat(claim.currentPath).pipe(
106 Effect.mapError(
107 (cause) =>
108 new AttachmentClaimError({
109 message: `Attachment '${attachment.name}' cannot be sent: attachment not found.`,
110 cause,
111 }),
112 ),
113 );
114 if (Number(info.size) !== attachment.sizeBytes) {
115 return yield* new AttachmentClaimError({
116 message: `Attachment '${attachment.name}' cannot be sent: stored size does not match.`,
117 });
118 }
119 const normalized: ChatAttachment = {
120 ...attachment,
121 id: ChatAttachmentId.make(claim.finalId),
122 name:
123 attachment.type === "file" &&
124 isSvgAttachment(attachment) &&
125 !/\.svg$/i.test(attachment.name)
126 ? `${attachment.name}.svg`
127 : attachment.name,
128 mimeType: attachment.mimeType.toLowerCase(),
129 };
130 const expectedPath = resolveAttachmentPath({
131 attachmentsDir: serverConfig.attachmentsDir,
132 attachment: normalized,
133 });
134 if (expectedPath !== claim.finalPath) {
135 return yield* new AttachmentClaimError({
136 message: `Attachment '${attachment.name}' cannot be sent: attachment type does not match the upload.`,
137 });
138 }
139 // Copies isolate edits from the retry source. Keep the uncancellable copy and rollback registration
140 // uninterruptible so cleanup always sees completed writes.
141 yield* fileSystem.copyFile(claim.currentPath, claim.finalPath).pipe(
142 Effect.mapError(
143 (cause) =>
144 new AttachmentClaimError({
145 message: `Failed to claim attachment '${attachment.name}' for this thread.`,
146 cause,
147 }),
148 ),
149 Effect.andThen(Effect.sync(() => claimedPaths.push(claim.finalPath))),
150 Effect.uninterruptible,
151 );
152 const sidecar = `${claim.currentPath}.txt`;
153 if (yield* fileSystem.exists(sidecar).pipe(Effect.orElseSucceed(() => false))) {
154 yield* fileSystem.copyFile(sidecar, `${claim.finalPath}.txt`).pipe(
155 Effect.andThen(Effect.sync(() => claimedPaths.push(`${claim.finalPath}.txt`))),
156 Effect.onError(() =>
157 fileSystem.remove(`${claim.finalPath}.txt`, { force: true }).pipe(Effect.ignore),
158 ),
159 Effect.uninterruptible,
160 Effect.ignore,
161 );
162 } else {
163 const extracted = yield* documents.ensure(claim.finalPath);
164 if (extracted) claimedPaths.push(extracted);
165 }
166 return normalized;
167 }),
168 { concurrency: 1 },
169 ).pipe(Effect.onError(() => releaseClaimedAttachments(claimedPaths)));
170 return { attachments, claimedPaths } satisfies ClaimedAttachments;
171 },
172);