1import * as DocumentText from "../provider/DocumentText.ts";
2// @effect-diagnostics nodeBuiltinImport:off
3import * as NodeFS from "node:fs";
4import * as NodePath from "node:path";
5
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import { describe, expect, it } from "@effect/vitest";
8import { ChatAttachmentId, type ChatAttachment } from "@t3tools/contracts";
9import * as Deferred from "effect/Deferred";
10import * as Effect from "effect/Effect";
11import * as Exit from "effect/Exit";
12import * as Fiber from "effect/Fiber";
13import * as FileSystem from "effect/FileSystem";
14import * as Layer from "effect/Layer";
15
16import {
17 createPendingAttachmentId,
18 parseThreadSegmentFromAttachmentId,
19} from "../attachmentStore.ts";
20import * as ServerConfig from "../config.ts";
21import {
22 attachmentIsPendingUpload,
23 claimPendingAttachments,
24 releaseClaimedAttachments,
25} from "./AttachmentClaims.ts";
26
27const layerTest = Layer.mergeAll(NodeServices.layer).pipe(
28 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-attachment-claims-" })),
29 Layer.provideMerge(DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer))),
30);
31
32const stagePendingUpload = Effect.fn("test.stagePendingUpload")(function* (input: {
33 readonly name: string;
34 readonly bytes: Uint8Array;
35 readonly mimeType: string;
36}) {
37 const config = yield* ServerConfig.ServerConfig;
38 const pendingId = createPendingAttachmentId();
39 const attachment: ChatAttachment = {
40 type: "image",
41 id: ChatAttachmentId.make(pendingId!),
42 name: input.name,
43 mimeType: input.mimeType,
44 sizeBytes: input.bytes.byteLength,
45 };
46 yield* Effect.sync(() => {
47 NodeFS.mkdirSync(config.attachmentsDir, { recursive: true });
48 NodeFS.writeFileSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`), input.bytes);
49 });
50 return attachment;
51});
52
53describe("AttachmentClaims", () => {
54 it.effect("copies a pending document's extraction without parsing it again", () =>
55 Effect.gen(function* () {
56 const config = yield* ServerConfig.ServerConfig;
57 const pendingId = createPendingAttachmentId("docx");
58 const path = NodePath.join(config.attachmentsDir, pendingId + ".docx");
59 NodeFS.writeFileSync(path, "invalid document");
60 NodeFS.writeFileSync(path + ".txt", "North calibration: 946");
61 const claimed = yield* claimPendingAttachments({
62 threadId: "thread-document",
63 attachments: [
64 {
65 type: "file",
66 id: ChatAttachmentId.make(pendingId),
67 name: "calibration.docx",
68 mimeType: "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
69 sizeBytes: 16,
70 },
71 ],
72 });
73 expect(claimed.claimedPaths).toHaveLength(2);
74 expect(NodeFS.readFileSync(claimed.claimedPaths[1]!, "utf8")).toBe("North calibration: 946");
75 const documents = yield* DocumentText.DocumentText;
76 expect(yield* documents.ensure(claimed.claimedPaths[0]!)).toBe(claimed.claimedPaths[1]);
77 yield* releaseClaimedAttachments(claimed.claimedPaths);
78 expect(NodeFS.existsSync(claimed.claimedPaths[1]!)).toBe(false);
79 expect(NodeFS.existsSync(path + ".txt")).toBe(true);
80 }).pipe(Effect.provide(layerTest)),
81 );
82
83 it.effect("rejects oversized image batches before copying a pending upload", () =>
84 Effect.gen(function* () {
85 const pending = yield* stagePendingUpload({
86 name: "pending.png",
87 bytes: new Uint8Array([1]),
88 mimeType: "image/png",
89 });
90 const images: ChatAttachment[] = Array.from({ length: 9 }, (_, index) => ({
91 type: "image",
92 id: ChatAttachmentId.make(`existing-image-${index}`),
93 name: `${index}.png`,
94 mimeType: "image/png",
95 sizeBytes: 10 * 1024 * 1024,
96 }));
97 const result = yield* Effect.exit(
98 claimPendingAttachments({
99 threadId: "thread-image-budget",
100 attachments: [pending, ...images],
101 }),
102 );
103 expect(result._tag).toBe("Failure");
104 if (result._tag === "Failure") expect(String(result.cause)).toContain("80 MiB");
105 const config = yield* ServerConfig.ServerConfig;
106 expect(NodeFS.readdirSync(config.attachmentsDir)).toHaveLength(1);
107 }).pipe(Effect.provide(layerTest)),
108 );
109
110 it.effect("claims a pending upload into the thread store and rewrites the id", () =>
111 Effect.gen(function* () {
112 const pending = yield* stagePendingUpload({
113 name: "screenshot.png",
114 bytes: new Uint8Array([1, 2, 3, 4]),
115 mimeType: "image/png",
116 });
117 const config = yield* ServerConfig.ServerConfig;
118
119 const claimed = yield* claimPendingAttachments({
120 threadId: "thread-claims-1",
121 attachments: [pending],
122 });
123
124 expect(claimed.attachments).toHaveLength(1);
125 const attachment = claimed.attachments[0]!;
126 expect(attachmentIsPendingUpload(attachment)).toBe(false);
127 expect(parseThreadSegmentFromAttachmentId(attachment.id)).toBe("thread-claims-1");
128 expect(claimed.claimedPaths).toHaveLength(1);
129 expect(NodeFS.existsSync(claimed.claimedPaths[0]!)).toBe(true);
130 const pendingFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
131 entry.startsWith("pending-"),
132 );
133 expect(pendingFiles).toHaveLength(1);
134 }).pipe(Effect.provide(layerTest)),
135 );
136
137 it.effect("rejects a size mismatch and cleans up earlier claims from the batch", () =>
138 Effect.gen(function* () {
139 const good = yield* stagePendingUpload({
140 name: "one.png",
141 bytes: new Uint8Array([1, 2, 3, 4]),
142 mimeType: "image/png",
143 });
144 const staged = yield* stagePendingUpload({
145 name: "two.png",
146 bytes: new Uint8Array([1, 2, 3, 4]),
147 mimeType: "image/png",
148 });
149 const mismatched = { ...staged, sizeBytes: 999 };
150 const config = yield* ServerConfig.ServerConfig;
151
152 const result = yield* Effect.exit(
153 claimPendingAttachments({
154 threadId: "thread-claims-4",
155 attachments: [good, mismatched],
156 }),
157 );
158
159 expect(result._tag).toBe("Failure");
160 if (result._tag === "Failure") {
161 const error = result.cause;
162 expect(String(error)).toContain("stored size does not match");
163 }
164 const claimedFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
165 entry.startsWith("thread-claims-4-"),
166 );
167 expect(claimedFiles).toHaveLength(0);
168 }).pipe(Effect.provide(layerTest)),
169 );
170
171 it.effect("rejects duplicate attachment ids before copying anything", () =>
172 Effect.gen(function* () {
173 const pending = yield* stagePendingUpload({
174 name: "dupe.png",
175 bytes: new Uint8Array([5, 5, 5]),
176 mimeType: "image/png",
177 });
178 const config = yield* ServerConfig.ServerConfig;
179
180 const result = yield* Effect.exit(
181 claimPendingAttachments({
182 threadId: "thread-claims-dupe",
183 attachments: [pending, pending],
184 }),
185 );
186
187 expect(result._tag).toBe("Failure");
188 if (result._tag === "Failure") {
189 expect(String(result.cause)).toContain("Duplicate attachment ids");
190 }
191 expect(
192 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
193 entry.startsWith("thread-claims-dupe-"),
194 ),
195 ).toHaveLength(0);
196 expect(
197 NodeFS.readdirSync(config.attachmentsDir).filter((entry) => entry.startsWith("pending-")),
198 ).toHaveLength(1);
199 }).pipe(Effect.provide(layerTest)),
200 );
201
202 it.effect("rolls back a copy that finishes after the claim is interrupted", () =>
203 Effect.gen(function* () {
204 const pending = yield* stagePendingUpload({
205 name: "interrupt.png",
206 bytes: new Uint8Array([7, 7, 7]),
207 mimeType: "image/png",
208 });
209 const config = yield* ServerConfig.ServerConfig;
210 const fileSystem = yield* FileSystem.FileSystem;
211
212 const copyArgs = yield* Deferred.make<readonly [string, string]>();
213 const copyDone = yield* Deferred.make<void>();
214 // fs.copyFile cannot be cancelled; its separate writer must finish after the awaiting effect is interrupted.
215 const gatedFileSystem = Object.create(fileSystem, {
216 copyFile: {
217 value: (from: string, to: string) =>
218 Deferred.succeed(copyArgs, [from, to] as const).pipe(
219 Effect.andThen(Deferred.await(copyDone)),
220 ),
221 },
222 }) as FileSystem.FileSystem;
223
224 const claimFiber = yield* claimPendingAttachments({
225 threadId: "thread-claims-interrupt",
226 attachments: [pending],
227 }).pipe(Effect.provideService(FileSystem.FileSystem, gatedFileSystem), Effect.forkChild);
228
229 const [sourcePath, claimedPath] = yield* Deferred.await(copyArgs);
230 claimFiber.interruptUnsafe();
231 const writer = yield* fileSystem
232 .copyFile(sourcePath, claimedPath)
233 .pipe(Effect.ensuring(Deferred.succeed(copyDone, undefined)), Effect.forkChild);
234
235 const exit = yield* Fiber.await(claimFiber);
236 const writerExit = yield* Fiber.await(writer);
237 expect(Exit.isSuccess(writerExit)).toBe(true);
238
239 expect(Exit.hasInterrupts(exit)).toBe(true);
240 expect(NodeFS.existsSync(claimedPath)).toBe(false);
241 expect(NodeFS.existsSync(sourcePath)).toBe(true);
242 expect(
243 NodeFS.readdirSync(config.attachmentsDir).filter((entry) => entry.startsWith("pending-")),
244 ).toHaveLength(1);
245 }).pipe(Effect.provide(layerTest)),
246 );
247
248 it.effect("rolls back an extracted sidecar published while the claim is interrupted", () =>
249 Effect.gen(function* () {
250 const pending = yield* stagePendingUpload({
251 name: "interrupt.png",
252 bytes: new Uint8Array([7]),
253 mimeType: "image/png",
254 });
255 const started = yield* Deferred.make<string>();
256 const publish = yield* Deferred.make<void>();
257 const fileSystem = yield* FileSystem.FileSystem;
258 const claimFiber = yield* claimPendingAttachments({
259 threadId: "thread-extraction-interrupt",
260 attachments: [pending],
261 }).pipe(
262 Effect.provideService(
263 DocumentText.DocumentText,
264 DocumentText.DocumentText.of({
265 ensure: (path) =>
266 Deferred.succeed(started, path).pipe(
267 Effect.andThen(Deferred.await(publish)),
268 Effect.andThen(fileSystem.writeFileString(path + ".txt", "extracted")),
269 Effect.as(path + ".txt"),
270 Effect.orDie,
271 Effect.uninterruptible,
272 ),
273 }),
274 ),
275 Effect.forkChild,
276 );
277 const claimedPath = yield* Deferred.await(started);
278 claimFiber.interruptUnsafe();
279 yield* Deferred.succeed(publish, undefined);
280 expect(Exit.hasInterrupts(yield* Fiber.await(claimFiber))).toBe(true);
281 expect(NodeFS.existsSync(claimedPath)).toBe(false);
282 expect(NodeFS.existsSync(claimedPath + ".txt")).toBe(false);
283 const config = yield* ServerConfig.ServerConfig;
284 expect(NodeFS.readdirSync(config.attachmentsDir)).toHaveLength(1);
285 }).pipe(Effect.provide(layerTest)),
286 );
287});