fix: dump stream

app/server/modules/repositories/__tests__/repositories.controller.test.ts

@@ -1,5 +1,6 @@
 import { test, describe, expect, spyOn } from "bun:test";
 import crypto from "node:crypto";
+import { PassThrough } from "node:stream";
 import { createApp } from "~/server/app";
 import { db } from "~/server/db/db";
 import { repositoriesTable } from "~/server/db/schema";
@@ -299,4 +300,42 @@ describe("repositories updates", () => {
 			}
 		});
 	});
+
+	describe("dump snapshot", () => {
+		test("continues streaming a download after the request signal aborts", async () => {
+			const { headers, organizationId } = await createTestSession();
+			const repository = await createRepositoryRecord(organizationId);
+			const { repositoriesService } = await import("~/server/modules/repositories/repositories.service");
+
+			const stream = new PassThrough();
+			const expectedContent = "downloaded snapshot contents";
+
+			const dumpSnapshotSpy = spyOn(repositoriesService, "dumpSnapshot").mockResolvedValue({
+				stream,
+				completion: Promise.resolve(),
+				abort: () => {
+					stream.destroy(new Error("download aborted"));
+				},
+				filename: "snapshot.txt",
+				contentType: "application/octet-stream",
+			});
+
+			try {
+				const controller = new AbortController();
+				const response = await app.request(`/api/v1/repositories/${repository.shortId}/snapshots/test-snapshot/dump`, {
+					headers,
+					signal: controller.signal,
+				});
+
+				queueMicrotask(() => {
+					controller.abort();
+					stream.end(expectedContent);
+				});
+
+				await expect(response.text()).resolves.toBe(expectedContent);
+			} finally {
+				dumpSnapshotSpy.mockRestore();
+			}
+		});
+	});
 });

app/server/modules/repositories/repositories.controller.ts

@@ -195,15 +195,28 @@ export const repositoriesController = new Hono()
 		const { path, kind } = c.req.valid("query");
 
 		const dumpStream = await repositoriesService.dumpSnapshot(shortId, snapshotId, path, kind);
-		const signal = c.req.raw.signal;
+		const sourceStream = Readable.toWeb(dumpStream.stream) as unknown as ReadableStream<Uint8Array>;
+		const reader = sourceStream.getReader();
+		const webStream = new ReadableStream<Uint8Array>({
+			async pull(controller) {
+				try {
+					const { done, value } = await reader.read();
 
-		if (signal.aborted) {
-			dumpStream.abort();
-		} else {
-			signal.addEventListener("abort", () => dumpStream.abort(), { once: true });
-		}
+					if (done) {
+						controller.close();
+						return;
+					}
 
-		const webStream = Readable.toWeb(dumpStream.stream) as unknown as ReadableStream<Uint8Array>;
+					controller.enqueue(value);
+				} catch (error) {
+					controller.error(error);
+				}
+			},
+			async cancel(reason) {
+				dumpStream.abort();
+				await reader.cancel(reason).catch(() => {});
+			},
+		});
 
 		return new Response(webStream, {
 			status: 200,