1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodePath from "node:path";
4
5import * as NodeServices from "@effect/platform-node/NodeServices";
6import { describe, expect, it } from "@effect/vitest";
7import { ChatAttachmentId, type ChatAttachment } from "@t3tools/contracts";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Exit from "effect/Exit";
11import * as Fiber from "effect/Fiber";
12import * as FileSystem from "effect/FileSystem";
13import * as Layer from "effect/Layer";
14
15import {
16 createPendingAttachmentId,
17 parseThreadSegmentFromAttachmentId,
18} from "../attachmentStore.ts";
19import * as ServerConfig from "../config.ts";
20import {
21 attachmentIsPendingUpload,
22 claimPendingAttachments,
23 releaseClaimedAttachments,
24} from "./AttachmentClaims.ts";
25
26const layerTest = Layer.mergeAll(NodeServices.layer).pipe(
27 Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-attachment-claims-" })),
28 Layer.provideMerge(NodeServices.layer),
29);
30
31const stagePendingUpload = Effect.fn("test.stagePendingUpload")(function* (input: {
32 readonly name: string;
33 readonly bytes: Uint8Array;
34 readonly mimeType: string;
35}) {
36 const config = yield* ServerConfig.ServerConfig;
37 const pendingId = createPendingAttachmentId();
38 const attachment: ChatAttachment = {
39 type: "image",
40 id: ChatAttachmentId.make(pendingId!),
41 name: input.name,
42 mimeType: input.mimeType,
43 sizeBytes: input.bytes.byteLength,
44 };
45 yield* Effect.sync(() => {
46 NodeFS.mkdirSync(config.attachmentsDir, { recursive: true });
47 NodeFS.writeFileSync(NodePath.join(config.attachmentsDir, `${pendingId}.png`), input.bytes);
48 });
49 return attachment;
50});
51
52describe("AttachmentClaims", () => {
53 it.effect("accepts 100 existing files without claiming new copies", () =>
54 Effect.gen(function* () {
55 const attachments: ChatAttachment[] = Array.from({ length: 100 }, (_, index) => ({
56 type: "file",
57 id: ChatAttachmentId.make(`existing-file-${index}`),
58 name: `${index}.txt`,
59 mimeType: "text/plain",
60 sizeBytes: 1,
61 }));
62 const claimed = yield* claimPendingAttachments({ threadId: "thread-many", attachments });
63 expect(claimed.attachments).toEqual(attachments);
64 expect(claimed.claimedPaths).toEqual([]);
65 }).pipe(Effect.provide(layerTest)),
66 );
67
68 it.effect("rejects oversized image batches before copying a pending upload", () =>
69 Effect.gen(function* () {
70 const pending = yield* stagePendingUpload({
71 name: "pending.png",
72 bytes: new Uint8Array([1]),
73 mimeType: "image/png",
74 });
75 const images: ChatAttachment[] = Array.from({ length: 9 }, (_, index) => ({
76 type: "image",
77 id: ChatAttachmentId.make(`existing-image-${index}`),
78 name: `${index}.png`,
79 mimeType: "image/png",
80 sizeBytes: 10 * 1024 * 1024,
81 }));
82 const result = yield* Effect.exit(
83 claimPendingAttachments({
84 threadId: "thread-image-budget",
85 attachments: [pending, ...images],
86 }),
87 );
88 expect(result._tag).toBe("Failure");
89 if (result._tag === "Failure") expect(String(result.cause)).toContain("80 MiB");
90 const config = yield* ServerConfig.ServerConfig;
91 expect(NodeFS.readdirSync(config.attachmentsDir)).toHaveLength(1);
92 }).pipe(Effect.provide(layerTest)),
93 );
94
95 it.effect("claims a pending upload into the thread store and rewrites the id", () =>
96 Effect.gen(function* () {
97 const pending = yield* stagePendingUpload({
98 name: "screenshot.png",
99 bytes: new Uint8Array([1, 2, 3, 4]),
100 mimeType: "image/png",
101 });
102 const config = yield* ServerConfig.ServerConfig;
103
104 const claimed = yield* claimPendingAttachments({
105 threadId: "thread-claims-1",
106 attachments: [pending],
107 });
108
109 expect(claimed.attachments).toHaveLength(1);
110 const attachment = claimed.attachments[0]!;
111 expect(attachmentIsPendingUpload(attachment)).toBe(false);
112 expect(parseThreadSegmentFromAttachmentId(attachment.id)).toBe("thread-claims-1");
113 expect(claimed.claimedPaths).toHaveLength(1);
114 expect(NodeFS.existsSync(claimed.claimedPaths[0]!)).toBe(true);
115 // The pending copy stays behind as the retry source.
116 const pendingFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
117 entry.startsWith("pending-"),
118 );
119 expect(pendingFiles).toHaveLength(1);
120 }).pipe(Effect.provide(layerTest)),
121 );
122
123 it.effect("passes already-claimed attachments through untouched", () =>
124 Effect.gen(function* () {
125 const stored: ChatAttachment = {
126 type: "image",
127 id: ChatAttachmentId.make("thread-claims-2-00000000-0000-4000-8000-000000000001"),
128 name: "existing.png",
129 mimeType: "image/png",
130 sizeBytes: 4,
131 };
132 const claimed = yield* claimPendingAttachments({
133 threadId: "thread-claims-2",
134 attachments: [stored],
135 });
136 expect(claimed.attachments[0]).toBe(stored);
137 expect(claimed.claimedPaths).toHaveLength(0);
138 }).pipe(Effect.provide(layerTest)),
139 );
140
141 it.effect("rejects a pending ref whose staged file is missing", () =>
142 Effect.gen(function* () {
143 const missing: ChatAttachment = {
144 type: "image",
145 id: ChatAttachmentId.make(createPendingAttachmentId()!),
146 name: "gone.png",
147 mimeType: "image/png",
148 sizeBytes: 4,
149 };
150 const result = yield* Effect.exit(
151 claimPendingAttachments({ threadId: "thread-claims-3", attachments: [missing] }),
152 );
153 expect(result._tag).toBe("Failure");
154 }).pipe(Effect.provide(layerTest)),
155 );
156
157 it.effect("rejects a size mismatch and cleans up earlier claims from the batch", () =>
158 Effect.gen(function* () {
159 const good = yield* stagePendingUpload({
160 name: "one.png",
161 bytes: new Uint8Array([1, 2, 3, 4]),
162 mimeType: "image/png",
163 });
164 const staged = yield* stagePendingUpload({
165 name: "two.png",
166 bytes: new Uint8Array([1, 2, 3, 4]),
167 mimeType: "image/png",
168 });
169 const mismatched = { ...staged, sizeBytes: 999 };
170 const config = yield* ServerConfig.ServerConfig;
171
172 const result = yield* Effect.exit(
173 claimPendingAttachments({
174 threadId: "thread-claims-4",
175 attachments: [good, mismatched],
176 }),
177 );
178
179 expect(result._tag).toBe("Failure");
180 if (result._tag === "Failure") {
181 const error = result.cause;
182 expect(String(error)).toContain("stored size does not match");
183 }
184 // The successfully claimed first attachment was rolled back.
185 const claimedFiles = NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
186 entry.startsWith("thread-claims-4-"),
187 );
188 expect(claimedFiles).toHaveLength(0);
189 }).pipe(Effect.provide(layerTest)),
190 );
191
192 it.effect("rejects duplicate attachment ids before copying anything", () =>
193 Effect.gen(function* () {
194 const pending = yield* stagePendingUpload({
195 name: "dupe.png",
196 bytes: new Uint8Array([5, 5, 5]),
197 mimeType: "image/png",
198 });
199 const config = yield* ServerConfig.ServerConfig;
200
201 const result = yield* Effect.exit(
202 claimPendingAttachments({
203 threadId: "thread-claims-dupe",
204 attachments: [pending, pending],
205 }),
206 );
207
208 expect(result._tag).toBe("Failure");
209 if (result._tag === "Failure") {
210 expect(String(result.cause)).toContain("Duplicate attachment ids");
211 }
212 expect(
213 NodeFS.readdirSync(config.attachmentsDir).filter((entry) =>
214 entry.startsWith("thread-claims-dupe-"),
215 ),
216 ).toHaveLength(0);
217 // The pending source survives for a corrected retry.
218 expect(
219 NodeFS.readdirSync(config.attachmentsDir).filter((entry) => entry.startsWith("pending-")),
220 ).toHaveLength(1);
221 }).pipe(Effect.provide(layerTest)),
222 );
223
224 it.effect("rolls back a copy that finishes after the claim is interrupted", () =>
225 Effect.gen(function* () {
226 const pending = yield* stagePendingUpload({
227 name: "interrupt.png",
228 bytes: new Uint8Array([7, 7, 7]),
229 mimeType: "image/png",
230 });
231 const config = yield* ServerConfig.ServerConfig;
232 const fileSystem = yield* FileSystem.FileSystem;
233
234 const copyArgs = yield* Deferred.make<readonly [string, string]>();
235 const copyDone = yield* Deferred.make<void>();
236 // fs.copyFile cannot be cancelled: the write runs on a separate fiber
237 // while the claim's copyFile only observes its completion, so the file
238 // still lands after the awaiting effect is interrupted.
239 const gatedFileSystem = Object.create(fileSystem, {
240 copyFile: {
241 value: (from: string, to: string) =>
242 Deferred.succeed(copyArgs, [from, to] as const).pipe(
243 Effect.andThen(Deferred.await(copyDone)),
244 ),
245 },
246 }) as FileSystem.FileSystem;
247
248 const claimFiber = yield* claimPendingAttachments({
249 threadId: "thread-claims-interrupt",
250 attachments: [pending],
251 }).pipe(Effect.provideService(FileSystem.FileSystem, gatedFileSystem), Effect.forkChild);
252
253 const [sourcePath, claimedPath] = yield* Deferred.await(copyArgs);
254 // Request the interrupt while the copy is in flight; the claim's
255 // uninterruptible copy region must wait for the write to settle.
256 claimFiber.interruptUnsafe();
257 // Settle the gate on the writer's exit too: the claim's masked await
258 // cannot be unwound, so a failed write must still release it.
259 const writer = yield* fileSystem
260 .copyFile(sourcePath, claimedPath)
261 .pipe(Effect.ensuring(Deferred.succeed(copyDone, undefined)), Effect.forkChild);
262
263 const exit = yield* Fiber.await(claimFiber);
264 const writerExit = yield* Fiber.await(writer);
265 expect(Exit.isSuccess(writerExit)).toBe(true);
266
267 expect(Exit.hasInterrupts(exit)).toBe(true);
268 // Rollback removed the copy that landed after the interrupt...
269 expect(NodeFS.existsSync(claimedPath)).toBe(false);
270 // ...and the pending source survived for a retry.
271 expect(NodeFS.existsSync(sourcePath)).toBe(true);
272 expect(
273 NodeFS.readdirSync(config.attachmentsDir).filter((entry) => entry.startsWith("pending-")),
274 ).toHaveLength(1);
275 }).pipe(Effect.provide(layerTest)),
276 );
277
278 it.effect("releaseClaimedAttachments removes claimed copies best-effort", () =>
279 Effect.gen(function* () {
280 const pending = yield* stagePendingUpload({
281 name: "release.png",
282 bytes: new Uint8Array([9, 9]),
283 mimeType: "image/png",
284 });
285 const claimed = yield* claimPendingAttachments({
286 threadId: "thread-claims-5",
287 attachments: [pending],
288 });
289 expect(NodeFS.existsSync(claimed.claimedPaths[0]!)).toBe(true);
290 yield* releaseClaimedAttachments(claimed.claimedPaths);
291 expect(NodeFS.existsSync(claimed.claimedPaths[0]!)).toBe(false);
292 // Releasing again is a no-op, not an error.
293 yield* releaseClaimedAttachments(claimed.claimedPaths);
294 }).pipe(Effect.provide(layerTest)),
295 );
296
297 it.effect("treats malformed ids as non-pending and passes them through", () =>
298 Effect.gen(function* () {
299 // Not a parseable pending ref: nothing to claim, so the ref flows
300 // through unchanged and downstream delivery decides what to do with it.
301 const bogus: ChatAttachment = {
302 type: "image",
303 id: ChatAttachmentId.make("pending-not-a-uuid"),
304 name: "weird.png",
305 mimeType: "image/png",
306 sizeBytes: 1,
307 };
308 const claimed = yield* claimPendingAttachments({
309 threadId: "thread-claims-6",
310 attachments: [bogus],
311 });
312 expect(claimed.attachments[0]).toBe(bogus);
313 expect(claimed.claimedPaths).toHaveLength(0);
314 }).pipe(Effect.provide(layerTest)),
315 );
316});