diff --git a/.changeset/stream-media-uploads.md b/.changeset/stream-media-uploads.md new file mode 100644 index 0000000000..5385518e18 --- /dev/null +++ b/.changeset/stream-media-uploads.md @@ -0,0 +1,7 @@ +--- +"emdash": patch +"@emdash-cms/admin": patch +--- + +Fixes media uploads with native R2 storage, keeps client-side hashing optional, and prevents deduplication from returning media with a different type or size. +Images larger than 8 MiB skip server-generated placeholders in signed and streamed upload flows. diff --git a/docs/src/content/docs/contributing/architecture.mdx b/docs/src/content/docs/contributing/architecture.mdx index 371cd1210d..ede47e4540 100644 --- a/docs/src/content/docs/contributing/architecture.mdx +++ b/docs/src/content/docs/contributing/architecture.mdx @@ -245,12 +245,12 @@ Portable Text fields edit in TipTap (ProseMirror). Content is converted at the l ### Signed uploads -Media uploads bypass Worker body-size limits with direct-to-storage signed URLs: +Media uploads use direct-to-storage signed URLs when the adapter supports them and a same-origin streaming endpoint otherwise: 1. The client requests an upload URL (`POST /api/media/upload-url`). -2. The client uploads directly to the signed URL (R2 or S3). +2. The client uploads to the returned target. S3-compatible adapters can return a signed URL that bypasses Worker body-size limits; native R2 bindings and local storage return an EmDash streaming endpoint. 3. The client confirms (`POST /api/media/:id/confirm`). 4. The server extracts metadata (dimensions, MIME type). diff --git a/packages/admin/src/lib/api/media.ts b/packages/admin/src/lib/api/media.ts index 9e0ccbdf10..1e70562536 100644 --- a/packages/admin/src/lib/api/media.ts +++ b/packages/admin/src/lib/api/media.ts @@ -13,10 +13,6 @@ import { type FindManyResult, } from "./client.js"; -/** - * Maximum length of the media filename search term. Mirrors the server-side - * zod schema (`q: z.string().trim().min(1).max(200)`); keep in sync. - */ export const MEDIA_SEARCH_MAX_LENGTH = 200; /** Trim and clamp a search term to the server-accepted range. */ @@ -103,6 +99,29 @@ interface UploadUrlResponse { expiresAt: string; } +interface ExistingMediaResponse { + existing: true; + mediaId: string; + storageKey: string; + url: string; +} + +const MAX_CLIENT_HASH_BYTES = 8 * 1024 * 1024; + +async function computeContentHash(file: File): Promise { + const subtle = globalThis.crypto?.subtle; + if (!subtle || file.size === 0 || file.size > MAX_CLIENT_HASH_BYTES) return undefined; + try { + const hash = await subtle.digest("SHA-1", await file.arrayBuffer()); + const hex = Array.from(new Uint8Array(hash), (byte) => byte.toString(16).padStart(2, "0")).join( + "", + ); + return `sha1:${hex}`; + } catch { + return undefined; + } +} + /** * Try to get a signed upload URL * Returns null if signed URLs are not supported (e.g., local storage) @@ -110,8 +129,9 @@ interface UploadUrlResponse { async function getUploadUrl( file: File, opts?: { fieldId?: string }, -): Promise { +): Promise { try { + const contentHash = await computeContentHash(file); const response = await apiFetch(`${API_BASE}/media/upload-url`, { method: "POST", headers: { "Content-Type": "application/json" }, @@ -119,6 +139,7 @@ async function getUploadUrl( filename: file.name, contentType: file.type, size: file.size, + ...(contentHash ? { contentHash } : {}), ...(opts?.fieldId ? { fieldId: opts.fieldId } : {}), }), }); @@ -128,7 +149,10 @@ async function getUploadUrl( return null; } - return parseApiResponse(response, i18n._(msg`Failed to get upload URL`)); + return parseApiResponse( + response, + i18n._(msg`Failed to get upload URL`), + ); } catch (error) { // If the endpoint doesn't exist, fall back to direct upload if (error instanceof TypeError && error.message.includes("fetch")) { @@ -234,6 +258,9 @@ export async function uploadMedia(file: File, opts?: { fieldId?: string }): Prom // Signed URLs not supported, use direct upload return uploadMediaDirect(file, opts); } + if ("existing" in uploadInfo) { + return fetchMediaItem(uploadInfo.mediaId); + } // Upload directly to storage via signed URL await uploadToSignedUrl(file, uploadInfo); diff --git a/packages/admin/src/router.tsx b/packages/admin/src/router.tsx index c39a322770..dba723fc94 100644 --- a/packages/admin/src/router.tsx +++ b/packages/admin/src/router.tsx @@ -1341,7 +1341,9 @@ function MediaPage() { isLoading={isLoading || isFetchingNextPage} hasMore={!!hasNextPage} onLoadMore={() => void fetchNextPage()} - onUpload={(file) => uploadMutation.mutate(file)} + onUpload={async (file) => { + await uploadMutation.mutateAsync(file); + }} onLocalSearchChange={setSearch} onLocalMimeFilterChange={setMimeFilter} /> diff --git a/packages/admin/tests/lib/media-upload.test.ts b/packages/admin/tests/lib/media-upload.test.ts new file mode 100644 index 0000000000..9ebde25aa3 --- /dev/null +++ b/packages/admin/tests/lib/media-upload.test.ts @@ -0,0 +1,198 @@ +import { afterEach, expect, it, vi } from "vitest"; + +import { uploadMedia } from "../../src/lib/api/media.js"; + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +it("deduplicates uploads using the file content hash", async () => { + let uploadUrlBody: Record | undefined; + const fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url === "/_emdash/api/media/upload-url") { + if (typeof init?.body !== "string") throw new TypeError("Expected a JSON request body"); + uploadUrlBody = JSON.parse(init.body) as Record; + return Response.json({ + success: true, + data: { + existing: true, + mediaId: "existing-media", + storageKey: "existing.pdf", + url: "/_emdash/api/media/file/existing.pdf", + }, + }); + } + if (url === "/_emdash/api/media/existing-media") { + return Response.json({ + success: true, + data: { + item: { + id: "existing-media", + filename: "existing.pdf", + mimeType: "application/pdf", + url: "/_emdash/api/media/file/existing.pdf", + storageKey: "existing.pdf", + size: 3, + createdAt: "2026-01-01T00:00:00.000Z", + }, + }, + }); + } + return new Response(null, { status: 500 }); + }); + const file = new File([new Uint8Array([97, 98, 99])], "document.pdf", { + type: "application/pdf", + }); + + const item = await uploadMedia(file); + + expect(uploadUrlBody?.contentHash).toBe("sha1:a9993e364706816aba3e25717850c26c9cd0d89d"); + expect(item.id).toBe("existing-media"); + expect(fetch).toHaveBeenCalledTimes(2); +}); + +it("uploads without deduplication when Web Crypto is unavailable", async () => { + vi.stubGlobal("crypto", {}); + let uploadUrlBody: Record | undefined; + vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url === "/_emdash/api/media/upload-url") { + if (typeof init?.body !== "string") throw new TypeError("Expected a JSON request body"); + uploadUrlBody = JSON.parse(init.body) as Record; + return Response.json({ + success: true, + data: { + uploadUrl: "/_emdash/api/media/new-media/upload", + method: "PUT", + headers: { "Content-Type": "application/pdf" }, + mediaId: "new-media", + storageKey: "new.pdf", + expiresAt: "2026-01-01T01:00:00.000Z", + }, + }); + } + if (url === "/_emdash/api/media/new-media/upload") { + return Response.json({ success: true, data: { uploaded: true, size: 3 } }); + } + if (url === "/_emdash/api/media/new-media/confirm") { + return Response.json({ + success: true, + data: { + item: { + id: "new-media", + filename: "new.pdf", + mimeType: "application/pdf", + url: "/_emdash/api/media/file/new.pdf", + storageKey: "new.pdf", + size: 3, + createdAt: "2026-01-01T00:00:00.000Z", + }, + }, + }); + } + return new Response(null, { status: 500 }); + }); + const file = new File([new Uint8Array([1, 2, 3])], "new.pdf", { + type: "application/pdf", + }); + + const item = await uploadMedia(file); + + expect(uploadUrlBody).not.toHaveProperty("contentHash"); + expect(item.id).toBe("new-media"); +}); + +it("uploads without deduplication when content hashing fails", async () => { + vi.stubGlobal("crypto", { + subtle: { + digest: vi.fn().mockRejectedValue(new Error("SHA-1 unavailable")), + }, + }); + let uploadUrlBody: Record | undefined; + const fetch = vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url === "/_emdash/api/media/upload-url") { + if (typeof init?.body !== "string") throw new TypeError("Expected a JSON request body"); + uploadUrlBody = JSON.parse(init.body) as Record; + return new Response(null, { status: 501 }); + } + if (url === "/_emdash/api/media") { + return Response.json({ + success: true, + data: { + item: { + id: "new-media", + filename: "new.pdf", + mimeType: "application/pdf", + url: "/_emdash/api/media/file/new.pdf", + storageKey: "new.pdf", + size: 3, + createdAt: "2026-01-01T00:00:00.000Z", + }, + }, + }); + } + return new Response(null, { status: 500 }); + }); + const file = new File([new Uint8Array([1, 2, 3])], "new.pdf", { + type: "application/pdf", + }); + + const item = await uploadMedia(file); + + expect(uploadUrlBody).not.toHaveProperty("contentHash"); + expect(item.id).toBe("new-media"); + expect(fetch).toHaveBeenCalledTimes(2); +}); + +it("does not deduplicate empty files by their shared hash", async () => { + let uploadUrlBody: Record | undefined; + vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url === "/_emdash/api/media/upload-url") { + if (typeof init?.body !== "string") throw new TypeError("Expected a JSON request body"); + uploadUrlBody = JSON.parse(init.body) as Record; + return Response.json({ + success: true, + data: { + uploadUrl: "/_emdash/api/media/empty-media/upload", + method: "PUT", + headers: { "Content-Type": "application/pdf" }, + mediaId: "empty-media", + storageKey: "empty.pdf", + expiresAt: "2026-01-01T01:00:00.000Z", + }, + }); + } + if (url === "/_emdash/api/media/empty-media/upload") { + return Response.json({ success: true, data: { uploaded: true, size: 0 } }); + } + if (url === "/_emdash/api/media/empty-media/confirm") { + return Response.json({ + success: true, + data: { + item: { + id: "empty-media", + filename: "empty.pdf", + mimeType: "application/pdf", + url: "/_emdash/api/media/file/empty.pdf", + storageKey: "empty.pdf", + size: 0, + createdAt: "2026-01-01T00:00:00.000Z", + }, + }, + }); + } + return new Response(null, { status: 500 }); + }); + + await uploadMedia(new File([], "empty.pdf", { type: "application/pdf" })); + + expect(uploadUrlBody).not.toHaveProperty("contentHash"); +}); diff --git a/packages/admin/tests/router.test.tsx b/packages/admin/tests/router.test.tsx index 03ca75344f..daf817503e 100644 --- a/packages/admin/tests/router.test.tsx +++ b/packages/admin/tests/router.test.tsx @@ -98,6 +98,31 @@ vi.mock("../src/components/ContentEditor", () => ({ ), })); +vi.mock("../src/components/MediaLibrary", () => ({ + MediaLibrary: ({ onUpload }: { onUpload?: (file: File) => Promise | void }) => { + const [uploadStatus, setUploadStatus] = React.useState("idle"); + + const upload = async () => { + setUploadStatus("uploading"); + try { + await onUpload?.(new File([new Uint8Array([1, 2, 3])], "photo.png", { type: "image/png" })); + setUploadStatus("success"); + } catch { + setUploadStatus("error"); + } + }; + + return ( +
+ + {uploadStatus} +
+ ); + }, +})); + // --------------------------------------------------------------------------- // Fixtures // --------------------------------------------------------------------------- @@ -166,6 +191,64 @@ describe("ConfigurationLoadingScreen", () => { }); }); +describe("MediaPage – upload completion", () => { + let mockFetch: ReturnType; + + beforeEach(() => { + mockFetch = createMockFetch(); + mockFetch + .on("GET", "/_emdash/api/manifest", { data: MANIFEST }) + .on("GET", "/_emdash/api/auth/me", { + data: { id: "user_01", role: 60 }, + }) + .on("GET", "/_emdash/api/media", { + data: { items: [], nextCursor: undefined }, + }); + }); + + afterEach(() => { + mockFetch.restore(); + }); + + it("waits for the upload request and propagates its failure", async () => { + const { router, TestApp } = buildRouter(); + await router.navigate({ to: "/media" }); + + const screen = await render(); + await expect.element(screen.getByText("idle")).toBeInTheDocument(); + + const interceptedFetch = globalThis.fetch; + let rejectUploadUrl: ((reason: Error) => void) | undefined; + let markUploadUrlStarted: () => void = () => undefined; + const uploadUrlStarted = new Promise((resolve) => { + markUploadUrlStarted = resolve; + }); + globalThis.fetch = (input, init) => { + const url = + typeof input === "string" ? input : input instanceof URL ? input.toString() : input.url; + if (url === "/_emdash/api/media/upload-url" && init?.method === "POST") { + return new Promise((_resolve, reject) => { + rejectUploadUrl = reject; + markUploadUrlStarted(); + }); + } + return interceptedFetch(input, init); + }; + + try { + await screen.getByRole("button", { name: "Upload test file" }).click(); + await expect.element(screen.getByText("uploading")).toBeInTheDocument(); + + await uploadUrlStarted; + if (!rejectUploadUrl) throw new Error("Upload URL request was not intercepted"); + rejectUploadUrl(new Error("connection closed")); + await expect.element(screen.getByText("error")).toBeInTheDocument(); + } finally { + globalThis.fetch = interceptedFetch; + } + }); +}); + // --------------------------------------------------------------------------- // Tests: ContentListPage – locale forwarded to "Add New" link // --------------------------------------------------------------------------- diff --git a/packages/cloudflare/tests/db/coalescing-d1.test.ts b/packages/cloudflare/tests/db/coalescing-d1.test.ts index 6b52455368..d4b778395c 100644 --- a/packages/cloudflare/tests/db/coalescing-d1.test.ts +++ b/packages/cloudflare/tests/db/coalescing-d1.test.ts @@ -177,6 +177,10 @@ describe("CoalescingD1Dialect", () => { const mock = createMockD1({ "insert into a (name) values (?)": { changes: 2, lastRowId: 7 }, "delete from a where 1 = 0": { changes: 0 }, + "delete from media where status = ? returning storage_key": { + rows: [{ storage_key: "expired.png" }], + changes: 1, + }, }); const db = createDb(mock); @@ -191,6 +195,12 @@ describe("CoalescingD1Dialect", () => { // Zero changes maps to undefined, matching kysely-d1. const deleted = await db.executeQuery(CompiledQuery.raw("delete from a where 1 = 0")); expect(deleted.numAffectedRows).toBeUndefined(); + + const returned = await db.executeQuery( + CompiledQuery.raw("delete from media where status = ? returning storage_key", ["pending"]), + ); + expect(returned.rows).toEqual([{ storage_key: "expired.png" }]); + expect(returned.numAffectedRows).toBe(1n); expect(mock.batchCalls).toHaveLength(0); }); diff --git a/packages/cloudflare/tests/db/d1-dialect.test.ts b/packages/cloudflare/tests/db/d1-dialect.test.ts index 4a14246ade..6b1cccde58 100644 --- a/packages/cloudflare/tests/db/d1-dialect.test.ts +++ b/packages/cloudflare/tests/db/d1-dialect.test.ts @@ -24,7 +24,7 @@ interface MockStatement { all: () => Promise; } -function createMockD1() { +function createMockD1(rows: Record[] = []) { const allCalls: string[] = []; let inFlight = 0; let maxInFlight = 0; @@ -45,7 +45,7 @@ function createMockD1() { await new Promise((resolve) => setTimeout(resolve, 5)); inFlight--; allCalls.push(sql); - return { success: true, results: [], meta: { changes: 0, last_row_id: 0 } }; + return { success: true, results: rows, meta: { changes: 0, last_row_id: 0 } }; }, }; return stmt; @@ -109,3 +109,22 @@ describe("EmDashD1Dialect (session path keeps the mutex)", () => { expect(maxInFlight()).toBe(1); }); }); + +describe("D1 write results", () => { + it.each([ + ["raw binding", RawBindingD1Dialect], + ["session", EmDashD1Dialect], + ])("preserves DELETE RETURNING rows through the %s dialect", async (_name, Dialect) => { + const rows = [{ storage_key: "expired.png" }]; + const { database } = createMockD1(rows); + const db = new Kysely({ dialect: new Dialect({ database }) }); + + const result = await db.executeQuery( + CompiledQuery.raw('delete from "media" where "status" = ? returning "storage_key"', [ + "pending", + ]), + ); + + expect(result.rows).toEqual(rows); + }); +}); diff --git a/packages/core/src/api/handlers/media.ts b/packages/core/src/api/handlers/media.ts index 2859a24101..1ad8b8e7c6 100644 --- a/packages/core/src/api/handlers/media.ts +++ b/packages/core/src/api/handlers/media.ts @@ -185,12 +185,12 @@ export async function handleMediaUpdate( export async function handleMediaDelete( db: Kysely, id: string, -): Promise> { +): Promise> { try { const repo = new MediaRepository(db); - const deleted = await repo.delete(id); + const storageKey = await repo.deleteWithStorageKey(id); - if (!deleted) { + if (!storageKey) { return { success: false, error: { @@ -202,7 +202,7 @@ export async function handleMediaDelete( return { success: true, - data: { deleted: true }, + data: { deleted: true, storageKey }, }; } catch { return { diff --git a/packages/core/src/api/openapi/document.ts b/packages/core/src/api/openapi/document.ts index 4dceded223..98ca1ce2f5 100644 --- a/packages/core/src/api/openapi/document.ts +++ b/packages/core/src/api/openapi/document.ts @@ -54,6 +54,7 @@ import { mediaListResponseSchema, mediaReadResponseSchema, mediaResponseSchema, + mediaStreamUploadResponseSchema, mediaUpdateBody, mediaUploadUrlBody, mediaUploadUrlResponseSchema, @@ -788,9 +789,9 @@ function buildMediaPaths(maxUploadSize: number) { "/_emdash/api/media/upload-url": { post: { operationId: "getMediaUploadUrl", - summary: "Get a signed URL for direct upload", + summary: "Get a media upload target", description: - "Returns a signed URL for direct-to-storage upload. Creates a pending media record.", + "Returns either a signed direct-to-storage URL or a same-origin streaming target. Creates a pending media record.", tags: ["Media"], requestBody: { content: { [JSON_CONTENT]: { schema: mediaUploadUrlBody(maxUploadSize) } } }, responses: { @@ -827,7 +828,37 @@ function buildMediaPaths(maxUploadSize: number) { }, }, ...authErrors, - ...standardErrors(400, 404, 500), + ...standardErrors(400, 404, 409, 500), + }, + }, + }, + "/_emdash/api/media/{id}/upload": { + put: { + operationId: "uploadPendingMedia", + summary: "Upload a pending media file through EmDash", + description: + "Streams a file to storage when the configured adapter cannot provide a signed upload URL. The Content-Type and byte count must match the pending media item.", + tags: ["Media"], + requestParams: { + path: z.object({ id: z.string().meta({ description: "Media ID" }) }), + }, + requestBody: { + required: true, + content: { + "*/*": { + schema: z.string().meta({ format: "binary" }), + }, + }, + }, + responses: { + "200": { + description: "File uploaded and ready for confirmation", + content: { + [JSON_CONTENT]: { schema: successEnvelope(mediaStreamUploadResponseSchema) }, + }, + }, + ...authErrors, + ...standardErrors(400, 404, 413, 500), }, }, }, diff --git a/packages/core/src/api/schemas/media.ts b/packages/core/src/api/schemas/media.ts index d95ad0a8ca..5a5f335dc1 100644 --- a/packages/core/src/api/schemas/media.ts +++ b/packages/core/src/api/schemas/media.ts @@ -74,7 +74,7 @@ export function mediaUploadUrlBody(maxSize: number) { size: z .number() .int() - .positive() + .nonnegative() .max(maxSize, `File size must not exceed ${formatFileSize(maxSize)}`), contentHash: z.string().optional(), fieldId: z.string().optional(), @@ -84,7 +84,7 @@ export function mediaUploadUrlBody(maxSize: number) { export const mediaConfirmBody = z .object({ - size: z.number().int().positive().optional(), + size: z.number().int().nonnegative().optional(), width: z.number().int().positive().optional(), height: z.number().int().positive().optional(), }) @@ -178,3 +178,10 @@ export const mediaConfirmResponseSchema = z item: mediaItemSchema.extend({ url: z.string() }), }) .meta({ id: "MediaConfirmResponse" }); + +export const mediaStreamUploadResponseSchema = z + .object({ + uploaded: z.literal(true), + size: z.number().int().nonnegative(), + }) + .meta({ id: "MediaStreamUploadResponse" }); diff --git a/packages/core/src/astro/integration/routes.ts b/packages/core/src/astro/integration/routes.ts index 1c0b98cd14..93f0efdfa3 100644 --- a/packages/core/src/astro/integration/routes.ts +++ b/packages/core/src/astro/integration/routes.ts @@ -223,6 +223,11 @@ export function injectCoreRoutes( entrypoint: resolveRoute("api/media/[id]/confirm.ts"), }); + injectRoute({ + pattern: "/_emdash/api/media/[id]/upload", + entrypoint: resolveRoute("api/media/[id]/upload.ts"), + }); + // Media provider routes injectRoute({ pattern: "/_emdash/api/media/providers", diff --git a/packages/core/src/astro/routes/api/media/[id].ts b/packages/core/src/astro/routes/api/media/[id].ts index b70123c8c9..acda88d3db 100644 --- a/packages/core/src/astro/routes/api/media/[id].ts +++ b/packages/core/src/astro/routes/api/media/[id].ts @@ -13,6 +13,8 @@ import { apiError, apiSuccess, handleError, unwrapResult } from "#api/error.js"; import { handleMediaUsageSummaries } from "#api/handlers/media-usage.js"; import { isParseError, parseBody, parseQuery } from "#api/parse.js"; import { mediaGetQuery, mediaUpdateBody } from "#api/schemas.js"; +import { MediaRepository } from "#db/repositories/media.js"; +import { removeUploadAttempt } from "#media/upload-attempts.js"; export const prerender = false; @@ -137,20 +139,27 @@ export const DELETE: APIRoute = async ({ params, locals }) => { ); if (ownerDenied) return ownerDenied; - // Delete file from storage via the storage adapter - if (emdash.storage) { - try { - await emdash.storage.delete(media.storageKey); - } catch { - // Best-effort — continue with database deletion - } - } - // Delete from database — site-settings cache invalidation happens // in `EmDashRuntime.handleMediaDelete` so MCP/plugin paths inherit it. const result = await emdash.handleMediaDelete(id); + if (!result.success) return unwrapResult(result); + if ( + typeof result.data !== "object" || + result.data === null || + !("storageKey" in result.data) || + typeof result.data.storageKey !== "string" + ) { + return apiError("MEDIA_DELETE_ERROR", "Failed to delete media", 500); + } - return unwrapResult(result); + if (emdash.storage) { + const repo = new MediaRepository(emdash.db); + await removeUploadAttempt(emdash.storage, repo, result.data.storageKey, { + allowUntracked: true, + }); + } + + return apiSuccess({ deleted: true }); } catch (error) { return handleError(error, "Failed to delete media", "MEDIA_DELETE_ERROR"); } diff --git a/packages/core/src/astro/routes/api/media/[id]/confirm.ts b/packages/core/src/astro/routes/api/media/[id]/confirm.ts index 944dabf10f..abf0019357 100644 --- a/packages/core/src/astro/routes/api/media/[id]/confirm.ts +++ b/packages/core/src/astro/routes/api/media/[id]/confirm.ts @@ -8,14 +8,16 @@ */ import type { APIRoute } from "astro"; -import { MediaRepository } from "emdash"; +import type { DownloadResult } from "emdash"; import { requireOwnerPerm, requirePerm } from "#api/authorize.js"; import { apiError, apiSuccess, handleError } from "#api/error.js"; import { isParseError, parseOptionalBody } from "#api/parse.js"; import { mediaConfirmBody } from "#api/schemas.js"; +import { MediaRepository } from "#db/repositories/media.js"; import { enrichImageMetadata } from "#media/enrich.js"; import type { MediaItem } from "#types"; +import { computeContentHash, MAX_CONTENT_HASH_BYTES } from "#utils/hash.js"; export const prerender = false; @@ -26,7 +28,7 @@ export const prerender = false; * on the very uploads that flow was designed for. LQIP is progressive * enhancement: large images simply ship without a server-generated placeholder. */ -const MAX_PLACEHOLDER_DOWNLOAD_BYTES = 8 * 1024 * 1024; +const MAX_PLACEHOLDER_DOWNLOAD_BYTES = MAX_CONTENT_HASH_BYTES; /** * Add URL to media item (relative URL for portability) @@ -38,6 +40,67 @@ function addUrlToMedia(item: MediaItem): MediaItem & { url: string } { }; } +async function cancelDownload(download: DownloadResult): Promise { + try { + await download.body.cancel(); + } catch (error) { + console.error("[media] confirm download cancellation failed:", error); + } +} + +async function forgetUploadAttempt(repo: MediaRepository, storageKey: string): Promise { + try { + await repo.deleteUploadAttempt(storageKey); + } catch (error) { + console.error("[media] confirm upload attempt cleanup failed:", error); + } +} + +async function confirmationConflict(repo: MediaRepository, id: string): Promise { + const current = await repo.findById(id); + if (!current) { + return apiError("NOT_FOUND", `Media item not found: ${id}`, 404); + } + if (current.status === "ready") { + await forgetUploadAttempt(repo, current.storageKey); + return apiSuccess({ item: addUrlToMedia(current) }); + } + if (current.status === "pending") { + return apiError("INVALID_STATE", "Media item changed during confirmation", 409); + } + return apiError("INVALID_STATE", `Media item is not pending: ${current.status}`, 400); +} + +async function consumeDownload(download: DownloadResult): Promise { + const reader = download.body.getReader(); + try { + const bytes = new Uint8Array(download.size); + let receivedSize = 0; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (receivedSize + value.byteLength > bytes.byteLength) { + throw new Error("Stored file exceeds its reported size"); + } + bytes.set(value, receivedSize); + receivedSize += value.byteLength; + } + if (receivedSize !== download.size) { + throw new Error("Stored file size does not match its reported size"); + } + return bytes; + } catch (error) { + try { + await reader.cancel(error); + } catch (cancelError) { + console.error("[media] confirm download cancellation failed:", cancelError); + } + throw error; + } finally { + reader.releaseLock(); + } +} + /** * Confirm upload completion */ @@ -68,90 +131,116 @@ export const POST: APIRoute = async ({ params, request, locals }) => { return apiError("NOT_FOUND", `Media item not found: ${id}`, 404); } - if (existing.status !== "pending") { - return apiError("INVALID_STATE", `Media item is not pending: ${existing.status}`, 400); - } - // Only the uploader or a user with media:edit_any can confirm/fail a pending upload const ownerDenied = requireOwnerPerm( user, existing.authorId ?? "", - "media:edit_own", + "media:upload", "media:edit_any", ); if (ownerDenied) return ownerDenied; - // Optionally verify the file exists in storage + if (existing.status === "ready") { + await forgetUploadAttempt(repo, existing.storageKey); + return apiSuccess({ item: addUrlToMedia(existing) }); + } + if (existing.status !== "pending") { + return apiError("INVALID_STATE", `Media item is not pending: ${existing.status}`, 400); + } + + if (body.size !== undefined && existing.size !== null && body.size !== existing.size) { + return apiError( + "UPLOAD_SIZE_MISMATCH", + "Confirmed size does not match the pending media item", + 400, + ); + } + + let confirmedSize = existing.size ?? body.size; + let contentHash = existing.contentHash; + let imageBytes: Uint8Array | undefined; + if (emdash.storage) { const exists = await emdash.storage.exists(existing.storageKey); if (!exists) { - // Mark as failed - await repo.markFailed(id); + const failed = await repo.markFailed(id, existing.storageKey); + if (!failed) return await confirmationConflict(repo, id); return apiError("FILE_NOT_FOUND", "File was not uploaded to storage", 400); } + + const storedFile = await emdash.storage.download(existing.storageKey); + if (confirmedSize !== undefined && storedFile.size !== confirmedSize) { + await cancelDownload(storedFile); + return apiError( + "UPLOAD_SIZE_MISMATCH", + "Stored file size does not match the pending media item", + 400, + ); + } + confirmedSize = storedFile.size; + + const isImage = existing.mimeType.startsWith("image/"); + const canBuffer = storedFile.size <= MAX_PLACEHOLDER_DOWNLOAD_BYTES; + const hasServerHash = + contentHash !== null && (await repo.hasUploadAttempt(existing.storageKey)); + if (canBuffer && (isImage || !hasServerHash)) { + const bytes = await consumeDownload(storedFile); + contentHash = bytes.byteLength > 0 ? await computeContentHash(bytes) : null; + if (isImage && bytes.byteLength > 0) imageBytes = bytes; + } else { + if (!hasServerHash) contentHash = null; + await cancelDownload(storedFile); + } + + if (isImage && !canBuffer) { + console.warn( + `[media] confirm skipping placeholder: object ${existing.storageKey} reported size ${storedFile.size} bytes (> ${MAX_PLACEHOLDER_DOWNLOAD_BYTES})`, + ); + } } - // For images, read the just-uploaded bytes back from storage once to - // generate LQIP placeholders (and server-side dimensions as a fallback). - // The signed-URL flow uploads directly to storage, so this confirm is the - // only point at which the server sees the bytes. Best-effort: a decode - // failure must not block the upload from being marked ready. We also cap - // the download size — buffering a large original into a Worker heap to - // compute a 32px blurhash would OOM on the uploads the signed-URL path - // exists to support, so oversized files skip the server-side placeholder. + // LQIP is best-effort; oversized images skip server-side placeholders. let blurhash: string | undefined; let dominantColor: string | undefined; let width = body.width; let height = body.height; - if (emdash.storage && existing.mimeType.startsWith("image/")) { - const knownSize = body.size ?? existing.size ?? undefined; - const tooLarge = knownSize != null && knownSize > MAX_PLACEHOLDER_DOWNLOAD_BYTES; - if (!tooLarge) { - try { - const { body: stream } = await emdash.storage.download(existing.storageKey); - const bytes = new Uint8Array(await new Response(stream).arrayBuffer()); - // Defense-in-depth for the unknown-size case: even though we - // already buffered it, refuse the decode so we don't also pay - // the (larger) RGBA allocation. - if (bytes.byteLength > MAX_PLACEHOLDER_DOWNLOAD_BYTES) { - console.warn( - `[media] confirm skipping placeholder: object ${existing.storageKey} is ${bytes.byteLength} bytes (> ${MAX_PLACEHOLDER_DOWNLOAD_BYTES})`, - ); - } else { - const enriched = await enrichImageMetadata(bytes, existing.mimeType, { - knownDimensions: - body.width != null && body.height != null - ? { width: body.width, height: body.height } - : undefined, - }); - blurhash = enriched.blurhash; - dominantColor = enriched.dominantColor; - width = width ?? enriched.width; - height = height ?? enriched.height; - } - } catch (error) { - console.error("[media] confirm placeholder generation failed:", error); - } - } else { - console.warn( - `[media] confirm skipping placeholder: object ${existing.storageKey} reported size ${knownSize} bytes (> ${MAX_PLACEHOLDER_DOWNLOAD_BYTES})`, - ); + if (imageBytes) { + try { + const enriched = await enrichImageMetadata(imageBytes, existing.mimeType, { + knownDimensions: + body.width != null && body.height != null + ? { width: body.width, height: body.height } + : undefined, + }); + blurhash = enriched.blurhash; + dominantColor = enriched.dominantColor; + width = width ?? enriched.width; + height = height ?? enriched.height; + } catch (error) { + console.error("[media] confirm placeholder generation failed:", error); } } // Confirm the upload - const item = await repo.confirmUpload(id, { - size: body.size, - width, - height, - blurhash, - dominantColor, - }); + const item = await repo.confirmUpload( + id, + { + size: confirmedSize, + width, + height, + blurhash, + dominantColor, + contentHash, + }, + existing.storageKey, + ); if (!item) { - return apiError("CONFIRM_FAILED", "Failed to confirm upload", 500); + return await confirmationConflict(repo, id); } + await forgetUploadAttempt(repo, item.storageKey); + // Add URL to the response (relative URL for portability) const itemWithUrl = addUrlToMedia(item); diff --git a/packages/core/src/astro/routes/api/media/[id]/upload.ts b/packages/core/src/astro/routes/api/media/[id]/upload.ts new file mode 100644 index 0000000000..37a575cba2 --- /dev/null +++ b/packages/core/src/astro/routes/api/media/[id]/upload.ts @@ -0,0 +1,262 @@ +import type { APIRoute } from "astro"; +import type { Storage } from "emdash"; +import { ulid } from "ulidx"; + +import { requireOwnerPerm, requirePerm } from "#api/authorize.js"; +import { apiError, apiSuccess, handleError } from "#api/error.js"; +import { MediaRepository } from "#db/repositories/media.js"; +import { normalizeMime } from "#media/mime.js"; +import { removeUploadAttempt } from "#media/upload-attempts.js"; +import { computeContentHash, MAX_CONTENT_HASH_BYTES } from "#utils/hash.js"; + +export const prerender = false; + +const INITIAL_HASH_BUFFER_BYTES = 64 * 1024; + +type FixedLengthStreamConstructor = new ( + expectedLength: number | bigint, +) => TransformStream; + +declare const FixedLengthStream: FixedLengthStreamConstructor | undefined; + +class UploadBodyError extends Error { + readonly code: "PAYLOAD_TOO_LARGE" | "UPLOAD_SIZE_MISMATCH"; + + constructor(code: "PAYLOAD_TOO_LARGE" | "UPLOAD_SIZE_MISMATCH") { + super(code); + this.code = code; + } +} + +function findUploadBodyError(error: unknown): UploadBodyError | null { + let current = error; + for (let depth = 0; depth < 5 && current instanceof Error; depth++) { + if (current instanceof UploadBodyError) return current; + current = current.cause; + } + return null; +} + +function preserveKnownLength( + body: ReadableStream, + expectedLength: number, +): ReadableStream { + if (typeof FixedLengthStream === "undefined") return body; + return body.pipeThrough(new FixedLengthStream(expectedLength)); +} + +function createUploadAttemptKey(key: string): string { + const pathSeparator = key.lastIndexOf("/"); + const extensionSeparator = key.lastIndexOf("."); + if (extensionSeparator > pathSeparator) { + return `${key.slice(0, extensionSeparator)}.${ulid()}${key.slice(extensionSeparator)}`; + } + return `${key}.${ulid()}`; +} + +async function getStoredSize(storage: Storage, key: string): Promise { + if (!(await storage.exists(key))) return null; + const download = await storage.download(key); + try { + return download.size; + } finally { + try { + await download.body.cancel(); + } catch (error) { + console.error("[media] upload download cancellation failed:", error); + } + } +} + +export const PUT: APIRoute = async ({ params, request, locals }) => { + const { emdash, user } = locals; + const { id } = params; + + const denied = requirePerm(user, "media:upload"); + if (denied) return denied; + + if (!id) { + return apiError("INVALID_REQUEST", "Media ID is required", 400); + } + + if (!emdash?.db) { + return apiError("NOT_CONFIGURED", "EmDash is not initialized", 500); + } + + if (!emdash.storage) { + return apiError("NO_STORAGE", "Storage not configured", 500); + } + + try { + const repo = new MediaRepository(emdash.db); + const media = await repo.findById(id); + if (!media) { + return apiError("NOT_FOUND", `Media item not found: ${id}`, 404); + } + + if (media.status !== "pending") { + return apiError("INVALID_STATE", `Media item is not pending: ${media.status}`, 400); + } + + const ownerDenied = requireOwnerPerm( + user, + media.authorId ?? "", + "media:upload", + "media:edit_any", + ); + if (ownerDenied) return ownerDenied; + + if (!Number.isSafeInteger(media.size) || media.size === null || media.size < 0) { + return apiError("INVALID_STATE", "Pending media item has no valid upload size", 400); + } + const expectedSize = media.size; + + const contentType = request.headers.get("Content-Type"); + if (!contentType || normalizeMime(contentType) !== media.mimeType) { + return apiError("INVALID_TYPE", "Upload content type does not match the media item", 400); + } + + const requestBody = + request.body ?? + (expectedSize === 0 + ? new ReadableStream({ + start(controller) { + controller.close(); + }, + }) + : null); + if (!requestBody) { + return apiError("NO_FILE", "No file provided", 400); + } + + const contentLength = request.headers.get("Content-Length"); + if (contentLength !== null) { + const declaredSize = Number(contentLength); + if (!Number.isSafeInteger(declaredSize) || declaredSize < 0) { + return apiError("INVALID_REQUEST", "Invalid Content-Length header", 400); + } + if (declaredSize > expectedSize) { + return apiError("PAYLOAD_TOO_LARGE", "Upload exceeds the expected size", 413); + } + if (declaredSize !== expectedSize) { + return apiError("UPLOAD_SIZE_MISMATCH", "Upload size does not match the media item", 400); + } + } + + let receivedSize = 0; + const shouldHash = expectedSize > 0 && expectedSize <= MAX_CONTENT_HASH_BYTES; + let hashBytes: Uint8Array | null = null; + const checkedBody = requestBody.pipeThrough( + new TransformStream({ + transform(chunk, controller) { + const offset = receivedSize; + receivedSize += chunk.byteLength; + if (receivedSize > expectedSize) { + controller.error(new UploadBodyError("PAYLOAD_TOO_LARGE")); + return; + } + if (shouldHash && chunk.byteLength > 0) { + if (!hashBytes) { + hashBytes = new Uint8Array( + Math.min(expectedSize, Math.max(INITIAL_HASH_BUFFER_BYTES, receivedSize)), + ); + } else if (receivedSize > hashBytes.byteLength) { + const grown = new Uint8Array( + Math.min(expectedSize, Math.max(receivedSize, hashBytes.byteLength * 2)), + ); + grown.set(hashBytes); + hashBytes = grown; + } + hashBytes.set(chunk, offset); + } + controller.enqueue(chunk); + }, + flush(controller) { + if (receivedSize !== expectedSize) { + controller.error(new UploadBodyError("UPLOAD_SIZE_MISMATCH")); + } + }, + }), + ); + const body = preserveKnownLength(checkedBody, expectedSize); + + const attemptKey = createUploadAttemptKey(media.storageKey); + await repo.createUploadAttempt(id, attemptKey); + + let attemptSize: number; + try { + const result = await emdash.storage.upload({ + key: attemptKey, + body, + contentType: media.mimeType, + }); + attemptSize = result.size; + } catch (error) { + await removeUploadAttempt(emdash.storage, repo, attemptKey); + const bodyError = findUploadBodyError(error); + if (bodyError?.code === "PAYLOAD_TOO_LARGE") { + return apiError("PAYLOAD_TOO_LARGE", "Upload exceeds the expected size", 413); + } + if (bodyError?.code === "UPLOAD_SIZE_MISMATCH") { + return apiError("UPLOAD_SIZE_MISMATCH", "Upload size does not match the media item", 400); + } + return handleError(error, "Upload failed", "UPLOAD_ERROR"); + } + + if (receivedSize !== expectedSize || attemptSize !== expectedSize) { + await removeUploadAttempt(emdash.storage, repo, attemptKey); + return apiError("UPLOAD_SIZE_MISMATCH", "Upload size does not match the media item", 400); + } + const contentHash = hashBytes ? await computeContentHash(hashBytes) : undefined; + + let published: boolean; + try { + published = await repo.publishPendingStorageKey( + id, + media.storageKey, + attemptKey, + contentHash, + ); + } catch (error) { + try { + const current = await repo.findById(id); + if ( + current?.storageKey === attemptKey && + (current.status === "pending" || current.status === "ready") && + current.size === expectedSize && + (await getStoredSize(emdash.storage, attemptKey)) === expectedSize + ) { + return apiSuccess({ uploaded: true, size: expectedSize }); + } + } catch (verificationError) { + console.error("[media] upload publication verification failed:", verificationError); + } + return handleError(error, "Upload failed", "UPLOAD_ERROR"); + } + + if (!published) { + const current = await repo.findById(id); + if ( + current && + (current.status === "pending" || current.status === "ready") && + current.size === expectedSize && + (current.storageKey === attemptKey || + expectedSize === 0 || + (contentHash !== undefined && current.contentHash === contentHash)) && + (await getStoredSize(emdash.storage, current.storageKey)) === expectedSize + ) { + if (current.storageKey !== attemptKey) { + await removeUploadAttempt(emdash.storage, repo, attemptKey); + } + return apiSuccess({ uploaded: true, size: expectedSize }); + } + await removeUploadAttempt(emdash.storage, repo, attemptKey); + return apiError("INVALID_STATE", "Media item is no longer pending", 400); + } + + await removeUploadAttempt(emdash.storage, repo, media.storageKey); + return apiSuccess({ uploaded: true, size: receivedSize }); + } catch (error) { + return handleError(error, "Upload failed", "UPLOAD_ERROR"); + } +}; diff --git a/packages/core/src/astro/routes/api/media/upload-url.ts b/packages/core/src/astro/routes/api/media/upload-url.ts index edfa95ff2e..9e35703279 100644 --- a/packages/core/src/astro/routes/api/media/upload-url.ts +++ b/packages/core/src/astro/routes/api/media/upload-url.ts @@ -39,6 +39,10 @@ interface ExistingMediaResponse { url: string; } +function isUnsupportedSignedUpload(error: unknown): boolean { + return error instanceof Error && "code" in error && error.code === "NOT_SUPPORTED"; +} + /** * Get a signed upload URL for direct-to-storage upload */ @@ -71,6 +75,7 @@ export const POST: APIRoute = async ({ request, locals }) => { } const body = await parseBody(request, mediaUploadUrlBody(maxSize)); if (isParseError(body)) return body; + const normalizedContentType = normalizeMime(body.contentType); // Validate content type (field-aware widening) const fieldAllowlist = body.fieldId @@ -85,9 +90,9 @@ export const POST: APIRoute = async ({ request, locals }) => { const repo = new MediaRepository(emdash.db); // Check for existing content with same hash (deduplication) - if (body.contentHash) { + if (body.contentHash && body.size > 0) { const existing = await repo.findByContentHash(body.contentHash); - if (existing) { + if (existing && existing.mimeType === normalizedContentType && existing.size === body.size) { const response: ExistingMediaResponse = { existing: true, mediaId: existing.id, @@ -103,49 +108,41 @@ export const POST: APIRoute = async ({ request, locals }) => { const ext = path.extname(body.filename) || ""; const storageKey = `${id}${ext}`; - // Create pending media record with content hash + let signedUrl: Awaited> | null; + try { + signedUrl = await emdash.storage.getSignedUploadUrl({ + key: storageKey, + contentType: body.contentType, + size: body.size, + expiresIn: 3600, + }); + } catch (error) { + if (!isUnsupportedSignedUpload(error)) throw error; + signedUrl = null; + } + const mediaItem = await repo.createPending({ filename: body.filename, - mimeType: normalizeMime(body.contentType), + mimeType: normalizedContentType, size: body.size, storageKey, - contentHash: body.contentHash, authorId: user?.id, }); - // Get signed upload URL from storage - const signedUrl = await emdash.storage.getSignedUploadUrl({ - key: storageKey, - contentType: body.contentType, - size: body.size, - expiresIn: 3600, // 1 hour - }); - const response: UploadUrlResponse = { - uploadUrl: signedUrl.url, - method: signedUrl.method, - headers: signedUrl.headers, + uploadUrl: signedUrl?.url ?? `/_emdash/api/media/${mediaItem.id}/upload`, + method: signedUrl?.method ?? "PUT", + headers: signedUrl?.headers ?? { + "Content-Type": normalizedContentType, + "X-EmDash-Request": "1", + }, mediaId: mediaItem.id, storageKey, - expiresAt: signedUrl.expiresAt, + expiresAt: signedUrl?.expiresAt ?? new Date(Date.now() + 3600 * 1000).toISOString(), }; return apiSuccess(response); } catch (error) { - // Check if storage doesn't support signed URLs (e.g., local storage) - if ( - error instanceof Error && - "code" in error && - // eslint-disable-next-line typescript/no-unsafe-type-assertion -- narrowing error to check custom code property after "code" in error guard - (error as { code: string }).code === "NOT_SUPPORTED" - ) { - return apiError( - "NOT_SUPPORTED", - "Storage does not support signed upload URLs. Use direct upload.", - 501, - ); - } - return handleError(error, "Failed to generate upload URL", "UPLOAD_URL_ERROR"); } }; diff --git a/packages/core/src/cleanup.ts b/packages/core/src/cleanup.ts index 3e61dc4b24..1dc295d204 100644 --- a/packages/core/src/cleanup.ts +++ b/packages/core/src/cleanup.ts @@ -16,6 +16,7 @@ import { cleanupExpiredChallenges } from "./auth/challenge-store.js"; import { MediaRepository } from "./database/repositories/media.js"; import { RevisionRepository } from "./database/repositories/revision.js"; import type { Database } from "./database/types.js"; +import { removeUploadAttempt } from "./media/upload-attempts.js"; import type { Storage } from "./storage/types.js"; /** @@ -27,6 +28,7 @@ export interface CleanupResult { expiredTokens: number; pendingUploads: number; pendingUploadFiles: number; + uploadAttempts: number; revisionsPruned: number; } @@ -56,6 +58,7 @@ export async function runSystemCleanup( expiredTokens: -1, pendingUploads: -1, pendingUploadFiles: -1, + uploadAttempts: -1, revisionsPruned: -1, }; @@ -106,7 +109,27 @@ export async function runSystemCleanup( console.error("[cleanup] Failed to clean pending uploads:", error); } - // 4. Revision pruning -- trim entries with excessive revision counts + // 4. Uploaded objects that lost publication races or outlived their media row + try { + const mediaRepo = new MediaRepository(db); + const completedAttemptsDeleted = await mediaRepo.deleteCompletedUploadAttempts(); + if (!storage) { + result.uploadAttempts = completedAttemptsDeleted; + } else { + const storageKeys = await mediaRepo.findUploadAttemptsForCleanup(); + let attemptsDeleted = completedAttemptsDeleted; + for (const storageKey of storageKeys) { + if (await removeUploadAttempt(storage, mediaRepo, storageKey)) { + attemptsDeleted++; + } + } + result.uploadAttempts = attemptsDeleted; + } + } catch (error) { + console.error("[cleanup] Failed to clean media upload attempts:", error); + } + + // 5. Revision pruning -- trim entries with excessive revision counts try { result.revisionsPruned = await pruneExcessiveRevisions(db); } catch (error) { diff --git a/packages/core/src/database/migrations/054_media_upload_attempts.ts b/packages/core/src/database/migrations/054_media_upload_attempts.ts new file mode 100644 index 0000000000..8e75a90a31 --- /dev/null +++ b/packages/core/src/database/migrations/054_media_upload_attempts.ts @@ -0,0 +1,41 @@ +import type { Kysely } from "kysely"; + +import { currentTimestamp } from "../dialect-helpers.js"; + +export async function up(db: Kysely): Promise { + await db.schema + .createTable("_emdash_media_upload_attempts") + .ifNotExists() + .addColumn("storage_key", "text", (col) => col.primaryKey()) + .addColumn("media_id", "text", (col) => col.notNull()) + .addColumn("status", "text", (col) => col.notNull().defaultTo("active")) + .addColumn("created_at", "text", (col) => col.notNull().defaultTo(currentTimestamp(db))) + .addColumn("updated_at", "text", (col) => col.notNull().defaultTo(currentTimestamp(db))) + .execute(); + + await db.schema + .createIndex("idx_media_upload_attempts_media_id") + .ifNotExists() + .on("_emdash_media_upload_attempts") + .column("media_id") + .execute(); + + await db.schema + .createIndex("idx_media_upload_attempts_status_created_at") + .ifNotExists() + .on("_emdash_media_upload_attempts") + .columns(["status", "created_at"]) + .execute(); + + await db.schema + .createIndex("idx_media_storage_key") + .ifNotExists() + .on("media") + .column("storage_key") + .execute(); +} + +export async function down(db: Kysely): Promise { + await db.schema.dropIndex("idx_media_storage_key").ifExists().execute(); + await db.schema.dropTable("_emdash_media_upload_attempts").ifExists().execute(); +} diff --git a/packages/core/src/database/migrations/runner.ts b/packages/core/src/database/migrations/runner.ts index a3ce955d45..5c1c7764ff 100644 --- a/packages/core/src/database/migrations/runner.ts +++ b/packages/core/src/database/migrations/runner.ts @@ -56,6 +56,7 @@ import * as m050 from "./050_media_usage_index_status.js"; import * as m051 from "./051_content_taxonomies_denorm.js"; import * as m052 from "./052_media_usage_read_index.js"; import * as m053 from "./053_plugin_mcp_tools.js"; +import * as m054 from "./054_media_upload_attempts.js"; const MIGRATIONS: Readonly> = Object.freeze({ "001_initial": m001, @@ -110,6 +111,7 @@ const MIGRATIONS: Readonly> = Object.freeze({ "051_content_taxonomies_denorm": m051, "052_media_usage_read_index": m052, "053_plugin_mcp_tools": m053, + "054_media_upload_attempts": m054, }); /** Total number of registered migrations. Exported for use in tests. */ diff --git a/packages/core/src/database/repositories/media.ts b/packages/core/src/database/repositories/media.ts index 035e61f25f..5d6335ff1d 100644 --- a/packages/core/src/database/repositories/media.ts +++ b/packages/core/src/database/repositories/media.ts @@ -85,6 +85,9 @@ export interface FindManyMediaOptions { q?: string; } +const UPLOAD_ATTEMPT_CLEANUP_AGE_MS = 60 * 60 * 1000; +const UPLOAD_ATTEMPT_CLEANUP_BATCH_SIZE = 100; + /** * Media repository for database operations */ @@ -138,6 +141,129 @@ export class MediaRepository { }); } + async createUploadAttempt(mediaId: string, storageKey: string): Promise { + const now = new Date().toISOString(); + await this.db + .insertInto("_emdash_media_upload_attempts") + .values({ + media_id: mediaId, + storage_key: storageKey, + status: "active", + created_at: now, + updated_at: now, + }) + .execute(); + } + + async hasUploadAttempt(storageKey: string): Promise { + const row = await this.db + .selectFrom("_emdash_media_upload_attempts") + .select("storage_key") + .where("storage_key", "=", storageKey) + .executeTakeFirst(); + return row !== undefined; + } + + async claimUploadAttemptForCleanup(storageKey: string): Promise { + const result = await this.db + .updateTable("_emdash_media_upload_attempts") + .set({ status: "cleanup", updated_at: new Date().toISOString() }) + .where("storage_key", "=", storageKey) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("media") + .select("media.id") + .whereRef("media.storage_key", "=", "_emdash_media_upload_attempts.storage_key"), + ), + ), + ) + .executeTakeFirst(); + + return Number(result.numUpdatedRows ?? 0) > 0; + } + + async deleteUploadAttempt(storageKey: string): Promise { + await this.db + .deleteFrom("_emdash_media_upload_attempts") + .where("storage_key", "=", storageKey) + .execute(); + } + + async deleteCompletedUploadAttempts(): Promise { + const result = await this.db + .deleteFrom("_emdash_media_upload_attempts") + .where((eb) => + eb.exists( + eb + .selectFrom("media") + .select("media.id") + .whereRef("media.id", "=", "_emdash_media_upload_attempts.media_id") + .whereRef("media.storage_key", "=", "_emdash_media_upload_attempts.storage_key") + .where("media.status", "=", "ready"), + ), + ) + .executeTakeFirst(); + return Number(result.numDeletedRows ?? 0); + } + + async findUploadAttemptsForCleanup( + maxAgeMs: number = UPLOAD_ATTEMPT_CLEANUP_AGE_MS, + limit: number = UPLOAD_ATTEMPT_CLEANUP_BATCH_SIZE, + ): Promise { + const cutoff = new Date(Date.now() - maxAgeMs).toISOString(); + const rows = await this.db + .selectFrom("_emdash_media_upload_attempts") + .select("storage_key") + .where((eb) => eb.or([eb("status", "=", "cleanup"), eb("created_at", "<", cutoff)])) + .where((eb) => + eb.not( + eb.exists( + eb + .selectFrom("media") + .select("media.id") + .whereRef("media.storage_key", "=", "_emdash_media_upload_attempts.storage_key"), + ), + ), + ) + .orderBy("created_at", "asc") + .limit(limit) + .execute(); + + return rows.map((row) => row.storage_key); + } + + async publishPendingStorageKey( + id: string, + expectedStorageKey: string, + storageKey: string, + contentHash?: string, + ): Promise { + const result = await this.db + .updateTable("media") + .set({ + storage_key: storageKey, + ...(contentHash !== undefined ? { content_hash: contentHash } : {}), + }) + .where("id", "=", id) + .where("status", "=", "pending") + .where("storage_key", "=", expectedStorageKey) + .where((eb) => + eb.exists( + eb + .selectFrom("_emdash_media_upload_attempts") + .select("storage_key") + .where("media_id", "=", id) + .where("storage_key", "=", storageKey) + .where("status", "=", "active"), + ), + ) + .executeTakeFirst(); + + return Number(result.numUpdatedRows ?? 0) > 0; + } + /** * Confirm upload (mark as ready) */ @@ -149,13 +275,10 @@ export class MediaRepository { size?: number; blurhash?: string; dominantColor?: string; + contentHash?: string | null; }, + expectedStorageKey?: string, ): Promise { - const existing = await this.findById(id); - if (!existing) { - return null; - } - const updates: Partial = { status: "ready", }; @@ -164,24 +287,33 @@ export class MediaRepository { if (metadata?.size !== undefined) updates.size = metadata.size; if (metadata?.blurhash !== undefined) updates.blurhash = metadata.blurhash; if (metadata?.dominantColor !== undefined) updates.dominant_color = metadata.dominantColor; + if (metadata?.contentHash !== undefined) updates.content_hash = metadata.contentHash; - await this.db.updateTable("media").set(updates).where("id", "=", id).execute(); + let query = this.db + .updateTable("media") + .set(updates) + .where("id", "=", id) + .where("status", "=", "pending"); + if (expectedStorageKey !== undefined) { + query = query.where("storage_key", "=", expectedStorageKey); + } - return this.findById(id); + const row = await query.returningAll().executeTakeFirst(); + + return row ? this.rowToItem(row) : null; } /** * Mark upload as failed */ - async markFailed(id: string): Promise { - const existing = await this.findById(id); - if (!existing) { - return null; + async markFailed(id: string, expectedStorageKey?: string): Promise { + let query = this.db.updateTable("media").set({ status: "failed" }).where("id", "=", id); + if (expectedStorageKey !== undefined) { + query = query.where("status", "=", "pending").where("storage_key", "=", expectedStorageKey); } - await this.db.updateTable("media").set({ status: "failed" }).where("id", "=", id).execute(); - - return this.findById(id); + const row = await query.returningAll().executeTakeFirst(); + return row ? this.rowToItem(row) : null; } /** @@ -319,10 +451,18 @@ export class MediaRepository { /** * Delete media item */ - async delete(id: string): Promise { - const result = await this.db.deleteFrom("media").where("id", "=", id).executeTakeFirst(); + async deleteWithStorageKey(id: string): Promise { + const deleted = await this.db + .deleteFrom("media") + .where("id", "=", id) + .returning("storage_key") + .executeTakeFirst(); + if (deleted) return deleted.storage_key; + return null; + } - return (result.numDeletedRows ?? 0) > 0; + async delete(id: string): Promise { + return (await this.deleteWithStorageKey(id)) !== null; } /** @@ -350,21 +490,11 @@ export class MediaRepository { async cleanupPendingUploads(maxAgeMs: number = 60 * 60 * 1000): Promise { const cutoff = new Date(Date.now() - maxAgeMs).toISOString(); - // Select the storage keys first -- SQLite doesn't support RETURNING - // on DELETE in all drivers, and Kysely's RETURNING isn't universal. const rows = await this.db - .selectFrom("media") - .select("storage_key") - .where("status", "=", "pending") - .where("created_at", "<", cutoff) - .execute(); - - if (rows.length === 0) return []; - - await this.db .deleteFrom("media") .where("status", "=", "pending") .where("created_at", "<", cutoff) + .returning("storage_key") .execute(); return rows.map((r) => r.storage_key); diff --git a/packages/core/src/database/types.ts b/packages/core/src/database/types.ts index 0f4ba13ed7..0b00056fc9 100644 --- a/packages/core/src/database/types.ts +++ b/packages/core/src/database/types.ts @@ -81,6 +81,15 @@ export interface MediaTable { author_id: string | null; } +export interface MediaUploadAttemptTable { + storage_key: string; + // No foreign key: this row must survive media deletion until storage cleanup succeeds. + media_id: string; + status: string; // 'active' | 'cleanup' + created_at: Generated; + updated_at: Generated; +} + export interface MediaUsageSourceTable { source_key: string; source_type: string; @@ -495,6 +504,7 @@ export interface Database { content_taxonomies: ContentTaxonomyTable; _emdash_taxonomy_defs: TaxonomyDefTable; media: MediaTable; + _emdash_media_upload_attempts: MediaUploadAttemptTable; _emdash_media_usage_sources: MediaUsageSourceTable; _emdash_media_usage: MediaUsageTable; _emdash_media_usage_index_status: MediaUsageIndexStatusTable; diff --git a/packages/core/src/media/upload-attempts.ts b/packages/core/src/media/upload-attempts.ts new file mode 100644 index 0000000000..cdb5825fd8 --- /dev/null +++ b/packages/core/src/media/upload-attempts.ts @@ -0,0 +1,33 @@ +import type { MediaRepository } from "../database/repositories/media.js"; +import type { Storage } from "../storage/types.js"; + +export async function removeUploadAttempt( + storage: Storage, + repo: MediaRepository, + storageKey: string, + options: { allowUntracked?: boolean } = {}, +): Promise { + try { + if (!(await repo.claimUploadAttemptForCleanup(storageKey))) { + const tracked = await repo.hasUploadAttempt(storageKey); + if (tracked || !options.allowUntracked) return false; + } + } catch (error) { + console.error("[media] upload cleanup claim failed:", error); + return false; + } + + try { + await storage.delete(storageKey); + } catch (error) { + console.error("[media] upload cleanup failed:", error); + return false; + } + + try { + await repo.deleteUploadAttempt(storageKey); + } catch (error) { + console.error("[media] upload cleanup record deletion failed:", error); + } + return true; +} diff --git a/packages/core/src/storage/s3.ts b/packages/core/src/storage/s3.ts index 99ca870b16..b0a95e6984 100644 --- a/packages/core/src/storage/s3.ts +++ b/packages/core/src/storage/s3.ts @@ -306,7 +306,7 @@ export class S3Storage implements Storage { method: "PUT", headers: { "Content-Type": options.contentType, - ...(options.size ? { "Content-Length": String(options.size) } : {}), + ...(options.size !== undefined ? { "Content-Length": String(options.size) } : {}), }, expiresAt, }; diff --git a/packages/core/src/utils/hash.ts b/packages/core/src/utils/hash.ts index 67e3a546c3..93db9df4a2 100644 --- a/packages/core/src/utils/hash.ts +++ b/packages/core/src/utils/hash.ts @@ -1,3 +1,10 @@ +export const MAX_CONTENT_HASH_BYTES = 8 * 1024 * 1024; + +function formatContentHash(hash: Uint8Array): string { + const hashHex = Array.from(hash, (byte) => byte.toString(16).padStart(2, "0")).join(""); + return `sha1:${hashHex}`; +} + /** * SHA-256 hash of a string, truncated to 16 hex chars (64 bits). * For cache invalidation / ETags — not for security. @@ -25,12 +32,16 @@ export async function computeContentHash(content: Uint8Array | ArrayBuffer): Pro let buf: ArrayBuffer; if (content instanceof ArrayBuffer) { buf = content; + } else if ( + content.buffer instanceof ArrayBuffer && + content.byteOffset === 0 && + content.byteLength === content.buffer.byteLength + ) { + buf = content.buffer; } else { buf = new ArrayBuffer(content.byteLength); new Uint8Array(buf).set(content); } const hashBuffer = await crypto.subtle.digest("SHA-1", buf); - const hashArray = new Uint8Array(hashBuffer); - const hashHex = Array.from(hashArray, (b) => b.toString(16).padStart(2, "0")).join(""); - return `sha1:${hashHex}`; + return formatContentHash(new Uint8Array(hashBuffer)); } diff --git a/packages/core/tests/integration/astro/media-confirm-placeholder.test.ts b/packages/core/tests/integration/astro/media-confirm-placeholder.test.ts index bcfc9d345c..618e6a8385 100644 --- a/packages/core/tests/integration/astro/media-confirm-placeholder.test.ts +++ b/packages/core/tests/integration/astro/media-confirm-placeholder.test.ts @@ -5,6 +5,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { POST as postConfirm } from "../../../src/astro/routes/api/media/[id]/confirm.js"; import { MediaRepository } from "../../../src/database/repositories/media.js"; import type { Database } from "../../../src/database/types.js"; +import { computeContentHash } from "../../../src/utils/hash.js"; import { JPEG_4x4 } from "../../utils/image-fixtures.js"; import { setupTestDatabase, teardownTestDatabase } from "../../utils/test-db.js"; @@ -25,11 +26,11 @@ function storageWith(bytes: Uint8Array) { } /** Storage stub whose download is spyable (to assert read-back never happens). */ -function spyableStorage(bytes: Uint8Array) { +function spyableStorage(bytes: Uint8Array, reportedSize = bytes.byteLength) { const download = vi.fn(async () => ({ body: new Response(bytes).body as ReadableStream, contentType: "image/jpeg", - size: bytes.byteLength, + size: reportedSize, })); return { exists: vi.fn(async () => true), @@ -42,6 +43,7 @@ function buildContext(opts: { id: string; storage: unknown; body: Record; + role?: 20 | 50; }): APIContext { const request = new Request(`http://localhost/_emdash/api/media/${opts.id}/confirm`, { method: "POST", @@ -54,7 +56,7 @@ function buildContext(opts: { request, locals: { emdash: { db: opts.db, storage: opts.storage }, - user: { id: "user-1", email: "t@example.com", name: "T", role: 50 as const }, + user: { id: "user-1", email: "t@example.com", name: "T", role: opts.role ?? 50 }, }, // eslint-disable-next-line typescript/no-unsafe-type-assertion -- minimal stub for tests } as unknown as APIContext; @@ -68,6 +70,7 @@ describe("POST /media/:id/confirm — placeholder read-back", () => { }); afterEach(async () => { + vi.restoreAllMocks(); await teardownTestDatabase(db); }); @@ -95,17 +98,162 @@ describe("POST /media/:id/confirm — placeholder read-back", () => { expect(row?.width).toBe(4); expect(row?.blurhash).toBeTruthy(); expect(row?.dominantColor).toMatch(/^rgb\(/); + expect(row?.contentHash).toBe(await computeContentHash(JPEG_4x4)); + }); + + it("allows a contributing uploader to confirm their own pending file", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.pdf", + authorId: "user-1", + }); + + const res = await postConfirm( + buildContext({ + db, + id: pending.id, + storage: { + async exists() { + return true; + }, + async download() { + return { + body: new Response(new Uint8Array([1, 2, 3])).body as ReadableStream, + contentType: "application/pdf", + size: 3, + }; + }, + }, + body: { size: 3 }, + role: 20, + }), + ); + + expect(res.status).toBe(200); + expect(await repo.findById(pending.id)).toMatchObject({ status: "ready" }); + }); + + it("cancels a stored download whose bytes exceed its reported size", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 1, + storageKey: "document.pdf", + authorId: "user-1", + }); + const cancel = vi.fn(); + const storage = { + async exists() { + return true; + }, + async download() { + return { + body: new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array([1, 2])); + }, + cancel, + }), + contentType: "application/pdf", + size: 1, + }; + }, + }; + + const res = await postConfirm(buildContext({ db, id: pending.id, storage, body: { size: 1 } })); + + expect(res.status).toBe(500); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it("does not re-read a proxied non-image after the server hashed it", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.attempt.pdf", + contentHash: "sha1:7037807198c22a7d2b0807371d763779a84fdfcf", + authorId: "user-1", + }); + await repo.createUploadAttempt(pending.id, pending.storageKey); + let bodyRead = false; + const cancel = vi.fn(async () => undefined); + const storage = { + async exists() { + return true; + }, + async download() { + return { + body: { + getReader() { + bodyRead = true; + throw new Error("Body should not be read"); + }, + cancel, + }, + contentType: "application/pdf", + size: 3, + }; + }, + }; + + const res = await postConfirm( + buildContext({ db, id: pending.id, storage, body: { size: 3 }, role: 20 }), + ); + + expect(res.status).toBe(200); + expect(bodyRead).toBe(false); + expect(cancel).toHaveBeenCalledOnce(); + }); + + it("rejects a client size that disagrees with the pending upload", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.jpg", + mimeType: "image/jpeg", + size: JPEG_4x4.byteLength, + storageKey: "photo.jpg", + authorId: "user-1", + }); + const storage = spyableStorage(JPEG_4x4); + + const res = await postConfirm( + buildContext({ + db, + id: pending.id, + storage, + body: { size: 1 }, + role: 20, + }), + ); + + expect(res.status).toBe(400); + expect(storage.download).not.toHaveBeenCalled(); + expect(await repo.findById(pending.id)).toMatchObject({ + status: "pending", + size: JPEG_4x4.byteLength, + }); }); it("skips placeholder read-back for oversized images (OOM guard) but still confirms", async () => { const repo = new MediaRepository(db); + const reportedSize = 64 * 1024 * 1024; const pending = await repo.createPending({ filename: "huge.jpg", mimeType: "image/jpeg", + size: reportedSize, storageKey: "huge.jpg", + contentHash: "sha1:a9993e364706816aba3e25717850c26c9cd0d89d", authorId: "user-1", }); - const storage = spyableStorage(JPEG_4x4); + await repo.createUploadAttempt(pending.id, pending.storageKey); + const storage = spyableStorage(JPEG_4x4, reportedSize); // Confirm claims a size far above the download cap. The signed-URL flow // exists so large files bypass server buffering; confirm must not re-read @@ -115,12 +263,12 @@ describe("POST /media/:id/confirm — placeholder read-back", () => { db, id: pending.id, storage, - body: { size: 64 * 1024 * 1024, width: 4000, height: 3000 }, + body: { size: reportedSize, width: 4000, height: 3000 }, }), ); expect(res.status).toBe(200); - expect(storage.download).not.toHaveBeenCalled(); + expect(storage.download).toHaveBeenCalledOnce(); const row = await repo.findById(pending.id); expect(row?.status).toBe("ready"); // Client-supplied dimensions are still recorded even when LQIP is skipped. @@ -129,4 +277,111 @@ describe("POST /media/:id/confirm — placeholder read-back", () => { expect(row?.blurhash).toBeNull(); expect(row?.dominantColor).toBeNull(); }); + + it("does not read back an oversized signed upload to compute a hash", async () => { + const repo = new MediaRepository(db); + const reportedSize = 64 * 1024 * 1024; + const pending = await repo.createPending({ + filename: "signed-huge.jpg", + mimeType: "image/jpeg", + size: reportedSize, + storageKey: "signed-huge.jpg", + authorId: "user-1", + }); + const storage = spyableStorage(JPEG_4x4, reportedSize); + + const res = await postConfirm( + buildContext({ + db, + id: pending.id, + storage, + body: { size: reportedSize, width: 4000, height: 3000 }, + }), + ); + + expect(res.status).toBe(200); + expect(await repo.findById(pending.id)).toMatchObject({ + status: "ready", + contentHash: null, + }); + }); + + it("reports invalid state when another request resolves the pending upload first", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.pdf", + authorId: "user-1", + }); + + const res = await postConfirm( + buildContext({ + db, + id: pending.id, + storage: { + async exists() { + return true; + }, + async download() { + await db + .updateTable("media") + .set({ status: "failed" }) + .where("id", "=", pending.id) + .execute(); + return { + body: new Response(new Uint8Array([1, 2, 3])).body as ReadableStream, + contentType: "application/pdf", + size: 3, + }; + }, + }, + body: { size: 3 }, + }), + ); + + expect(res.status).toBe(400); + await expect(res.json()).resolves.toMatchObject({ + error: { code: "INVALID_STATE" }, + }); + expect(await repo.findById(pending.id)).toMatchObject({ status: "failed" }); + }); + + it("reports not found when pending cleanup deletes the row first", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.pdf", + authorId: "user-1", + }); + + const res = await postConfirm( + buildContext({ + db, + id: pending.id, + storage: { + async exists() { + return true; + }, + async download() { + await db.deleteFrom("media").where("id", "=", pending.id).execute(); + return { + body: new Response(new Uint8Array([1, 2, 3])).body as ReadableStream, + contentType: "application/pdf", + size: 3, + }; + }, + }, + body: { size: 3 }, + }), + ); + + expect(res.status).toBe(404); + await expect(res.json()).resolves.toMatchObject({ + error: { code: "NOT_FOUND" }, + }); + }); }); diff --git a/packages/core/tests/integration/astro/media-stream-upload.test.ts b/packages/core/tests/integration/astro/media-stream-upload.test.ts new file mode 100644 index 0000000000..8156690480 --- /dev/null +++ b/packages/core/tests/integration/astro/media-stream-upload.test.ts @@ -0,0 +1,1155 @@ +import type { APIContext } from "astro"; +import { sql, type Kysely } from "kysely"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { handleMediaDelete, handleMediaGet } from "../../../src/api/handlers/media.js"; +import { DELETE as deleteMedia } from "../../../src/astro/routes/api/media/[id].js"; +import { POST as postConfirm } from "../../../src/astro/routes/api/media/[id]/confirm.js"; +import { PUT as putUpload } from "../../../src/astro/routes/api/media/[id]/upload.js"; +import { POST as postUploadUrl } from "../../../src/astro/routes/api/media/upload-url.js"; +import { runSystemCleanup } from "../../../src/cleanup.js"; +import { MediaRepository } from "../../../src/database/repositories/media.js"; +import type { Database } from "../../../src/database/types.js"; +import { EmDashStorageError } from "../../../src/storage/types.js"; +import { computeContentHash } from "../../../src/utils/hash.js"; +import { setupTestDatabase, teardownTestDatabase } from "../../utils/test-db.js"; + +function buildContext(options: { + db: Kysely; + request: Request; + storage: unknown; + id?: string; + user?: { id: string; role: 20 | 30 | 40 | 50 }; +}): APIContext { + return { + params: options.id ? { id: options.id } : {}, + url: new URL(options.request.url), + request: options.request, + locals: { + emdash: { db: options.db, config: {}, storage: options.storage }, + user: { + id: options.user?.id ?? "user-1", + email: "test@example.com", + name: "Test User", + role: options.user?.role ?? 30, + }, + }, + } as unknown as APIContext; +} + +function uploadUrlRequest() { + return new Request("http://localhost/_emdash/api/media/upload-url", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ filename: "photo.png", contentType: "image/png", size: 3 }), + }); +} + +function uploadUrlRequestWithHash(contentHash: string) { + return new Request("http://localhost/_emdash/api/media/upload-url", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ + filename: "photo.png", + contentType: "image/png", + size: 3, + contentHash, + }), + }); +} + +function uploadRequest(id: string, bytes: Uint8Array, contentType = "image/png") { + return new Request(`http://localhost/_emdash/api/media/${id}/upload`, { + method: "PUT", + headers: { "Content-Type": contentType, "X-EmDash-Request": "1" }, + body: bytes, + }); +} + +function unsupportedSignedUrlStorage() { + return { + async getSignedUploadUrl() { + throw new EmDashStorageError("Signed URLs unavailable", "NOT_SUPPORTED"); + }, + }; +} + +function streamingStorage() { + const objects = new Map(); + const upload = vi.fn( + async (options: { key: string; body: ReadableStream; contentType: string }) => { + const bytes = new Uint8Array(await new Response(options.body).arrayBuffer()); + objects.set(options.key, bytes); + return { key: options.key, url: `/media/${options.key}`, size: bytes.byteLength }; + }, + ); + const deleteObject = vi.fn(async (key: string) => { + objects.delete(key); + }); + const exists = vi.fn(async (key: string) => objects.has(key)); + const download = vi.fn(async (key: string) => { + const bytes = objects.get(key); + if (!bytes) throw new EmDashStorageError("File not found", "NOT_FOUND"); + return { + body: new Response(bytes).body as ReadableStream, + contentType: "image/png", + size: bytes.byteLength, + }; + }); + return { objects, upload, delete: deleteObject, exists, download }; +} + +describe("streamed media upload fallback", () => { + let db: Kysely; + + beforeEach(async () => { + db = await setupTestDatabase(); + }); + + afterEach(async () => { + vi.useRealTimers(); + vi.restoreAllMocks(); + vi.unstubAllGlobals(); + await teardownTestDatabase(db); + }); + + it("returns a same-origin upload target when signed URLs are unsupported", async () => { + const response = await postUploadUrl( + buildContext({ + db, + request: uploadUrlRequest(), + storage: unsupportedSignedUrlStorage(), + }), + ); + + expect(response.status).toBe(200); + const body = (await response.json()) as { + data: { uploadUrl: string; headers: Record; mediaId: string }; + }; + expect(body.data.uploadUrl).toBe(`/_emdash/api/media/${body.data.mediaId}/upload`); + expect(body.data.headers).toMatchObject({ + "Content-Type": "image/png", + "X-EmDash-Request": "1", + }); + + const pending = await new MediaRepository(db).findById(body.data.mediaId); + expect(pending).toMatchObject({ status: "pending", size: 3, authorId: "user-1" }); + }); + + it("does not create a pending row when signed URL generation fails", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + const response = await postUploadUrl( + buildContext({ + db, + request: uploadUrlRequest(), + storage: { + async getSignedUploadUrl() { + throw new EmDashStorageError("Storage unavailable", "UPLOAD_FAILED"); + }, + }, + }), + ); + + expect(response.status).toBe(500); + expect(await new MediaRepository(db).findMany({ status: "all" })).toMatchObject({ items: [] }); + }); + + it("uses a client content hash only as a deduplication probe", async () => { + const response = await postUploadUrl( + buildContext({ + db, + request: uploadUrlRequestWithHash("sha1:a9993e364706816aba3e25717850c26c9cd0d89d"), + storage: unsupportedSignedUrlStorage(), + }), + ); + + expect(response.status).toBe(200); + const body = (await response.json()) as { data: { mediaId: string } }; + expect(await new MediaRepository(db).findById(body.data.mediaId)).toMatchObject({ + contentHash: null, + }); + }); + + it("accepts existing client hash formats for deduplication", async () => { + const repo = new MediaRepository(db); + const contentHash = "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"; + const existing = await repo.create({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + contentHash, + }); + + const response = await postUploadUrl( + buildContext({ + db, + request: uploadUrlRequestWithHash(contentHash), + storage: unsupportedSignedUrlStorage(), + }), + ); + + expect(response.status).toBe(200); + const body = (await response.json()) as { data: { existing?: boolean; mediaId: string } }; + expect(body.data).toMatchObject({ existing: true, mediaId: existing.id }); + }); + + it.each([ + { difference: "MIME type", contentType: "video/mp4", size: 3 }, + { difference: "size", contentType: "audio/mp4", size: 4 }, + ])( + "does not deduplicate to media with a different $difference", + async ({ contentType, size }) => { + const repo = new MediaRepository(db); + const contentHash = "sha1:a9993e364706816aba3e25717850c26c9cd0d89d"; + const existing = await repo.create({ + filename: "sound.mp4", + mimeType: "audio/mp4", + size: 3, + storageKey: "sound.mp4", + contentHash, + }); + const request = new Request("http://localhost/_emdash/api/media/upload-url", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ + filename: "movie.mp4", + contentType, + size, + contentHash, + }), + }); + + const response = await postUploadUrl( + buildContext({ db, request, storage: unsupportedSignedUrlStorage() }), + ); + + expect(response.status).toBe(200); + const body = (await response.json()) as { + data: { existing?: boolean; mediaId: string }; + }; + expect(body.data.existing).not.toBe(true); + expect(body.data.mediaId).not.toBe(existing.id); + }, + ); + + it("does not deduplicate an empty file by the shared empty hash", async () => { + const repo = new MediaRepository(db); + const existing = await repo.create({ + filename: "legacy-empty.txt", + mimeType: "text/plain", + size: 0, + storageKey: "legacy-empty.txt", + contentHash: "sha1:da39a3ee5e6b4b0d3255bfef95601890afd80709", + }); + const request = new Request("http://localhost/_emdash/api/media/upload-url", { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ + filename: "empty.pdf", + contentType: "application/pdf", + size: 0, + contentHash: "sha1:da39a3ee5e6b4b0d3255bfef95601890afd80709", + }), + }); + + const response = await postUploadUrl( + buildContext({ db, request, storage: unsupportedSignedUrlStorage() }), + ); + + expect(response.status).toBe(200); + const body = (await response.json()) as { data: { mediaId: string } }; + expect(body.data.mediaId).not.toBe(existing.id); + expect(await repo.findById(body.data.mediaId)).toMatchObject({ + filename: "empty.pdf", + contentHash: null, + }); + }); + + it("streams the exact request body to storage and leaves confirmation separate", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + user: { id: "user-1", role: 20 }, + }), + ); + + expect(response.status).toBe(200); + expect(storage.upload).toHaveBeenCalledOnce(); + const uploadedBody = storage.upload.mock.calls[0]?.[0].body; + expect(uploadedBody).toBeInstanceOf(ReadableStream); + const uploaded = await repo.findById(pending.id); + expect(uploaded).toMatchObject({ + status: "pending", + contentHash: "sha1:7037807198c22a7d2b0807371d763779a84fdfcf", + }); + expect(storage.objects.get(uploaded!.storageKey)).toEqual(new Uint8Array([1, 2, 3])); + }); + + it("hashes a fragmented upload across buffer growth", async () => { + const size = 64 * 1024 + 1; + const bytes = new Uint8Array(size); + bytes[size - 1] = 1; + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "fragmented.pdf", + mimeType: "application/pdf", + size, + storageKey: "fragmented.pdf", + authorId: "user-1", + }); + const storage = streamingStorage(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(bytes.subarray(0, size - 1)); + controller.enqueue(bytes.subarray(size - 1)); + controller.close(); + }, + }); + const request = new Request(`http://localhost/_emdash/api/media/${pending.id}/upload`, { + method: "PUT", + headers: { + "Content-Type": "application/pdf", + "Content-Length": String(size), + "X-EmDash-Request": "1", + }, + body, + duplex: "half", + } as RequestInit & { duplex: "half" }); + + const response = await putUpload( + buildContext({ db, id: pending.id, request, storage, user: { id: "user-1", role: 20 } }), + ); + + expect(response.status).toBe(200); + expect(await repo.findById(pending.id)).toMatchObject({ + contentHash: await computeContentHash(bytes), + }); + }); + + it("preserves the known body length for Worker storage bindings", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + const fixedLengths = new WeakMap, number>(); + class TestFixedLengthStream { + readonly readable: ReadableStream; + readonly writable: WritableStream; + + constructor(expectedLength: number | bigint) { + const stream = new TransformStream(); + this.readable = stream.readable; + this.writable = stream.writable; + fixedLengths.set(this.readable, Number(expectedLength)); + } + } + vi.stubGlobal("FixedLengthStream", TestFixedLengthStream); + storage.upload.mockImplementationOnce(async (options) => { + if (fixedLengths.get(options.body) !== 3) { + throw new TypeError("Provided readable stream must have a known length"); + } + const bytes = new Uint8Array(await new Response(options.body).arrayBuffer()); + storage.objects.set(options.key, bytes); + return { key: options.key, url: `/media/${options.key}`, size: bytes.byteLength }; + }); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + + expect(response.status).toBe(200); + expect(storage.upload).toHaveBeenCalledOnce(); + const uploaded = await repo.findById(pending.id); + expect(storage.objects.get(uploaded!.storageKey)).toEqual(new Uint8Array([1, 2, 3])); + }); + + it("reports a truncated fixed-length upload as a size mismatch", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + class TestFixedLengthStream { + readonly readable: ReadableStream; + readonly writable: WritableStream; + + constructor(expectedLength: number | bigint) { + let received = 0; + const stream = new TransformStream({ + transform(chunk, controller) { + received += chunk.byteLength; + controller.enqueue(chunk); + }, + flush(controller) { + if (received !== Number(expectedLength)) { + controller.error(new TypeError("Fixed-length stream ended early")); + } + }, + }); + this.readable = stream.readable; + this.writable = stream.writable; + } + } + vi.stubGlobal("FixedLengthStream", TestFixedLengthStream); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2])), + storage, + }), + ); + + expect(response.status).toBe(400); + await expect(response.json()).resolves.toMatchObject({ + error: { code: "UPLOAD_SIZE_MISMATCH" }, + }); + }); + + it("accepts an empty file", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "empty.pdf", + mimeType: "application/pdf", + size: 0, + storageKey: "empty.pdf", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const request = new Request(`http://localhost/_emdash/api/media/${pending.id}/upload`, { + method: "PUT", + headers: { + "Content-Type": "application/pdf", + "Content-Length": "0", + "X-EmDash-Request": "1", + }, + }); + const response = await putUpload(buildContext({ db, id: pending.id, request, storage })); + + expect(response.status).toBe(200); + const uploaded = await new MediaRepository(db).findById(pending.id); + expect(storage.objects.get(uploaded!.storageKey)).toEqual(new Uint8Array()); + expect(uploaded?.contentHash).toBeNull(); + }); + + it("does not hash large uploads on the Worker request path", async () => { + const size = 8 * 1024 * 1024 + 1; + const pending = await new MediaRepository(db).createPending({ + filename: "large.pdf", + mimeType: "application/pdf", + size, + storageKey: "large.pdf", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array(size), "application/pdf"), + storage, + }), + ); + + expect(response.status).toBe(200); + expect(await new MediaRepository(db).findById(pending.id)).toMatchObject({ + contentHash: null, + }); + }); + + it("rejects a mismatched MIME type without writing to storage", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3]), "image/jpeg"), + storage, + }), + ); + + expect(response.status).toBe(400); + expect(storage.upload).not.toHaveBeenCalled(); + }); + + it("rejects and removes an upload whose streamed byte count does not match", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2])), + storage, + }), + ); + + expect(response.status).toBe(400); + expect(storage.delete).toHaveBeenCalledOnce(); + expect(storage.objects.size).toBe(0); + }); + + it("aborts and cleans up a stream that exceeds the expected size", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 2, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + + expect(response.status).toBe(413); + expect(storage.delete).toHaveBeenCalledOnce(); + expect(storage.objects.size).toBe(0); + }); + + it("preserves a completed object when a retry upload fails", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const firstResponse = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + expect(firstResponse.status).toBe(200); + + const completed = await repo.findById(pending.id); + expect(completed).not.toBeNull(); + const completedKey = completed!.storageKey; + expect(storage.objects.get(completedKey)).toEqual(new Uint8Array([1, 2, 3])); + + storage.upload.mockRejectedValueOnce( + new EmDashStorageError("Storage unavailable", "UPLOAD_FAILED"), + ); + const retryResponse = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + + expect(retryResponse.status).toBe(500); + expect(storage.upload).toHaveBeenCalledTimes(2); + expect(await repo.findById(pending.id)).toMatchObject({ storageKey: completedKey }); + expect(storage.objects.get(completedKey)).toEqual(new Uint8Array([1, 2, 3])); + }); + + it("processes different same-size bytes when an upload target is retried", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const firstResponse = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + expect(firstResponse.status).toBe(200); + const firstUpload = await repo.findById(pending.id); + expect(firstUpload).not.toBeNull(); + + const secondResponse = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([4, 5, 6])), + storage, + }), + ); + + expect(secondResponse.status).toBe(200); + expect(storage.upload).toHaveBeenCalledTimes(2); + const retried = await repo.findById(pending.id); + expect(retried?.storageKey).not.toBe(firstUpload!.storageKey); + expect(storage.objects.get(retried!.storageKey)).toEqual(new Uint8Array([4, 5, 6])); + expect(storage.objects.has(firstUpload!.storageKey)).toBe(false); + expect(storage.delete).toHaveBeenCalledOnce(); + }); + + it("does not confirm stale metadata after a replacement upload", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.pdf", + authorId: "user-1", + }); + const storage = streamingStorage(); + const firstBytes = new Uint8Array([1, 2, 3]); + const replacementBytes = new Uint8Array([4, 5, 6]); + + await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, firstBytes, "application/pdf"), + storage, + }), + ); + + let markConfirmationReading: () => void = () => undefined; + const confirmationReading = new Promise((resolve) => { + markConfirmationReading = resolve; + }); + let finishConfirmation: () => void = () => undefined; + const confirmationCanFinish = new Promise((resolve) => { + finishConfirmation = resolve; + }); + storage.download.mockImplementationOnce(async (key: string) => { + const bytes = storage.objects.get(key); + if (!bytes) throw new EmDashStorageError("File not found", "NOT_FOUND"); + return { + body: new ReadableStream({ + cancel() { + markConfirmationReading(); + return confirmationCanFinish; + }, + }), + contentType: "application/pdf", + size: bytes.byteLength, + }; + }); + + const confirm = () => + postConfirm( + buildContext({ + db, + id: pending.id, + request: new Request(`http://localhost/_emdash/api/media/${pending.id}/confirm`, { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: "{}", + }), + storage, + }), + ); + const staleConfirmation = confirm(); + await confirmationReading; + + await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, replacementBytes, "application/pdf"), + storage, + }), + ); + finishConfirmation(); + + expect((await staleConfirmation).status).toBe(409); + const replacement = await repo.findById(pending.id); + expect(replacement).toMatchObject({ + status: "pending", + contentHash: await computeContentHash(replacementBytes), + }); + expect(storage.objects.get(replacement!.storageKey)).toEqual(replacementBytes); + + expect((await confirm()).status).toBe(200); + expect(await repo.findById(pending.id)).toMatchObject({ + status: "ready", + contentHash: await computeContentHash(replacementBytes), + }); + }); + + it("does not mark a replacement upload as failed from a stale storage check", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "document.pdf", + mimeType: "application/pdf", + size: 3, + storageKey: "document.pdf", + authorId: "user-1", + }); + const storage = streamingStorage(); + const replacementBytes = new Uint8Array([4, 5, 6]); + + let markStorageChecked: () => void = () => undefined; + const storageChecked = new Promise((resolve) => { + markStorageChecked = resolve; + }); + let finishStorageCheck: () => void = () => undefined; + const storageCheckCanFinish = new Promise((resolve) => { + finishStorageCheck = resolve; + }); + storage.exists.mockImplementationOnce(async () => { + markStorageChecked(); + await storageCheckCanFinish; + return false; + }); + + const confirm = () => + postConfirm( + buildContext({ + db, + id: pending.id, + request: new Request(`http://localhost/_emdash/api/media/${pending.id}/confirm`, { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: "{}", + }), + storage, + }), + ); + const staleConfirmation = confirm(); + await storageChecked; + + await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, replacementBytes, "application/pdf"), + storage, + }), + ); + finishStorageCheck(); + + expect((await staleConfirmation).status).toBe(409); + expect(await repo.findById(pending.id)).toMatchObject({ + status: "pending", + contentHash: await computeContentHash(replacementBytes), + }); + + expect((await confirm()).status).toBe(200); + expect(await repo.findById(pending.id)).toMatchObject({ status: "ready" }); + }); + + it("publishes only one object when two uploads race", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const [firstResponse, secondResponse] = await Promise.all([ + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ), + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ), + ]); + + expect(firstResponse.status).toBe(200); + expect(secondResponse.status).toBe(200); + expect(storage.upload).toHaveBeenCalledTimes(2); + expect(storage.delete).toHaveBeenCalledOnce(); + expect(storage.objects.size).toBe(1); + const published = await repo.findById(pending.id); + expect(storage.objects.get(published!.storageKey)).toEqual(new Uint8Array([1, 2, 3])); + }); + + it("rejects a losing concurrent upload when its content differs", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + let uploadsStarted = 0; + let releaseUploads: (() => void) | undefined; + const bothUploadsStarted = new Promise((resolve) => { + releaseUploads = resolve; + }); + storage.upload.mockImplementation(async (options) => { + const bytes = new Uint8Array(await new Response(options.body).arrayBuffer()); + storage.objects.set(options.key, bytes); + uploadsStarted++; + if (uploadsStarted === 2) releaseUploads?.(); + await bothUploadsStarted; + return { key: options.key, url: `/media/${options.key}`, size: bytes.byteLength }; + }); + + const [firstResponse, secondResponse] = await Promise.all([ + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ), + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([4, 5, 6])), + storage, + }), + ), + ]); + + const statuses = [firstResponse.status, secondResponse.status]; + expect(statuses.filter((status) => status === 200)).toHaveLength(1); + expect(statuses.filter((status) => status === 400)).toHaveLength(1); + expect(storage.delete).toHaveBeenCalledOnce(); + expect(storage.objects.size).toBe(1); + }); + + it("preserves an object when publication commits before reporting an error", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + const publish = MediaRepository.prototype.publishPendingStorageKey; + vi.spyOn(MediaRepository.prototype, "publishPendingStorageKey").mockImplementationOnce( + async function ( + this: MediaRepository, + id: string, + expectedStorageKey: string, + storageKey: string, + contentHash: string, + ) { + await publish.call(this, id, expectedStorageKey, storageKey, contentHash); + throw new Error("publication acknowledgement lost"); + }, + ); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + + expect(response.status).toBe(200); + const published = await repo.findById(pending.id); + expect(published?.storageKey).not.toBe(pending.storageKey); + expect(storage.objects.get(published!.storageKey)).toEqual(new Uint8Array([1, 2, 3])); + }); + + it("retries cleanup for an unreferenced upload attempt", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + storage.delete.mockRejectedValueOnce(new Error("temporary R2 failure")); + + const responses = await Promise.all([ + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ), + putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ), + ]); + + expect(responses.map((response) => response.status)).toEqual([200, 200]); + expect(storage.objects.size).toBe(2); + await repo.confirmUpload(pending.id); + + vi.useFakeTimers({ toFake: ["Date"] }); + vi.setSystemTime(new Date(Date.now() + 2 * 60 * 60 * 1000)); + await runSystemCleanup(db, storage); + + expect(storage.objects.size).toBe(1); + const published = await repo.findById(pending.id); + expect(storage.objects.has(published!.storageKey)).toBe(true); + }); + + it("preserves an object confirmed at the pending-cleanup delete boundary", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "expired.png", + mimeType: "image/png", + size: 3, + storageKey: "expired.png", + authorId: "user-1", + }); + await db + .updateTable("media") + .set({ created_at: new Date(0).toISOString() }) + .where("id", "=", pending.id) + .execute(); + + const storage = streamingStorage(); + storage.objects.set(pending.storageKey, new Uint8Array([1, 2, 3])); + + await sql` + CREATE TRIGGER confirm_during_pending_cleanup + BEFORE DELETE ON media + WHEN OLD.status = 'pending' + BEGIN + UPDATE media SET status = 'ready' WHERE id = OLD.id; + SELECT RAISE(IGNORE); + END + `.execute(db); + + const result = await runSystemCleanup(db, storage); + + expect(result.pendingUploads).toBe(0); + expect(result.pendingUploadFiles).toBe(0); + expect(await repo.findById(pending.id)).toMatchObject({ status: "ready" }); + expect(storage.objects.has(pending.storageKey)).toBe(true); + }); + + it("does not keep retrying when the database vetoes a media deletion", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "guarded.png", + mimeType: "image/png", + storageKey: "guarded.png", + }); + await sql`CREATE TABLE delete_guard (attempts integer NOT NULL)`.execute(db); + await sql`INSERT INTO delete_guard (attempts) VALUES (0)`.execute(db); + await sql` + CREATE TRIGGER veto_first_media_deletes + BEFORE DELETE ON media + WHEN (SELECT attempts FROM delete_guard) < 4 + BEGIN + UPDATE delete_guard SET attempts = attempts + 1; + SELECT RAISE(IGNORE); + END + `.execute(db); + + await expect(repo.deleteWithStorageKey(pending.id)).resolves.toBeNull(); + expect(await repo.findById(pending.id)).not.toBeNull(); + }); + + it("removes the upload attempt after confirmation and allows confirmation retries", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.attempt.png", + authorId: "user-1", + }); + await repo.createUploadAttempt(pending.id, pending.storageKey); + const storage = streamingStorage(); + storage.objects.set(pending.storageKey, new Uint8Array([1, 2, 3])); + const confirm = () => + postConfirm( + buildContext({ + db, + id: pending.id, + request: new Request(`http://localhost/_emdash/api/media/${pending.id}/confirm`, { + method: "POST", + headers: { "Content-Type": "application/json", "X-EmDash-Request": "1" }, + body: JSON.stringify({ size: 3 }), + }), + storage, + }), + ); + + expect((await confirm()).status).toBe(200); + expect(await repo.hasUploadAttempt(pending.storageKey)).toBe(false); + expect((await confirm()).status).toBe(200); + }); + + it("reaps completed upload-attempt bookkeeping without deleting the live object", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.attempt.png", + }); + await repo.createUploadAttempt(pending.id, pending.storageKey); + await repo.confirmUpload(pending.id); + + const result = await runSystemCleanup(db); + + expect(result.uploadAttempts).toBe(1); + expect(await repo.hasUploadAttempt(pending.storageKey)).toBe(false); + expect(await repo.findById(pending.id)).toMatchObject({ status: "ready" }); + }); + + it("does not strand the published object when deletion races an upload", async () => { + const repo = new MediaRepository(db); + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + let startDelete: (() => void) | undefined; + const deleteStarted = new Promise((resolve) => { + startDelete = resolve; + }); + let finishDelete: (() => void) | undefined; + const allowDelete = new Promise((resolve) => { + finishDelete = resolve; + }); + + const deletion = deleteMedia({ + params: { id: pending.id }, + locals: { + emdash: { + db, + storage, + handleMediaGet: (id: string) => handleMediaGet(db, id), + handleMediaDelete: async (id: string) => { + startDelete?.(); + await allowDelete; + return handleMediaDelete(db, id); + }, + }, + user: { + id: "user-1", + email: "test@example.com", + name: "Test User", + role: 30, + }, + }, + } as unknown as APIContext); + await deleteStarted; + + const uploadResponse = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + }), + ); + finishDelete?.(); + const deleteResponse = await deletion; + + expect(uploadResponse.status).toBe(200); + expect(deleteResponse.status).toBe(200); + expect(await repo.findById(pending.id)).toBeNull(); + expect(storage.objects.size).toBe(0); + }); + + it("rejects a non-owner without media:edit_any", async () => { + const pending = await new MediaRepository(db).createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "photo.png", + authorId: "user-1", + }); + const storage = streamingStorage(); + + const response = await putUpload( + buildContext({ + db, + id: pending.id, + request: uploadRequest(pending.id, new Uint8Array([1, 2, 3])), + storage, + user: { id: "user-2", role: 30 }, + }), + ); + + expect(response.status).toBe(403); + expect(storage.upload).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/core/tests/integration/database/media-upload-publish.test.ts b/packages/core/tests/integration/database/media-upload-publish.test.ts new file mode 100644 index 0000000000..e1e34808fa --- /dev/null +++ b/packages/core/tests/integration/database/media-upload-publish.test.ts @@ -0,0 +1,147 @@ +import type { Kysely } from "kysely"; +import { afterEach, beforeEach, expect, it } from "vitest"; + +import { MediaRepository } from "../../../src/database/repositories/media.js"; +import type { Database } from "../../../src/database/types.js"; +import { + type DialectTestContext, + describeEachDialect, + setupForDialect, + teardownForDialect, +} from "../../utils/test-db.js"; + +describeEachDialect("pending media upload publication", (dialect) => { + let ctx: DialectTestContext; + let repo: MediaRepository; + + beforeEach(async () => { + ctx = await setupForDialect(dialect); + repo = new MediaRepository(ctx.db as Kysely); + }); + + afterEach(async () => { + await teardownForDialect(ctx); + }); + + it("allows only one upload attempt to publish its storage key", async () => { + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "pending.png", + }); + await Promise.all([ + repo.createUploadAttempt(pending.id, "attempt-a.png"), + repo.createUploadAttempt(pending.id, "attempt-b.png"), + ]); + + const results = await Promise.all([ + repo.publishPendingStorageKey( + pending.id, + "pending.png", + "attempt-a.png", + "sha1:a9993e364706816aba3e25717850c26c9cd0d89d", + ), + repo.publishPendingStorageKey( + pending.id, + "pending.png", + "attempt-b.png", + "sha1:a9993e364706816aba3e25717850c26c9cd0d89d", + ), + ]); + + expect(results).toContain(true); + expect(results).toContain(false); + expect((await repo.findById(pending.id))?.storageKey).toMatch(/^attempt-[ab]\.png$/); + }); + + it("does not apply stale confirmation state after the storage key changes", async () => { + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "pending.png", + }); + await repo.createUploadAttempt(pending.id, "replacement.png"); + await repo.publishPendingStorageKey( + pending.id, + pending.storageKey, + "replacement.png", + "sha1:589c22335a381f122d129225f5c0ba3056ed5811", + ); + + await expect( + repo.confirmUpload(pending.id, { width: 100 }, pending.storageKey), + ).resolves.toBeNull(); + await expect(repo.markFailed(pending.id, pending.storageKey)).resolves.toBeNull(); + expect(await repo.findById(pending.id)).toMatchObject({ + status: "pending", + storageKey: "replacement.png", + width: null, + }); + }); + + it("does not claim a fresh active attempt for cleanup", async () => { + const pending = await repo.createPending({ + filename: "photo.png", + mimeType: "image/png", + size: 3, + storageKey: "pending.png", + }); + await repo.createUploadAttempt(pending.id, "fresh-attempt.png"); + + expect(await repo.findUploadAttemptsForCleanup()).not.toContain("fresh-attempt.png"); + }); + + it("returns only storage keys deleted by pending cleanup", async () => { + const expired = await repo.createPending({ + filename: "expired.png", + mimeType: "image/png", + storageKey: "expired.png", + }); + const ready = await repo.create({ + filename: "ready.png", + mimeType: "image/png", + storageKey: "ready.png", + status: "ready", + }); + await ctx.db + .updateTable("media") + .set({ created_at: new Date(0).toISOString() }) + .where("id", "in", [expired.id, ready.id]) + .execute(); + + expect(await repo.cleanupPendingUploads()).toEqual([expired.storageKey]); + expect(await repo.findById(expired.id)).toBeNull(); + expect(await repo.findById(ready.id)).toMatchObject({ status: "ready" }); + }); + + it("keeps cleanup and confirmation outcomes consistent under concurrency", async () => { + for (let i = 0; i < 10; i++) { + const pending = await repo.createPending({ + filename: `race-${i}.png`, + mimeType: "image/png", + storageKey: `race-${i}.png`, + }); + await ctx.db + .updateTable("media") + .set({ created_at: new Date(0).toISOString() }) + .where("id", "=", pending.id) + .execute(); + + const [confirmed, deletedKeys] = await Promise.all([ + repo.confirmUpload(pending.id), + repo.cleanupPendingUploads(), + ]); + const stored = await repo.findById(pending.id); + + if (confirmed) { + expect(stored).toMatchObject({ status: "ready" }); + expect(deletedKeys).not.toContain(pending.storageKey); + } else { + expect(stored).toBeNull(); + expect(deletedKeys).toContain(pending.storageKey); + } + } + }); +}); diff --git a/packages/core/tests/integration/database/migrations.test.ts b/packages/core/tests/integration/database/migrations.test.ts index 6767b491d5..ce2e6de4c3 100644 --- a/packages/core/tests/integration/database/migrations.test.ts +++ b/packages/core/tests/integration/database/migrations.test.ts @@ -140,6 +140,7 @@ describe("Database Migrations (Integration)", () => { "051_content_taxonomies_denorm", "052_media_usage_read_index", "053_plugin_mcp_tools", + "054_media_upload_attempts", ]; await db.deleteFrom("_emdash_migrations").where("name", "in", trailing).execute(); diff --git a/packages/core/tests/unit/api/openapi.test.ts b/packages/core/tests/unit/api/openapi.test.ts index 61d4b3ce16..81a103500a 100644 --- a/packages/core/tests/unit/api/openapi.test.ts +++ b/packages/core/tests/unit/api/openapi.test.ts @@ -35,7 +35,9 @@ describe("OpenAPI document generation", () => { expect(paths).toContain("/_emdash/api/media/{id}/usage"); expect(paths).toContain("/_emdash/api/media/upload-url"); expect(paths).toContain("/_emdash/api/media/{id}/confirm"); + expect(paths).toContain("/_emdash/api/media/{id}/upload"); expect(paths).toContain("/_emdash/api/admin/media-usage/repair"); + expect(doc.paths?.["/_emdash/api/media/{id}/confirm"]?.post?.responses).toHaveProperty("409"); }); it("documents media usage summary opt-in parameters and read responses", () => { diff --git a/packages/core/tests/unit/api/schemas.test.ts b/packages/core/tests/unit/api/schemas.test.ts index 7346c699bf..5a8e995860 100644 --- a/packages/core/tests/unit/api/schemas.test.ts +++ b/packages/core/tests/unit/api/schemas.test.ts @@ -247,6 +247,24 @@ describe("mediaUploadUrlBody schema factory", () => { expect(result.size).toBe(500); }); + it("accepts an empty file", () => { + const schema = mediaUploadUrlBody(1_000); + const result = schema.parse({ filename: "empty.pdf", contentType: "application/pdf", size: 0 }); + expect(result.size).toBe(0); + }); + + it("accepts client content hashes in existing formats", () => { + const schema = mediaUploadUrlBody(1_000); + const contentHash = "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"; + const result = schema.parse({ + filename: "a.jpg", + contentType: "image/jpeg", + size: 500, + contentHash, + }); + expect(result.contentHash).toBe(contentHash); + }); + it("each call returns an independent schema with its own limit", () => { const strict = mediaUploadUrlBody(100); const loose = mediaUploadUrlBody(1_000_000); diff --git a/packages/core/tests/unit/astro/routes.test.ts b/packages/core/tests/unit/astro/routes.test.ts index dc632faa95..cb1bcc31c3 100644 --- a/packages/core/tests/unit/astro/routes.test.ts +++ b/packages/core/tests/unit/astro/routes.test.ts @@ -9,6 +9,7 @@ import { hasUserDefinedPublicRoute, injectCoreRoutes, } from "../../../src/astro/integration/routes.js"; +import * as mediaUploadRoute from "../../../src/astro/routes/api/media/[id]/upload.js"; import { GET as getMediaFile } from "../../../src/astro/routes/api/media/file/[...key].js"; function mockMediaContext(key: string | undefined) { @@ -73,6 +74,19 @@ describe("core media route injection", () => { ); }); + it("registers the pending-media upload route with PUT only", () => { + const routes: Array<{ pattern: string; entrypoint: string }> = []; + injectCoreRoutes((route) => routes.push(route)); + + expect(routes).toContainEqual( + expect.objectContaining({ pattern: "/_emdash/api/media/[id]/upload" }), + ); + expect(mediaUploadRoute.PUT).toBeTypeOf("function"); + for (const method of ["GET", "POST", "PATCH", "DELETE"]) { + expect(mediaUploadRoute).not.toHaveProperty(method); + } + }); + it("injects default root SEO routes when the site does not define them", () => { const routes = collectRoutePatterns(); diff --git a/packages/core/tests/unit/storage/s3.test.ts b/packages/core/tests/unit/storage/s3.test.ts index 5c83befeb2..8bbd9c9935 100644 --- a/packages/core/tests/unit/storage/s3.test.ts +++ b/packages/core/tests/unit/storage/s3.test.ts @@ -263,3 +263,22 @@ describe("resolveS3Config", () => { }); }); }); + +describe("S3Storage.getSignedUploadUrl", () => { + it("returns a zero Content-Length when zero is included in the signature", async () => { + const storage = createStorage({ + endpoint: "https://bucket.s3.example.com", + bucket: "my-bucket", + accessKeyId: "key", + secretAccessKey: "secret", + }); + + const signed = await storage.getSignedUploadUrl({ + key: "empty.pdf", + contentType: "application/pdf", + size: 0, + }); + + expect(signed.headers["Content-Length"]).toBe("0"); + }); +});