diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 5114b521..89d250e3 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -12,4 +12,5 @@ createWorker( Tasks.GRANT_AUTHOR_ACHIEVEMENTS, "./tasks/grant-author-achievements/processor.ts", ); +createWorker(Tasks.DELETE_S3_OBJECT, "./tasks/delete-s3-object/processor.ts"); createHealthcheck(); diff --git a/apps/worker/src/tasks/delete-s3-object/processor.test.ts b/apps/worker/src/tasks/delete-s3-object/processor.test.ts new file mode 100644 index 00000000..63d43f3d --- /dev/null +++ b/apps/worker/src/tasks/delete-s3-object/processor.test.ts @@ -0,0 +1,55 @@ +import processor from "./processor.ts"; +import type { TaskInputs } from "@playfulprogramming/bullmq"; +import type { Job } from "bullmq"; +import { s3 } from "@playfulprogramming/s3"; + +test("removes the object when no lastModified was captured at scheduling time", async () => { + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + }, + } as unknown as Job); + + expect(s3.unmodifiedSince).not.toBeCalled(); + expect(s3.remove).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + ); +}); + +test("removes the object when it hasn't been modified since scheduling", async () => { + vi.mocked(s3.unmodifiedSince).mockResolvedValueOnce(true); + + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + lastModified: "2025-05-05T00:00:00.000Z", + }, + } as unknown as Job); + + expect(s3.unmodifiedSince).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + new Date("2025-05-05T00:00:00.000Z"), + ); + expect(s3.remove).toBeCalledWith( + "example-bucket", + "posts/example-post/attachments/notes.pdf", + ); +}); + +test("skips removal when the object was rewritten since scheduling", async () => { + vi.mocked(s3.unmodifiedSince).mockResolvedValueOnce(false); + + await processor({ + data: { + bucket: "example-bucket", + key: "posts/example-post/attachments/notes.pdf", + lastModified: "2025-05-05T00:00:00.000Z", + }, + } as unknown as Job); + + expect(s3.remove).not.toBeCalled(); +}); diff --git a/apps/worker/src/tasks/delete-s3-object/processor.ts b/apps/worker/src/tasks/delete-s3-object/processor.ts new file mode 100644 index 00000000..c9c9fffb --- /dev/null +++ b/apps/worker/src/tasks/delete-s3-object/processor.ts @@ -0,0 +1,25 @@ +import { Tasks } from "@playfulprogramming/bullmq"; +import { s3 } from "@playfulprogramming/s3"; +import { createProcessor } from "../../createProcessor.ts"; + +export default createProcessor(Tasks.DELETE_S3_OBJECT, async (job) => { + const { bucket, key, lastModified } = job.data; + + if (lastModified !== undefined) { + const stillUnmodified = await s3.unmodifiedSince( + bucket, + key, + new Date(lastModified), + ); + + if (!stillUnmodified) { + console.log( + `Skipped removal of ${bucket}/${key} - object was rewritten since deletion was scheduled`, + ); + return; + } + } + + await s3.remove(bucket, key); + console.log(`Removed ${bucket}/${key} from S3 after grace period`); +}); diff --git a/apps/worker/src/tasks/sync-post/processor.ts b/apps/worker/src/tasks/sync-post/processor.ts index 0ebb8a88..6a75a1e7 100644 --- a/apps/worker/src/tasks/sync-post/processor.ts +++ b/apps/worker/src/tasks/sync-post/processor.ts @@ -237,6 +237,9 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => { // Phase 3: Discover, resize, diff, and upload post attachments // ========================================================================= const attachmentRows: AttachmentRow[] = []; + // Note: this can pick up attachments from different branches + // Attachments are only keyed by post/sha, so an unchanged attachment will + // reference the same record const existingAttachmentRecords = await db .select({ attachmentKey: attachments.attachmentKey }) .from(attachments) diff --git a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts index 59a92f44..e30c7ae6 100644 --- a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts +++ b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts @@ -7,6 +7,7 @@ import { urlMetadataGist, urlMetadataGistFile, } from "@playfulprogramming/db"; +import { scheduleS3ObjectDeletion } from "../../utils/scheduleS3ObjectDeletion.ts"; test("fetches the expected information for a successful gist response", async () => { const gistUrl = new URL( @@ -59,3 +60,36 @@ test("fetches the expected information for a successful gist response", async () language: "text", }); }); + +test("schedules S3 removal for gist files that were deleted from the gist", async () => { + const gistUrl = new URL( + "https://gist.github.com/crutchcorn/36fe5553219c05ea38bacf1c7396085b", + ); + + (getGistById as Mock).mockReturnValueOnce( + Promise.resolve({ + description: "This is a description of the gist.", + files: {}, + }), + ); + + ( + db.delete(urlMetadataGistFile).where(undefined).returning as Mock + ).mockReturnValueOnce(Promise.resolve([{ filename: "old-file.txt" }])); + + const result = await getEmbedDataFromGist( + gistUrl, + new AbortController().signal, + ); + expect(result).toEqual({ + error: false, + gistId: "36fe5553219c05ea38bacf1c7396085b", + }); + + // Assert: S3 removal was scheduled (not performed immediately), keyed the + // same way getFileKey derives it - a hash of the filename under the gist's ID + expect(scheduleS3ObjectDeletion).toBeCalledWith( + "example-bucket", + "remote-gist/36fe5553219c05ea38bacf1c7396085b/775d94f3d7c5ee0d18ee08d4b65152b5", + ); +}); diff --git a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts index d56118de..d28879dd 100644 --- a/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts +++ b/apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts @@ -5,6 +5,7 @@ import { } from "@playfulprogramming/db"; import { s3 } from "@playfulprogramming/s3"; import { fetchAsBot } from "../../utils/fetchAsBot.ts"; +import { scheduleS3ObjectDeletion } from "../../utils/scheduleS3ObjectDeletion.ts"; import * as github from "@playfulprogramming/github-api"; import { and, eq, inArray, not } from "drizzle-orm"; import { type EmbedData, BUCKET } from "./common.ts"; @@ -103,9 +104,10 @@ export async function getEmbedDataFromGist( }); }); - // Clean up deleted files from S3 + // Schedule cleanup of deleted files from S3, after a grace period so any + // in-flight or cached request for the old key doesn't 404 immediately for (const { filename } of deletedFilesResult) { - await s3.remove(BUCKET, getFileKey(filename)); + await scheduleS3ObjectDeletion(BUCKET, getFileKey(filename)); } return { diff --git a/apps/worker/src/utils/scheduleS3ObjectDeletion.test.ts b/apps/worker/src/utils/scheduleS3ObjectDeletion.test.ts new file mode 100644 index 00000000..22dfca31 --- /dev/null +++ b/apps/worker/src/utils/scheduleS3ObjectDeletion.test.ts @@ -0,0 +1,33 @@ +import { scheduleS3ObjectDeletion } from "./scheduleS3ObjectDeletion.ts"; +import { enqueueS3ObjectDeletion } from "@playfulprogramming/bullmq"; +import { s3 } from "@playfulprogramming/s3"; + +// This module is mocked wholesale in test-utils/setup.ts for every other +// test file's benefit (they only care that scheduling happened, not how) - +// undo that here so this file exercises the real implementation. +vi.unmock("./scheduleS3ObjectDeletion.ts"); + +test("skips scheduling and warns when lastModified can't be determined", async () => { + vi.mocked(s3.getLastModified).mockResolvedValueOnce(undefined); + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + expect(enqueueS3ObjectDeletion).not.toBeCalled(); + expect(warnSpy).toBeCalledWith( + expect.stringContaining("posts/example/notes.pdf"), + ); +}); + +test("passes the object's lastModified through to enqueueS3ObjectDeletion", async () => { + const lastModified = new Date("2026-01-01T00:00:00.000Z"); + vi.mocked(s3.getLastModified).mockResolvedValueOnce(lastModified); + + await scheduleS3ObjectDeletion("example-bucket", "posts/example/notes.pdf"); + + expect(enqueueS3ObjectDeletion).toBeCalledWith( + "example-bucket", + "posts/example/notes.pdf", + lastModified, + ); +}); diff --git a/apps/worker/src/utils/scheduleS3ObjectDeletion.ts b/apps/worker/src/utils/scheduleS3ObjectDeletion.ts new file mode 100644 index 00000000..6b5e5365 --- /dev/null +++ b/apps/worker/src/utils/scheduleS3ObjectDeletion.ts @@ -0,0 +1,25 @@ +import { s3 } from "@playfulprogramming/s3"; +import { enqueueS3ObjectDeletion } from "@playfulprogramming/bullmq"; + +export async function scheduleS3ObjectDeletion( + bucket: string, + key: string, +): Promise { + const lastModified = await s3.getLastModified(bucket, key); + + if (lastModified === undefined) { + // Without a LastModified to check at execution time, the processor + // would have no way to detect a rewrite during the grace period and + // would unconditionally delete whatever's at this key 24h from now - + // including a legitimate new upload. Bail out instead of scheduling + // an unsafe deletion, but log it: a genuine transient failure to read + // the object's metadata here means this object never gets scheduled + // for cleanup at all, so it'd otherwise leak in S3 with no trace. + console.warn( + `Skipped scheduling deletion of ${bucket}/${key} - could not read its LastModified`, + ); + return; + } + + await enqueueS3ObjectDeletion(bucket, key, lastModified); +} diff --git a/apps/worker/test-utils/setup.ts b/apps/worker/test-utils/setup.ts index b98f9978..f190cce3 100644 --- a/apps/worker/test-utils/setup.ts +++ b/apps/worker/test-utils/setup.ts @@ -27,15 +27,25 @@ vi.mock("@playfulprogramming/bullmq", async () => { flowProducer: { add: vi.fn() }, createQueue: vi.fn(), createJob: vi.fn(), + // enqueueS3ObjectDeletion calls the real createJob internally via a + // relative import, which bypasses the createJob mock above - it needs + // its own override so tests don't hit a real BullMQ queue/Redis. + enqueueS3ObjectDeletion: vi.fn(), }; }); +vi.mock("../src/utils/scheduleS3ObjectDeletion.ts", () => ({ + scheduleS3ObjectDeletion: vi.fn(), +})); + vi.mock("@playfulprogramming/s3", () => { return { s3: { ensureBucket: vi.fn(() => "example-bucket"), upload: vi.fn(), remove: vi.fn(), + getLastModified: vi.fn(), + unmodifiedSince: vi.fn(() => true), }, }; }); diff --git a/packages/bullmq/package.json b/packages/bullmq/package.json index db8d2502..fc5a5a92 100644 --- a/packages/bullmq/package.json +++ b/packages/bullmq/package.json @@ -7,11 +7,15 @@ "scripts": { "test:eslint": "eslint ./src", "test:build": "publint --strict", + "test": "vitest run", "build": "tsc --noEmit" }, "dependencies": { "@playfulprogramming/redis": "workspace:*", "bullmq": "catalog:", "typebox": "catalog:" + }, + "devDependencies": { + "vitest": "catalog:" } } diff --git a/packages/bullmq/src/queues.ts b/packages/bullmq/src/queues.ts index 3de3695e..8f89f793 100644 --- a/packages/bullmq/src/queues.ts +++ b/packages/bullmq/src/queues.ts @@ -35,6 +35,7 @@ export async function createJob( task: T, id: string, data: TaskInputs[T], + opts?: { delay?: number }, ) { // eslint-disable-next-line @typescript-eslint/no-explicit-any const queue = createQueue(task) as Queue; @@ -42,5 +43,6 @@ export async function createJob( deduplication: { id: id, }, + delay: opts?.delay, }); } diff --git a/packages/bullmq/src/tasks/delete-s3-object.test.ts b/packages/bullmq/src/tasks/delete-s3-object.test.ts new file mode 100644 index 00000000..f5e33aa8 --- /dev/null +++ b/packages/bullmq/src/tasks/delete-s3-object.test.ts @@ -0,0 +1,46 @@ +import { enqueueS3ObjectDeletion } from "./delete-s3-object.ts"; +import { createJob } from "../queues.ts"; + +vi.mock("../queues.ts", () => ({ + createJob: vi.fn(), +})); + +afterEach(() => { + vi.clearAllMocks(); +}); + +function jobIdFromCall(callIndex: number): string { + return vi.mocked(createJob).mock.calls[callIndex][1] as string; +} + +test("reuses the same job id when lastModified is unchanged across calls", async () => { + const lastModified = new Date("2026-01-01T00:00:00.000Z"); + + await enqueueS3ObjectDeletion( + "example-bucket", + "posts/example/notes.pdf", + lastModified, + ); + await enqueueS3ObjectDeletion( + "example-bucket", + "posts/example/notes.pdf", + lastModified, + ); + + expect(jobIdFromCall(0)).toEqual(jobIdFromCall(1)); +}); + +test("uses a different job id when lastModified changes between calls", async () => { + await enqueueS3ObjectDeletion( + "example-bucket", + "posts/example/notes.pdf", + new Date("2026-01-01T00:00:00.000Z"), + ); + await enqueueS3ObjectDeletion( + "example-bucket", + "posts/example/notes.pdf", + new Date("2026-01-02T00:00:00.000Z"), + ); + + expect(jobIdFromCall(0)).not.toEqual(jobIdFromCall(1)); +}); diff --git a/packages/bullmq/src/tasks/delete-s3-object.ts b/packages/bullmq/src/tasks/delete-s3-object.ts new file mode 100644 index 00000000..c0c72fd7 --- /dev/null +++ b/packages/bullmq/src/tasks/delete-s3-object.ts @@ -0,0 +1,38 @@ +import { createJob } from "../queues.ts"; +import { Tasks } from "./types.ts"; + +export interface DeleteS3ObjectInput { + bucket: string; + key: string; + // ISO timestamp of the object's LastModified at scheduling time. The + // processor re-checks this before deleting, so a key that gets rewritten + // in the meantime (even with byte-identical content, which leaves its + // ETag unchanged) doesn't get deleted out from under its new reference. + lastModified: string; +} + +export type DeleteS3ObjectOutput = void; + +// Grace period before a scheduled S3 deletion actually runs, so the frontend +// or CDN doesn't hit a 404 for a key it just fetched or cached. +export const DELETE_S3_OBJECT_GRACE_PERIOD_MS = 24 * 60 * 60 * 1000; + +export async function enqueueS3ObjectDeletion( + bucket: string, + key: string, + lastModified: Date, +): Promise { + const lastModifiedIso = lastModified.toISOString(); + + // The job ID includes a generation marker (the object's LastModified) so + // that scheduling a deletion for a key that's since been rewritten gets + // its own job instead of silently deduplicating against - and being + // dropped in favor of - a still-pending job for the previous generation + // of that key. + await createJob( + Tasks.DELETE_S3_OBJECT, + `delete-s3-object:${bucket}:${key}:${lastModifiedIso}`, + { bucket, key, lastModified: lastModifiedIso }, + { delay: DELETE_S3_OBJECT_GRACE_PERIOD_MS }, + ); +} diff --git a/packages/bullmq/src/tasks/index.ts b/packages/bullmq/src/tasks/index.ts index 1f56e3ef..d47e91b8 100644 --- a/packages/bullmq/src/tasks/index.ts +++ b/packages/bullmq/src/tasks/index.ts @@ -1,3 +1,4 @@ +export * from "./delete-s3-object.ts"; export * from "./grant-author-achievements.ts"; export * from "./post-image.ts"; export * from "./sync-all.ts"; diff --git a/packages/bullmq/src/tasks/types.ts b/packages/bullmq/src/tasks/types.ts index 465ff1f3..efadacc7 100644 --- a/packages/bullmq/src/tasks/types.ts +++ b/packages/bullmq/src/tasks/types.ts @@ -11,6 +11,10 @@ import type { GrantAuthorAchievementsInput, GrantAuthorAchievementsOutput, } from "./grant-author-achievements.ts"; +import type { + DeleteS3ObjectInput, + DeleteS3ObjectOutput, +} from "./delete-s3-object.ts"; export const Tasks = { SYNC_ALL: "sync-all", @@ -20,6 +24,7 @@ export const Tasks = { URL_METADATA: "url-metadata", POST_IMAGES: "post-images", GRANT_AUTHOR_ACHIEVEMENTS: "grant-author-achievements", + DELETE_S3_OBJECT: "delete-s3-object", } as const; export type TasksKeys = keyof typeof Tasks; @@ -33,6 +38,7 @@ export interface TaskInputs { [Tasks.URL_METADATA]: UrlMetadataInput; [Tasks.POST_IMAGES]: PostImageInput; [Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsInput; + [Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectInput; } export type TaskInputsValues = TaskInputs[TasksValues]; @@ -45,6 +51,7 @@ export interface TaskOutputs { [Tasks.URL_METADATA]: UrlMetadataOutput; [Tasks.POST_IMAGES]: PostImageOutput; [Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsOutput; + [Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectOutput; } export type TaskOutputsValues = TaskOutputs[TasksValues]; diff --git a/packages/bullmq/tsconfig.json b/packages/bullmq/tsconfig.json index f27b9ce9..42296d79 100644 --- a/packages/bullmq/tsconfig.json +++ b/packages/bullmq/tsconfig.json @@ -1,4 +1,7 @@ { "extends": "../../tsconfig.json", + "compilerOptions": { + "types": ["vitest/globals"] + }, "include": ["src", "eslint.config.mjs"] } diff --git a/packages/bullmq/vitest.config.ts b/packages/bullmq/vitest.config.ts new file mode 100644 index 00000000..076c92fe --- /dev/null +++ b/packages/bullmq/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + globals: true, + }, +}); diff --git a/packages/s3/src/utils.ts b/packages/s3/src/utils.ts index 7c4eed77..3c40c9f0 100644 --- a/packages/s3/src/utils.ts +++ b/packages/s3/src/utils.ts @@ -85,6 +85,39 @@ export async function matchesEtag( } } +export async function getLastModified( + bucket: string, + key: string, +): Promise { + try { + const obj = await client.send( + new HeadObjectCommand({ Bucket: bucket, Key: key }), + ); + return obj.LastModified; + } catch (_e) { + return undefined; + } +} + +export async function unmodifiedSince( + bucket: string, + key: string, + since: Date, +): Promise { + try { + const obj = await client.send( + new HeadObjectCommand({ + Bucket: bucket, + Key: key, + IfUnmodifiedSince: since, + }), + ); + return !!obj; + } catch (_e) { + return false; + } +} + export async function upload( bucket: string, key: string, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 72487e8c..c59117a4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -272,6 +272,10 @@ importers: typebox: specifier: 'catalog:' version: 1.1.38 + devDependencies: + vitest: + specifier: 'catalog:' + version: 4.1.7(@types/node@25.9.1)(@vitest/coverage-v8@4.1.7)(vite@8.0.13(@types/node@25.9.1)(jiti@2.7.0)(yaml@2.9.0)) packages/common: dependencies: