diff --git a/convex/crons.ts b/convex/crons.ts index a62a0b67..0fa8b993 100644 --- a/convex/crons.ts +++ b/convex/crons.ts @@ -3,6 +3,10 @@ import { purgeOldOperationEventsRef, purgeOldTombstonesRef, } from "./maintenance"; +import { + markStaleImageUploadsDeletedRef, + sweepDeletedImageAssetsRef, +} from "./images"; const crons = cronJobs(); @@ -18,5 +22,17 @@ crons.interval( purgeOldTombstonesRef, {}, ); +crons.interval( + "mark-stale-image-uploads-deleted", + { hours: 1 }, + markStaleImageUploadsDeletedRef, + {}, +); +crons.interval( + "sweep-deleted-image-assets", + { hours: 1 }, + sweepDeletedImageAssetsRef, + {}, +); export default crons; diff --git a/convex/function_spec.json b/convex/function_spec.json index 25546aba..8933060f 100644 --- a/convex/function_spec.json +++ b/convex/function_spec.json @@ -39809,6 +39809,25 @@ "kind": "public" } }, + { + "args": { + "type": "object", + "value": { + "limit": { + "fieldType": { + "type": "number" + }, + "optional": true + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:claimDeletedImageAssets", + "returns": null, + "visibility": { + "kind": "internal" + } + }, { "args": { "type": "object", @@ -40112,6 +40131,32 @@ "kind": "public" } }, + { + "args": { + "type": "object", + "value": { + "assetId": { + "fieldType": { + "tableName": "imageAssets", + "type": "id" + }, + "optional": false + }, + "r2ObjectDeleted": { + "fieldType": { + "type": "boolean" + }, + "optional": false + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:finalizeDeletedImageAsset", + "returns": null, + "visibility": { + "kind": "internal" + } + }, { "args": { "type": "object", @@ -40291,6 +40336,26 @@ "kind": "public" } }, + { + "args": { + "type": "object", + "value": { + "assetId": { + "fieldType": { + "tableName": "imageAssets", + "type": "id" + }, + "optional": false + } + } + }, + "functionType": "Query", + "identifier": "images.js:getDeletedImageAssetTarget", + "returns": null, + "visibility": { + "kind": "internal" + } + }, { "args": { "type": "object", @@ -40527,15 +40592,13 @@ "args": { "type": "object", "value": { - "limit": { - "fieldType": { - "type": "number" - }, - "optional": true - }, - "staleBefore": { + "assetIds": { "fieldType": { - "type": "number" + "type": "array", + "value": { + "tableName": "imageAssets", + "type": "id" + } }, "optional": false }, @@ -40547,8 +40610,8 @@ } } }, - "functionType": "Query", - "identifier": "images.js:listStaleUploadDeletionTargets", + "functionType": "Mutation", + "identifier": "images.js:markDeletedAssetRefsForStrategy", "returns": null, "visibility": { "kind": "internal" @@ -40558,26 +40621,53 @@ "args": { "type": "object", "value": { - "assetIds": { + "assetPublicIds": { "fieldType": { "type": "array", "value": { - "tableName": "imageAssets", - "type": "id" + "type": "string" } }, "optional": false }, - "strategyPublicId": { + "pageId": { "fieldType": { - "type": "string" + "tableName": "pages", + "type": "id" + }, + "optional": false + }, + "strategyId": { + "fieldType": { + "tableName": "strategies", + "type": "id" }, "optional": false } } }, "functionType": "Mutation", - "identifier": "images.js:markDeletedAssetRefsForStrategy", + "identifier": "images.js:markDeletedPageImageAssets", + "returns": null, + "visibility": { + "kind": "internal" + } + }, + { + "args": { + "type": "object", + "value": { + "strategyId": { + "fieldType": { + "tableName": "strategies", + "type": "id" + }, + "optional": false + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:markDeletedStrategyImageAssets", "returns": null, "visibility": { "kind": "internal" @@ -40691,18 +40781,79 @@ "fieldType": { "type": "number" }, + "optional": true + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:markStaleImageUploadsDeleted", + "returns": null, + "visibility": { + "kind": "internal" + } + }, + { + "args": { + "type": "object", + "value": { + "assetIds": { + "fieldType": { + "type": "array", + "value": { + "tableName": "imageAssets", + "type": "id" + } + }, "optional": false }, - "strategyPublicId": { + "retryAfterMs": { "fieldType": { - "type": "string" + "type": "number" }, - "optional": false + "optional": true + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:releaseImageAssetDeletionClaims", + "returns": null, + "visibility": { + "kind": "internal" + } + }, + { + "args": { + "type": "object", + "value": { + "delayMs": { + "fieldType": { + "type": "number" + }, + "optional": true + } + } + }, + "functionType": "Mutation", + "identifier": "images.js:scheduleDeletedImageAssetSweep", + "returns": null, + "visibility": { + "kind": "internal" + } + }, + { + "args": { + "type": "object", + "value": { + "limit": { + "fieldType": { + "type": "number" + }, + "optional": true } } }, "functionType": "Action", - "identifier": "images.js:sweepStaleUploadsForStrategy", + "identifier": "images.js:sweepDeletedImageAssets", "returns": null, "visibility": { "kind": "internal" @@ -47715,6 +47866,13 @@ "type": "id" }, "optional": false + }, + "strategyId": { + "fieldType": { + "tableName": "strategies", + "type": "id" + }, + "optional": false } } }, diff --git a/convex/imageAssetLifecycle.test.ts b/convex/imageAssetLifecycle.test.ts new file mode 100644 index 00000000..9c6a425b --- /dev/null +++ b/convex/imageAssetLifecycle.test.ts @@ -0,0 +1,639 @@ +import { + convexTest, + type TestConvexForDataModel, + type TestConvexForDataModelAndIdentity, +} from "convex-test"; +import { makeFunctionReference } from "convex/server"; +import { afterEach, beforeAll, describe, expect, test, vi } from "vitest"; +import type { DataModel } from "./_generated/dataModel"; +import cronDefinitions from "./crons"; +import schema from "./schema"; +import { modules } from "./test.setup"; + +const ensureCurrentUser = makeFunctionReference<"mutation">( + "users:ensureCurrentUser", +); +const createStrategy = makeFunctionReference<"mutation">( + "strategies:createWithInitialPage", +); +const addPage = makeFunctionReference<"mutation">("pages:add"); +const deletePage = makeFunctionReference<"mutation">("pages:delete"); +const deleteStrategy = makeFunctionReference<"mutation">("strategies:delete"); +const markStaleImageUploadsDeleted = makeFunctionReference<"mutation">( + "images:markStaleImageUploadsDeleted", +); +const sweepDeletedImageAssets = makeFunctionReference<"action">( + "images:sweepDeletedImageAssets", +); +const completeUpload = makeFunctionReference<"action">("images:completeUpload"); +const getAssetUrl = makeFunctionReference<"query">("images:getAssetUrl"); + +type Harness = TestConvexForDataModel; +type RootHarness = TestConvexForDataModelAndIdentity; + +const strategyPublicId = "asset-lifecycle-strategy"; +const pageA = "asset-page-a"; +const pageB = "asset-page-b"; + +function identity() { + return { + issuer: "https://asset-lifecycle.test", + subject: "owner", + tokenIdentifier: "asset-lifecycle|owner", + name: "Asset Owner", + }; +} + +async function createHarness(): Promise<{ + t: RootHarness; + owner: Harness; +}> { + const t = convexTest(schema, modules); + const owner = t.withIdentity(identity()); + await owner.mutation(ensureCurrentUser, {}); + return { t, owner }; +} + +async function seedStrategy(owner: Harness): Promise { + await owner.mutation(createStrategy, { + publicId: strategyPublicId, + name: "Asset lifecycle", + mapData: "ascent", + initialPagePublicId: pageA, + initialPageName: "Page 1", + initialPageIsAttack: true, + }); +} + +async function getStrategyAndPages(t: RootHarness) { + return await t.run(async (ctx) => { + const strategy = await ctx.db + .query("strategies") + .withIndex("by_publicId", (q) => q.eq("publicId", strategyPublicId)) + .unique(); + if (strategy === null) { + throw new Error("Missing Strategy test row"); + } + const pages = await ctx.db + .query("pages") + .withIndex("by_strategyId", (q) => q.eq("strategyId", strategy._id)) + .collect(); + return { strategy, pages }; + }); +} + +function imagePayload(assetPublicId: string) { + return { + kind: "image" as const, + payloadVersion: 1, + data: { id: assetPublicId, elementType: "image" }, + }; +} + +function lineupPayload(assetPublicId: string) { + return { + kind: "lineupGroup" as const, + payloadVersion: 1, + data: { items: [{ images: [{ id: assetPublicId }] }] }, + }; +} + +function mockR2Deletes(statuses: number[] = [204]) { + let callIndex = 0; + const fetchMock = vi.fn( + async (_input: RequestInfo | URL, _init?: RequestInit) => { + const status = statuses[Math.min(callIndex, statuses.length - 1)]!; + callIndex += 1; + return new Response(null, { status }); + }, + ); + vi.stubGlobal("fetch", fetchMock); + return fetchMock; +} + +async function allAssets(t: RootHarness) { + return await t.run( + async (ctx) => await ctx.db.query("imageAssets").collect(), + ); +} + +beforeAll(() => { + process.env.R2_ACCOUNT_ID = "asset-lifecycle-account"; + process.env.R2_BUCKET = "asset-lifecycle-bucket"; + process.env.R2_ACCESS_KEY_ID = "asset-lifecycle-access-key"; + process.env.R2_SECRET_ACCESS_KEY = "asset-lifecycle-secret"; + process.env.R2_PUBLIC_BASE_URL = "https://assets.asset-lifecycle.test"; + process.env.R2_S3_ENDPOINT = "https://asset-lifecycle.r2.test"; +}); + +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); +}); + +describe("image asset lifecycle", () => { + test("page deletion removes only assets unreferenced by remaining Pages and Lineups", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + await owner.mutation(addPage, { + strategyPublicId, + expectedRevision: 0, + pagePublicId: pageB, + name: "Page 2", + sortIndex: 1, + isAttack: false, + }); + const { strategy, pages } = await getStrategyAndPages(t); + const pageAId = pages.find((page) => page.publicId === pageA)?._id; + const pageBId = pages.find((page) => page.publicId === pageB)?._id; + if (pageAId === undefined || pageBId === undefined) { + throw new Error("Missing Page test rows"); + } + + await t.run(async (ctx) => { + const now = Date.now(); + for (const [publicId, objectKey] of [ + ["page-only", "pages/page-only.png"], + ["tombstoned-page-only", "pages/tombstoned-page-only.png"], + ["still-used", "pages/still-used.png"], + ["still-used-by-element", "pages/still-used-by-element.png"], + ] as const) { + await ctx.db.insert("imageAssets", { + publicId, + provider: "r2", + strategyId: strategy._id, + objectKey, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + } + for (const [publicId, assetPublicId] of [ + ["page-only-element", "page-only"], + ["tombstoned-element", "tombstoned-page-only"], + ["shared-element", "still-used"], + ["shared-page-element", "still-used-by-element"], + ] as const) { + await ctx.db.insert("elements", { + publicId, + strategyId: strategy._id, + pageId: pageAId, + elementType: "image", + payloadKind: "image", + payloadVersion: 1, + payload: imagePayload(assetPublicId), + sortIndex: 0, + revision: 1, + deleted: publicId === "tombstoned-element", + createdAt: now, + updatedAt: now, + }); + } + await ctx.db.insert("lineups", { + publicId: "remaining-lineup", + strategyId: strategy._id, + pageId: pageBId, + payloadKind: "lineupGroup", + payloadVersion: 1, + payload: lineupPayload("still-used"), + sortIndex: 0, + revision: 1, + deleted: false, + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("elements", { + publicId: "remaining-image-element", + strategyId: strategy._id, + pageId: pageBId, + elementType: "image", + payloadKind: "image", + payloadVersion: 1, + payload: imagePayload("still-used-by-element"), + sortIndex: 1, + revision: 1, + deleted: false, + createdAt: now, + updatedAt: now, + }); + }); + + await owner.mutation(deletePage, { + strategyPublicId, + pagePublicId: pageA, + expectedRevision: 1, + }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + const assets = await allAssets(t); + expect(assets).toMatchObject([ + { publicId: "still-used", uploadStatus: "active" }, + { publicId: "still-used-by-element", uploadStatus: "active" }, + ]); + expect(fetchMock).toHaveBeenCalledTimes(2); + expect(fetchMock.mock.calls[0]?.[1]).toMatchObject({ method: "DELETE" }); + expect(fetchMock.mock.calls.map((call) => String(call[0])).sort()).toEqual( + expect.arrayContaining([ + expect.stringContaining("page-only.png"), + expect.stringContaining("tombstoned-page-only.png"), + ]), + ); + }); + + test("Strategy deletion reclaims exact-owned R2 and Convex assets but preserves ambiguous shared legacy rows", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + const { uniqueStorageId, sharedStorageId } = await t.run(async (ctx) => ({ + uniqueStorageId: await ctx.storage.store(new Blob(["unique"])), + sharedStorageId: await ctx.storage.store(new Blob(["shared"])), + })); + + await t.run(async (ctx) => { + const now = Date.now(); + await ctx.db.insert("imageAssets", { + publicId: "owned-r2", + provider: "r2", + strategyId: strategy._id, + objectKey: "strategies/owned/delete.png", + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "owned-shared-r2", + provider: "r2", + strategyId: strategy._id, + objectKey: "legacy/shared.png", + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "legacy-shared-r2", + provider: "r2", + objectKey: "legacy/shared.png", + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "legacy-deleted-r2", + provider: "r2", + objectKey: "legacy/ambiguous-deleted.png", + uploadStatus: "deleted", + deletedAt: now, + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "owned-convex", + provider: "convex", + strategyId: strategy._id, + storageId: uniqueStorageId, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "owned-shared-convex", + provider: "convex", + strategyId: strategy._id, + storageId: sharedStorageId, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("imageAssets", { + publicId: "legacy-shared-convex", + provider: "convex", + storageId: sharedStorageId, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + }); + + await owner.mutation(deleteStrategy, { + strategyPublicId, + expectedRevision: 0, + }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + expect(await allAssets(t)).toMatchObject([ + { publicId: "legacy-shared-r2", objectKey: "legacy/shared.png" }, + { + publicId: "legacy-deleted-r2", + objectKey: "legacy/ambiguous-deleted.png", + }, + { publicId: "legacy-shared-convex", storageId: sharedStorageId }, + ]); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(String(fetchMock.mock.calls[0]?.[0])).toContain("delete.png"); + await expect( + t.run(async (ctx) => (await ctx.storage.get(uniqueStorageId)) === null), + ).resolves.toBe(true); + await expect( + t.run(async (ctx) => (await ctx.storage.get(sharedStorageId)) !== null), + ).resolves.toBe(true); + }); + + test("R2 failure keeps the tombstone target for an idempotent retry", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes([500, 404]); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + const assetId = await t.run(async (ctx) => { + const now = Date.now(); + return await ctx.db.insert("imageAssets", { + publicId: "retry-r2", + provider: "r2", + strategyId: strategy._id, + objectKey: "retry/keep-target.png", + uploadStatus: "deleted", + deletedAt: now, + createdAt: now, + updatedAt: now, + }); + }); + + await expect( + t.action(sweepDeletedImageAssets, { limit: 1 }), + ).resolves.toMatchObject({ deleted: 0, failed: 1 }); + await expect( + t.run(async (ctx) => await ctx.db.get(assetId)), + ).resolves.toMatchObject({ + uploadStatus: "deleted", + objectKey: "retry/keep-target.png", + }); + + await expect( + t.action(sweepDeletedImageAssets, { limit: 1 }), + ).resolves.toMatchObject({ deleted: 1, failed: 0 }); + await expect( + t.run(async (ctx) => await ctx.db.get(assetId)), + ).resolves.toBeNull(); + expect(fetchMock).toHaveBeenCalledTimes(2); + await t.finishAllScheduledFunctions(vi.runAllTimers); + }); + + test("the last deleted row sharing an R2 key removes the object", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + await t.run(async (ctx) => { + const now = Date.now(); + for (const publicId of ["duplicate-a", "duplicate-b"]) { + await ctx.db.insert("imageAssets", { + publicId, + provider: "r2", + strategyId: strategy._id, + objectKey: "duplicates/shared-deleted.png", + uploadStatus: "deleted", + deletedAt: now, + createdAt: now, + updatedAt: now, + }); + } + }); + + await expect( + t.action(sweepDeletedImageAssets, { limit: 2 }), + ).resolves.toMatchObject({ deleted: 2, failed: 0 }); + expect(await allAssets(t)).toEqual([]); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(String(fetchMock.mock.calls[0]?.[0])).toContain( + "shared-deleted.png", + ); + await t.finishAllScheduledFunctions(vi.runAllTimers); + }); + + test("overlapping cleanup actions claim one R2 tombstone once", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + await t.run(async (ctx) => { + const now = Date.now(); + await ctx.db.insert("imageAssets", { + publicId: "single-claim", + provider: "r2", + strategyId: strategy._id, + objectKey: "claims/single.png", + uploadStatus: "deleted", + deletedAt: now, + createdAt: now, + updatedAt: now, + }); + }); + + const results = (await Promise.all([ + t.action(sweepDeletedImageAssets, { limit: 1 }), + t.action(sweepDeletedImageAssets, { limit: 1 }), + ])) as Array<{ deleted: number; failed: number }>; + + expect(results.reduce((total, result) => total + result.deleted, 0)).toBe( + 1, + ); + expect(results.reduce((total, result) => total + result.failed, 0)).toBe(0); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(await allAssets(t)).toEqual([]); + await t.finishAllScheduledFunctions(vi.runAllTimers); + }); + + test("the hourly path releases a stranded claim and completes its deletion", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + const now = Date.now(); + await t.run(async (ctx) => { + await ctx.db.insert("imageAssets", { + publicId: "stranded-claim", + provider: "r2", + strategyId: strategy._id, + objectKey: "claims/stranded.png", + uploadStatus: "deleted", + deletedAt: now - 60 * 60 * 1000, + cleanupClaimedAt: now - 16 * 60 * 1000, + createdAt: now - 60 * 60 * 1000, + updatedAt: now - 60 * 60 * 1000, + }); + }); + + await expect( + t.mutation(markStaleImageUploadsDeleted, {}), + ).resolves.toMatchObject({ deleted: 0, released: 1 }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + expect(await allAssets(t)).toEqual([]); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + test("legacy reads survive while completion inserts an exact-owned replacement", async () => { + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy, pages } = await getStrategyAndPages(t); + const pageId = pages.find((page) => page.publicId === pageA)?._id; + if (pageId === undefined) { + throw new Error("Missing Page test row"); + } + const { legacyStorageId, replacementStorageId } = await t.run( + async (ctx) => ({ + legacyStorageId: await ctx.storage.store(new Blob(["legacy"])), + replacementStorageId: await ctx.storage.store( + new Blob(["replacement"]), + ), + }), + ); + await t.run(async (ctx) => { + const now = Date.now(); + await ctx.db.insert("imageAssets", { + publicId: "legacy-readable", + provider: "convex", + storageId: legacyStorageId, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + await ctx.db.insert("elements", { + publicId: "legacy-image-element", + strategyId: strategy._id, + pageId, + elementType: "image", + payloadKind: "image", + payloadVersion: 1, + payload: imagePayload("legacy-readable"), + sortIndex: 0, + revision: 1, + deleted: false, + createdAt: now, + updatedAt: now, + }); + }); + + const legacyResult = (await owner.query(getAssetUrl, { + strategyPublicId, + assetPublicId: "legacy-readable", + })) as { url: string | null }; + expect(legacyResult.url).toMatch( + /^https:\/\/some-deployment\.convex\.cloud\//, + ); + await owner.action(completeUpload, { + strategyPublicId, + assetPublicId: "legacy-readable", + provider: "convex", + storageId: replacementStorageId, + fileExtension: ".png", + mimeType: "image/png", + }); + + const assets = await allAssets(t); + expect( + assets.find((asset) => asset.strategyId === undefined), + ).toMatchObject({ + publicId: "legacy-readable", + storageId: legacyStorageId, + }); + expect( + assets.find((asset) => asset.strategyId === strategy._id), + ).toMatchObject({ + publicId: "legacy-readable", + storageId: replacementStorageId, + }); + const replacementResult = (await owner.query(getAssetUrl, { + strategyPublicId, + assetPublicId: "legacy-readable", + })) as { url: string | null }; + expect(replacementResult.url).toMatch( + /^https:\/\/some-deployment\.convex\.cloud\//, + ); + expect(replacementResult.url).not.toBe(legacyResult.url); + }); + + test("the cron path marks stale owned uploads without user auth and leaves ambiguous legacy rows", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + const staleAt = Date.now() - 48 * 60 * 60 * 1000; + await t.run(async (ctx) => { + await ctx.db.insert("imageAssets", { + publicId: "stale-owned", + provider: "r2", + strategyId: strategy._id, + objectKey: "stale/owned.png", + uploadStatus: "pending", + createdAt: staleAt, + updatedAt: staleAt, + }); + await ctx.db.insert("imageAssets", { + publicId: "stale-legacy", + provider: "r2", + objectKey: "stale/legacy.png", + uploadStatus: "failed", + createdAt: staleAt, + updatedAt: staleAt, + }); + }); + + expect( + cronDefinitions.crons["mark-stale-image-uploads-deleted"], + ).toMatchObject({ + name: "images:markStaleImageUploadsDeleted", + schedule: { hours: 1, type: "interval" }, + }); + await expect( + t.mutation(markStaleImageUploadsDeleted, { + staleBefore: Date.now() - 24 * 60 * 60 * 1000, + }), + ).resolves.toMatchObject({ deleted: 1 }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + expect(await allAssets(t)).toMatchObject([ + { publicId: "stale-legacy", uploadStatus: "failed" }, + ]); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + test("Strategy cleanup continues through bounded database and external deletion batches", async () => { + vi.useFakeTimers(); + const fetchMock = mockR2Deletes(); + const { t, owner } = await createHarness(); + await seedStrategy(owner); + const { strategy } = await getStrategyAndPages(t); + await t.run(async (ctx) => { + const now = Date.now(); + for (let index = 0; index < 126; index += 1) { + await ctx.db.insert("imageAssets", { + publicId: `bounded-${index}`, + provider: "r2", + strategyId: strategy._id, + objectKey: `bounded/${index}.png`, + uploadStatus: "active", + createdAt: now, + updatedAt: now, + }); + } + }); + + await owner.mutation(deleteStrategy, { + strategyPublicId, + expectedRevision: 0, + }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + expect(await allAssets(t)).toEqual([]); + expect(fetchMock).toHaveBeenCalledTimes(126); + }, 30_000); +}); diff --git a/convex/images.ts b/convex/images.ts index 8a6e85bb..4f2e8995 100644 --- a/convex/images.ts +++ b/convex/images.ts @@ -8,7 +8,6 @@ import { getViewerAssetForStrategy, inferProvider, inferUploadStatus, - isVisibleAsset, serializeAssetForViewer, type Provider, type UploadStatus, @@ -48,6 +47,7 @@ import { imageProviderValidator, okResultValidator, } from "./lib/publicValidators"; +import { makeFunctionReference } from "convex/server"; type AnyCtx = MutationCtx | QueryCtx; @@ -55,9 +55,41 @@ type DeletionTarget = { assetId: Id<"imageAssets">; provider: Provider; objectKey: string | null; + sharedTarget: boolean; }; const maxDeletionBatch = 100; +const physicalDeletionBatch = 25; +const pageAssetIdBatch = 50; +const staleUploadAgeMs = 24 * 60 * 60 * 1000; +const staleDeletionClaimAgeMs = 15 * 60 * 1000; +const deletionRetryDelayMs = 60 * 1000; + +export const markDeletedPageImageAssetsRef = makeFunctionReference<"mutation">( + "images:markDeletedPageImageAssets", +); +export const markDeletedStrategyImageAssetsRef = + makeFunctionReference<"mutation">("images:markDeletedStrategyImageAssets"); +export const markStaleImageUploadsDeletedRef = + makeFunctionReference<"mutation">("images:markStaleImageUploadsDeleted"); +export const sweepDeletedImageAssetsRef = makeFunctionReference<"action">( + "images:sweepDeletedImageAssets", +); +const claimDeletedImageAssetsRef = makeFunctionReference<"mutation">( + "images:claimDeletedImageAssets", +); +const getDeletedImageAssetTargetRef = makeFunctionReference<"query">( + "images:getDeletedImageAssetTarget", +); +const finalizeDeletedImageAssetRef = makeFunctionReference<"mutation">( + "images:finalizeDeletedImageAsset", +); +const scheduleDeletedImageAssetSweepRef = makeFunctionReference<"mutation">( + "images:scheduleDeletedImageAssetSweep", +); +const releaseImageAssetDeletionClaimsRef = makeFunctionReference<"mutation">( + "images:releaseImageAssetDeletionClaims", +); function createUploadAttemptPublicId(): string { return crypto.randomUUID(); @@ -66,6 +98,7 @@ function createUploadAttemptPublicId(): string { async function collectReferencedAssetIdsForStrategy( ctx: AnyCtx, strategyId: Doc<"strategies">["_id"], + excludedPageId?: Id<"pages">, ): Promise> { const assetIds = new Set(); @@ -73,7 +106,11 @@ async function collectReferencedAssetIdsForStrategy( .query("elements") .withIndex("by_strategyId", (q) => q.eq("strategyId", strategyId)); for await (const element of elementQuery) { - if (element.deleted || element.elementType !== "image") { + if ( + element.deleted || + element.pageId === excludedPageId || + element.elementType !== "image" + ) { continue; } const assetId = collectAssetIdFromElementPayload(element.payload); @@ -86,7 +123,7 @@ async function collectReferencedAssetIdsForStrategy( .query("lineups") .withIndex("by_strategyId", (q) => q.eq("strategyId", strategyId)); for await (const lineup of lineupQuery) { - if (lineup.deleted) { + if (lineup.deleted || lineup.pageId === excludedPageId) { continue; } for (const assetId of collectAssetIdsFromLineupPayload(lineup.payload)) { @@ -112,20 +149,7 @@ async function getDeletionCandidateForStrategy( const ownedCandidate = strategyCandidates.find((asset) => inferUploadStatus(asset) !== "deleted") ?? null; - if (ownedCandidate !== null) { - return ownedCandidate; - } - - const legacyCandidates = await ctx.db - .query("imageAssets") - .withIndex("by_publicId", (q) => q.eq("publicId", assetPublicId)) - .order("desc") - .take(20); - return ( - legacyCandidates.find( - (asset) => asset.strategyId === undefined && isVisibleAsset(asset), - ) ?? null - ); + return ownedCandidate; } async function strategyReferencesAsset( @@ -140,29 +164,52 @@ async function strategyReferencesAsset( return referencedAssetIds.has(assetPublicId); } -function deletionTargetForAsset(asset: Doc<"imageAssets">): DeletionTarget { - return { - assetId: asset._id, - provider: inferProvider(asset), - objectKey: asset.objectKey ?? null, - }; -} - async function markImageAssetDeleted( ctx: MutationCtx, asset: Doc<"imageAssets">, now: number, ): Promise { - if (asset.storageId !== undefined) { - await ctx.storage.delete(asset.storageId); - } await ctx.db.patch(asset._id, { uploadStatus: "deleted", deletedAt: now, + cleanupClaimedAt: undefined, updatedAt: now, }); } +async function schedulePhysicalDeletion( + ctx: MutationCtx, + delayMs = 0, +): Promise { + await ctx.scheduler.runAfter(delayMs, sweepDeletedImageAssetsRef, {}); +} + +function chunks(values: T[], size: number): T[][] { + const result: T[][] = []; + for (let index = 0; index < values.length; index += size) { + result.push(values.slice(index, index + size)); + } + return result; +} + +export async function captureDeletedPageImageAssets( + ctx: MutationCtx, + args: { + strategyId: Id<"strategies">; + pageId: Id<"pages">; + assetPublicIds: Iterable; + }, +): Promise { + const assetPublicIds = [...new Set(args.assetPublicIds)]; + for (const assetIdChunk of chunks(assetPublicIds, pageAssetIdBatch)) { + await ctx.scheduler.runAfter(0, markDeletedPageImageAssetsRef, { + strategyId: args.strategyId, + pageId: args.pageId, + assetPublicIds: assetIdChunk, + }); + } +} + export const generateUploadUrl = action({ args: { strategyPublicId: v.string(), @@ -384,24 +431,17 @@ export const completeUpload = action({ throw invalidPayloadError("Uploaded image failed size or MIME validation."); } - const result: { ok: true; replaced: DeletionTarget[] } = - await ctx.runMutation(internal.images.markR2UploadActive, { - strategyPublicId: args.strategyPublicId, - assetPublicId: args.assetPublicId, - uploadId: intent.uploadId, - byteSize: actualByteSize, - etag: metadata.etag ?? args.etag, - mimeType: actualMimeType, - fileExtension: args.fileExtension ?? intent.fileExtension, - width: args.width, - height: args.height, - }); - - for (const target of result.replaced) { - if (target.provider === "r2" && target.objectKey !== null) { - await deleteR2Object(config, target.objectKey); - } - } + await ctx.runMutation(internal.images.markR2UploadActive, { + strategyPublicId: args.strategyPublicId, + assetPublicId: args.assetPublicId, + uploadId: intent.uploadId, + byteSize: actualByteSize, + etag: metadata.etag ?? args.etag, + mimeType: actualMimeType, + fileExtension: args.fileExtension ?? intent.fileExtension, + width: args.width, + height: args.height, + }); return { ok: true as const, @@ -432,7 +472,8 @@ export const getR2UploadIntentForCompletion = internalQuery({ asset.strategyId !== strategy._id || asset.publicId !== args.assetPublicId || inferProvider(asset) !== "r2" || - asset.objectKey === undefined + asset.objectKey === undefined || + inferUploadStatus(asset) === "deleted" ) { throw errorWithCode("UPLOAD_INTENT_NOT_FOUND", "Upload intent not found."); } @@ -476,7 +517,8 @@ export const markR2UploadActive = internalMutation({ asset.strategyId !== strategy._id || asset.publicId !== args.assetPublicId || inferProvider(asset) !== "r2" || - asset.objectKey === undefined + asset.objectKey === undefined || + inferUploadStatus(asset) === "deleted" ) { throw errorWithCode("UPLOAD_INTENT_NOT_FOUND", "Upload intent not found."); } @@ -503,15 +545,18 @@ export const markR2UploadActive = internalMutation({ .eq("uploadStatus", "active"), ) .take(20); - const replaced: DeletionTarget[] = []; + let replaced = 0; for (const olderAsset of olderActiveAssets) { if (olderAsset._id === asset._id) { continue; } - replaced.push(deletionTargetForAsset(olderAsset)); await markImageAssetDeleted(ctx, olderAsset, now); + replaced += 1; } + if (replaced > 0) { + await schedulePhysicalDeletion(ctx); + } return { ok: true as const, replaced }; }, }); @@ -581,7 +626,8 @@ export const completeLegacyUpload = internalMutation({ }); if ( previousStorageId !== undefined && - previousStorageId !== args.storageId + previousStorageId !== args.storageId && + !(await hasSharedDeletionTarget(ctx, existing)) ) { await ctx.storage.delete(previousStorageId); } @@ -680,15 +726,11 @@ export const deleteAssetRef = action({ }, returns: okResultValidator, handler: async (ctx, args) => { - const target: DeletionTarget = await ctx.runQuery( + const target: { assetId: Id<"imageAssets"> } = await ctx.runQuery( internal.images.getAssetDeletionTarget, args, ); - if (target.provider === "r2" && target.objectKey !== null) { - await deleteR2Object(getR2Config(), target.objectKey); - } - await ctx.runMutation(internal.images.markDeletedAssetRefsForStrategy, { strategyPublicId: args.strategyPublicId, assetIds: [target.assetId], @@ -715,14 +757,11 @@ export const getAssetDeletionTarget = internalQuery({ throw notFoundError("Asset", args.assetPublicId); } - if ( - asset.strategyId === undefined && - !(await strategyReferencesAsset(ctx, strategy._id, args.assetPublicId)) - ) { - throw notFoundError("Asset", args.assetPublicId); + if (await strategyReferencesAsset(ctx, strategy._id, args.assetPublicId)) { + throw conflictError("Asset is still referenced by this Strategy."); } - return deletionTargetForAsset(asset); + return { assetId: asset._id }; }, }); @@ -737,132 +776,417 @@ export const markDeletedAssetRefsForStrategy = internalMutation({ const now = Date.now(); let deleted = 0; + let shouldSweep = false; + const referencedAssetIds = await collectReferencedAssetIdsForStrategy( + ctx, + strategy._id, + ); for (const assetId of args.assetIds.slice(0, maxDeletionBatch)) { const asset = await ctx.db.get(assetId); - if (asset === null || inferUploadStatus(asset) === "deleted") { + if (asset === null) { continue; } - if (asset.strategyId !== undefined && asset.strategyId !== strategy._id) { + if ( + asset.strategyId !== strategy._id || + referencedAssetIds.has(asset.publicId) + ) { continue; } - await markImageAssetDeleted(ctx, asset, now); - deleted += 1; + shouldSweep = true; + if (inferUploadStatus(asset) !== "deleted") { + await markImageAssetDeleted(ctx, asset, now); + deleted += 1; + } + } + if (shouldSweep) { + await schedulePhysicalDeletion(ctx); } return { ok: true, deleted }; }, }); -export const listPotentiallyStale = internalQuery({ +export const markDeletedPageImageAssets = internalMutation({ args: { - strategyPublicId: v.string(), - limit: v.optional(v.number()), + strategyId: v.id("strategies"), + pageId: v.id("pages"), + assetPublicIds: v.array(v.string()), }, handler: async (ctx, args) => { - const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); - await assertStrategyRole(ctx, strategy, "editor"); - const limit = Math.max(1, Math.min(args.limit ?? 200, 500)); - - const referencedAssetIds = await collectReferencedAssetIdsForStrategy( + const candidateIds = new Set( + args.assetPublicIds.slice(0, pageAssetIdBatch), + ); + const remainingReferences = await collectReferencedAssetIdsForStrategy( ctx, - strategy._id, + args.strategyId, + args.pageId, ); + for (const referencedId of remainingReferences) { + candidateIds.delete(referencedId); + } + + const assets: Doc<"imageAssets">[] = []; + for (const assetPublicId of candidateIds) { + const remainingSlots = maxDeletionBatch - assets.length; + if (remainingSlots <= 0) { + break; + } + const matches = await ctx.db + .query("imageAssets") + .withIndex("by_strategyId_and_publicId", (q) => + q.eq("strategyId", args.strategyId).eq("publicId", assetPublicId), + ) + .filter((q) => q.neq(q.field("uploadStatus"), "deleted")) + .take(remainingSlots); + assets.push(...matches); + } + + const now = Date.now(); + for (const asset of assets) { + await markImageAssetDeleted(ctx, asset, now); + } + if (assets.length > 0) { + await schedulePhysicalDeletion(ctx); + } + if (assets.length === maxDeletionBatch) { + await ctx.scheduler.runAfter(0, markDeletedPageImageAssetsRef, args); + } + return { ok: true as const, deleted: assets.length }; + }, +}); + +export const markDeletedStrategyImageAssets = internalMutation({ + args: { + strategyId: v.id("strategies"), + }, + handler: async (ctx, args) => { const assets = await ctx.db .query("imageAssets") - .withIndex("by_strategyId", (q) => q.eq("strategyId", strategy._id)) - .order("desc") + .withIndex("by_strategyId", (q) => q.eq("strategyId", args.strategyId)) + .filter((q) => q.neq(q.field("uploadStatus"), "deleted")) + .take(maxDeletionBatch); + const now = Date.now(); + for (const asset of assets) { + await markImageAssetDeleted(ctx, asset, now); + } + if (assets.length > 0) { + await schedulePhysicalDeletion(ctx); + } + if (assets.length === maxDeletionBatch) { + await ctx.scheduler.runAfter(0, markDeletedStrategyImageAssetsRef, args); + } + return { ok: true as const, deleted: assets.length }; + }, +}); + +export const markStaleImageUploadsDeleted = internalMutation({ + args: { + staleBefore: v.optional(v.number()), + limit: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const staleBefore = args.staleBefore ?? Date.now() - staleUploadAgeMs; + const limit = Math.max( + 1, + Math.min(args.limit ?? maxDeletionBatch, maxDeletionBatch), + ); + const stuckClaims = await ctx.db + .query("imageAssets") + .withIndex("by_uploadStatus_and_updatedAt", (q) => + q.eq("uploadStatus", "deleted"), + ) + .filter((q) => + q.and( + q.neq(q.field("strategyId"), undefined), + q.neq(q.field("cleanupClaimedAt"), undefined), + q.lte( + q.field("cleanupClaimedAt"), + Date.now() - staleDeletionClaimAgeMs, + ), + ), + ) .take(limit); + for (const asset of stuckClaims) { + await ctx.db.patch(asset._id, { cleanupClaimedAt: undefined }); + } - const candidates = assets.filter((asset) => { - const status = inferUploadStatus(asset); - if (status === "deleted") { - return false; + const assets: Doc<"imageAssets">[] = []; + for (const status of ["pending", "failed"] as UploadStatus[]) { + const remainingSlots = limit - stuckClaims.length - assets.length; + if (remainingSlots <= 0) { + break; } - if (status === "pending" || status === "failed") { - return true; + const matches = await ctx.db + .query("imageAssets") + .withIndex("by_uploadStatus_and_updatedAt", (q) => + q.eq("uploadStatus", status).lte("updatedAt", staleBefore), + ) + .filter((q) => q.neq(q.field("strategyId"), undefined)) + .take(remainingSlots); + assets.push(...matches); } - return !referencedAssetIds.has(asset.publicId); + + const now = Date.now(); + for (const asset of assets) { + await markImageAssetDeleted(ctx, asset, now); + } + if (assets.length > 0 || stuckClaims.length > 0) { + await schedulePhysicalDeletion(ctx); + } + if (assets.length + stuckClaims.length === limit) { + await ctx.scheduler.runAfter(0, markStaleImageUploadsDeletedRef, { + staleBefore, + limit, + }); + } + return { + ok: true as const, + deleted: assets.length, + released: stuckClaims.length, + }; + }, }); - return await Promise.all( - candidates.map((asset) => serializeAssetForViewer(ctx, asset)), +async function hasSharedDeletionTarget( + ctx: QueryCtx | MutationCtx, + asset: Doc<"imageAssets">, +): Promise { + if (inferProvider(asset) === "r2") { + if (asset.objectKey === undefined) { + return false; + } + const matches = await ctx.db + .query("imageAssets") + .withIndex("by_objectKey", (q) => q.eq("objectKey", asset.objectKey)) + .take(2); + return matches.some((candidate) => candidate._id !== asset._id); + } + if (asset.storageId === undefined) { + return false; + } + const matches = await ctx.db + .query("imageAssets") + .withIndex("by_storageId", (q) => q.eq("storageId", asset.storageId)) + .take(2); + return matches.some((candidate) => candidate._id !== asset._id); +} + +export const claimDeletedImageAssets = internalMutation({ + args: { + limit: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const limit = Math.max( + 1, + Math.min(args.limit ?? physicalDeletionBatch, physicalDeletionBatch), ); + const assets = await ctx.db + .query("imageAssets") + .withIndex("by_uploadStatus_and_updatedAt", (q) => + q.eq("uploadStatus", "deleted"), + ) + .filter((q) => + q.and( + q.neq(q.field("strategyId"), undefined), + q.eq(q.field("cleanupClaimedAt"), undefined), + ), + ) + .take(limit); + const now = Date.now(); + for (const asset of assets) { + await ctx.db.patch(asset._id, { cleanupClaimedAt: now }); + } + return assets.map((asset) => asset._id); }, }); -export const sweepStaleUploadsForStrategy = internalAction({ +export const getDeletedImageAssetTarget = internalQuery({ args: { - strategyPublicId: v.string(), - staleBefore: v.number(), - limit: v.optional(v.number()), + assetId: v.id("imageAssets"), }, - handler: async ( + handler: async (ctx, args): Promise => { + const asset = await ctx.db.get(args.assetId); + if ( + asset === null || + inferUploadStatus(asset) !== "deleted" || + asset.cleanupClaimedAt === undefined + ) { + return null; + } + return { + assetId: asset._id, + provider: inferProvider(asset), + objectKey: asset.objectKey ?? null, + sharedTarget: await hasSharedDeletionTarget(ctx, asset), + }; + }, +}); + +export const finalizeDeletedImageAsset = internalMutation({ + args: { + assetId: v.id("imageAssets"), + r2ObjectDeleted: v.boolean(), + }, + handler: async (ctx, args) => { + const asset = await ctx.db.get(args.assetId); + if (asset === null) { + return { ok: true as const, finalized: true }; + } + if (inferUploadStatus(asset) !== "deleted") { + return { ok: true as const, finalized: false }; + } + + const sharedTarget = await hasSharedDeletionTarget(ctx, asset); + if (inferProvider(asset) === "r2") { + if ( + asset.objectKey !== undefined && + !sharedTarget && + !args.r2ObjectDeleted + ) { + return { ok: true as const, finalized: false }; + } + } else if (asset.storageId !== undefined && !sharedTarget) { + await ctx.storage.delete(asset.storageId); + } + await ctx.db.delete(asset._id); + return { ok: true as const, finalized: true }; + }, +}); + +export const scheduleDeletedImageAssetSweep = internalMutation({ + args: { + delayMs: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const delayMs = Math.max(0, Math.min(args.delayMs ?? 0, 60 * 60 * 1000)); + await schedulePhysicalDeletion(ctx, delayMs); + return { ok: true as const }; + }, +}); + +export const releaseImageAssetDeletionClaims = internalMutation({ + args: { + assetIds: v.array(v.id("imageAssets")), + retryAfterMs: v.optional(v.number()), + }, + handler: async (ctx, args) => { + for (const assetId of args.assetIds.slice(0, physicalDeletionBatch)) { + const asset = await ctx.db.get(assetId); + if ( + asset !== null && + inferUploadStatus(asset) === "deleted" && + asset.cleanupClaimedAt !== undefined + ) { + await ctx.db.patch(asset._id, { cleanupClaimedAt: undefined }); + } + } + await schedulePhysicalDeletion( ctx, - args, - ): Promise<{ - ok: true; - deleted: number; - }> => { - const targets: DeletionTarget[] = await ctx.runQuery( - internal.images.listStaleUploadDeletionTargets, - args, + Math.max(0, args.retryAfterMs ?? deletionRetryDelayMs), ); - const config = targets.some( - (target) => target.provider === "r2" && target.objectKey !== null, - ) - ? getR2Config() - : null; + return { ok: true as const }; + }, +}); - for (const target of targets) { +export const sweepDeletedImageAssets = internalAction({ + args: { + limit: v.optional(v.number()), + }, + handler: async (ctx, args) => { + const limit = Math.max( + 1, + Math.min(args.limit ?? physicalDeletionBatch, physicalDeletionBatch), + ); + const assetIds: Id<"imageAssets">[] = await ctx.runMutation( + claimDeletedImageAssetsRef, + { limit }, + ); + let deleted = 0; + let failed = 0; + const failedAssetIds: Id<"imageAssets">[] = []; + let config: ReturnType | null = null; + + for (const assetId of assetIds) { + try { + const target: DeletionTarget | null = await ctx.runQuery( + getDeletedImageAssetTargetRef, + { assetId }, + ); + if (target === null) { + continue; + } + let r2ObjectDeleted = false; if ( - config !== null && target.provider === "r2" && - target.objectKey !== null + target.objectKey !== null && + !target.sharedTarget ) { + config ??= getR2Config(); await deleteR2Object(config, target.objectKey); + r2ObjectDeleted = true; + } + const result: { finalized: boolean } = await ctx.runMutation( + finalizeDeletedImageAssetRef, + { assetId, r2ObjectDeleted }, + ); + if (result.finalized) { + deleted += 1; + } else { + failed += 1; + failedAssetIds.push(assetId); + } + } catch { + failed += 1; + failedAssetIds.push(assetId); } } - const result: { ok: boolean; deleted: number } = await ctx.runMutation( - internal.images.markDeletedAssetRefsForStrategy, - { - strategyPublicId: args.strategyPublicId, - assetIds: targets.map((target) => target.assetId), - }, - ); - return { ok: true, deleted: result.deleted }; + if (failedAssetIds.length > 0) { + await ctx.runMutation(releaseImageAssetDeletionClaimsRef, { + assetIds: failedAssetIds, + retryAfterMs: deletionRetryDelayMs, + }); + } else if (assetIds.length === limit) { + await ctx.runMutation(scheduleDeletedImageAssetSweepRef, { + delayMs: 0, + }); + } + return { ok: true as const, deleted, failed }; }, }); -export const listStaleUploadDeletionTargets = internalQuery({ +export const listPotentiallyStale = internalQuery({ args: { strategyPublicId: v.string(), - staleBefore: v.number(), limit: v.optional(v.number()), }, handler: async (ctx, args) => { const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); - const limit = Math.max(1, Math.min(args.limit ?? 50, maxDeletionBatch)); + const limit = Math.max(1, Math.min(args.limit ?? 200, 500)); - const targets: DeletionTarget[] = []; - for (const status of ["pending", "failed"] as UploadStatus[]) { - const candidates = await ctx.db + const referencedAssetIds = await collectReferencedAssetIdsForStrategy( + ctx, + strategy._id, + ); + const assets = await ctx.db .query("imageAssets") - .withIndex("by_strategyId_and_uploadStatus_and_updatedAt", (q) => - q - .eq("strategyId", strategy._id) - .eq("uploadStatus", status) - .lte("updatedAt", args.staleBefore), - ) - .take(limit - targets.length); - for (const asset of candidates) { - targets.push(deletionTargetForAsset(asset)); - } - if (targets.length >= limit) { - break; + .withIndex("by_strategyId", (q) => q.eq("strategyId", strategy._id)) + .order("desc") + .take(limit); + + const candidates = assets.filter((asset) => { + const status = inferUploadStatus(asset); + if (status === "deleted") { + return false; } + if (status === "pending" || status === "failed") { + return true; } + return !referencedAssetIds.has(asset.publicId); + }); - return targets; + return await Promise.all( + candidates.map((asset) => serializeAssetForViewer(ctx, asset)), + ); }, }); diff --git a/convex/maintenance.ts b/convex/maintenance.ts index a8339005..c27839a0 100644 --- a/convex/maintenance.ts +++ b/convex/maintenance.ts @@ -1,6 +1,11 @@ import { makeFunctionReference } from "convex/server"; import { internalMutation } from "./_generated/server"; import { v } from "convex/values"; +import { + collectAssetIdFromElementPayload, + collectAssetIdsFromLineupPayload, +} from "./lib/imageAssets"; +import { captureDeletedPageImageAssets } from "./images"; const MAINTENANCE_BATCH_SIZE = 200; const DAYS_30_MS = 30 * 24 * 60 * 60 * 1000; @@ -20,16 +25,13 @@ export const purgeOldTombstonesRef = makeFunctionReference<"mutation">( export const purgeDeletedPageOrphans = internalMutation({ args: { pageId: v.id("pages"), + strategyId: v.id("strategies"), }, handler: async (ctx, args) => { const elements = await ctx.db .query("elements") .withIndex("by_pageId", (q) => q.eq("pageId", args.pageId)) .take(MAINTENANCE_BATCH_SIZE); - for (const element of elements) { - await ctx.db.delete(element._id); - } - const remainingSlots = MAINTENANCE_BATCH_SIZE - elements.length; const lineups = remainingSlots > 0 @@ -38,6 +40,30 @@ export const purgeDeletedPageOrphans = internalMutation({ .withIndex("by_pageId", (q) => q.eq("pageId", args.pageId)) .take(remainingSlots) : []; + + const assetPublicIds = new Set(); + for (const element of elements) { + const assetPublicId = collectAssetIdFromElementPayload(element.payload); + if (assetPublicId !== null) { + assetPublicIds.add(assetPublicId); + } + } + for (const lineup of lineups) { + for (const assetPublicId of collectAssetIdsFromLineupPayload( + lineup.payload, + )) { + assetPublicIds.add(assetPublicId); + } + } + await captureDeletedPageImageAssets(ctx, { + strategyId: args.strategyId, + pageId: args.pageId, + assetPublicIds, + }); + + for (const element of elements) { + await ctx.db.delete(element._id); + } for (const lineup of lineups) { await ctx.db.delete(lineup._id); } @@ -50,7 +76,7 @@ export const purgeDeletedPageOrphans = internalMutation({ await ctx.scheduler.runAfter( 0, purgeDeletedPageOrphansRef, - { pageId: args.pageId }, + { pageId: args.pageId, strategyId: args.strategyId }, ); } }, diff --git a/convex/ops.ts b/convex/ops.ts index a86470af..72d66e3c 100644 --- a/convex/ops.ts +++ b/convex/ops.ts @@ -775,6 +775,7 @@ async function applyPageOp( await ctx.db.delete(existing._id); await ctx.scheduler.runAfter(0, purgeDeletedPageOrphansRef, { pageId: existing._id, + strategyId: strategy._id, }); const now = Date.now(); const remaining = sortByNumberField( diff --git a/convex/pages.ts b/convex/pages.ts index 216c9e73..e75b6135 100644 --- a/convex/pages.ts +++ b/convex/pages.ts @@ -220,6 +220,7 @@ const deletePage = mutation({ await ctx.db.delete(page._id); await ctx.scheduler.runAfter(0, purgeDeletedPageOrphansRef, { pageId: page._id, + strategyId: strategy._id, }); const ordered = sortByNumberField( diff --git a/convex/schema.ts b/convex/schema.ts index 3d433aca..47eb17f7 100644 --- a/convex/schema.ts +++ b/convex/schema.ts @@ -177,6 +177,7 @@ export default defineSchema({ etag: v.optional(v.string()), uploadedAt: v.optional(v.number()), deletedAt: v.optional(v.number()), + cleanupClaimedAt: v.optional(v.number()), createdAt: v.optional(v.number()), updatedAt: v.optional(v.number()), // Legacy rows may still have a storagePath that can help infer the extension. @@ -197,6 +198,7 @@ export default defineSchema({ "uploadStatus", ]) .index("by_uploadStatus_and_updatedAt", ["uploadStatus", "updatedAt"]) + .index("by_storageId", ["storageId"]) .index("by_objectKey", ["objectKey"]), operationEvents: defineTable({ strategyId: v.id("strategies"), diff --git a/convex/strategies.ts b/convex/strategies.ts index a0031190..81b7c730 100644 --- a/convex/strategies.ts +++ b/convex/strategies.ts @@ -21,6 +21,7 @@ import { forbiddenError, } from "./lib/errors"; import { purgeDeletedPageOrphansRef } from "./maintenance"; +import { markDeletedStrategyImageAssetsRef } from "./images"; import { createResultValidator, okResultValidator, @@ -625,6 +626,7 @@ const deleteStrategy = mutation({ await ctx.db.delete(page._id); await ctx.scheduler.runAfter(0, purgeDeletedPageOrphansRef, { pageId: page._id, + strategyId: strategy._id, }); } @@ -652,6 +654,9 @@ const deleteStrategy = mutation({ await ctx.db.delete(shareLink._id); } + await ctx.scheduler.runAfter(0, markDeletedStrategyImageAssetsRef, { + strategyId: strategy._id, + }); await ctx.db.delete(strategy._id); return { ok: true } as const; }, diff --git a/docs/cloudflare_r2_media_storage.md b/docs/cloudflare_r2_media_storage.md index 6c8f0a53..cc720b63 100644 --- a/docs/cloudflare_r2_media_storage.md +++ b/docs/cloudflare_r2_media_storage.md @@ -21,7 +21,7 @@ New uploads fail with an actionable Convex error if the required R2 env vars are 2. Convex checks editor access, inserts a pending `imageAssets` row, creates a high-entropy immutable R2 object key, and returns a short-lived signed PUT URL. 3. The client uploads bytes directly to R2 with the signed `Content-Type` header. 4. The client calls `images:completeUpload` with the upload intent metadata. -5. Convex verifies the R2 object exists, checks size and MIME metadata, marks the row active, and deletes replaced objects after the new row is active. +5. Convex verifies the R2 object exists, checks size and MIME metadata, and marks the row active. Replaced objects become durable deletion tombstones and are removed by the cleanup worker. Strategy/page/lineup payloads store image IDs and local metadata only. Public render URLs are returned from `images:listForStrategy` and `images:getAssetUrl`; they are not persisted in strategy payloads. @@ -30,9 +30,11 @@ Strategy/page/lineup payloads store image IDs and local metadata only. Public re - Expired upload URL: the client does not persist the signed URL. A retry requests a fresh pending upload intent. - MIME mismatch: `Content-Type` is signed for PUT and completion verifies R2 metadata against the file extension. - Oversized image: completion rejects and deletes the uploaded R2 object if it exceeds `R2_MAX_IMAGE_BYTES`. -- PUT succeeds but completion fails: the pending row and object key remain available for retry. `images:sweepStaleUploadsForStrategy` can delete old pending/failed objects later. -- Pending upload never completed: pending/failed rows are indexed by `uploadStatus` and `updatedAt` for sweep. +- PUT succeeds but completion fails: the pending row and object key remain available for retry. The hourly `mark-stale-image-uploads-deleted` job reclaims old pending and failed uploads without user authentication. +- Pending upload never completed: pending/failed rows are indexed by `uploadStatus` and `updatedAt`. The cron job marks them for cleanup after 24 hours. - Replacing an asset: the new immutable R2 object is activated before older active rows for the same strategy asset ID are marked deleted. +- Page and Strategy deletion: page cleanup only marks strategy-owned assets that no remaining Page or Lineup references. Strategy cleanup marks every asset with that Strategy's exact Convex ID and leaves legacy rows without a `strategyId` alone. +- R2 deletion fails: the deleted `imageAssets` row keeps its object key. The worker retries after one minute, and the hourly sweep provides a second recovery path. A 404 counts as success, so retries are safe. - Duplicate upload attempts: each upload intent gets a unique object key; completion is tied to its `uploadId`. - Legacy dev data: rows with `storageId` and no R2 provider are treated as active Convex-storage assets. - Strategy access revoked: Convex stops returning URLs to unauthorized viewers, but already-copied public custom-domain URLs can remain reachable until the object is deleted or Cloudflare access controls/cache expire.