feat(worker): delay S3 object deletion instead of removing immediately

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();

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<TaskInputs["delete-s3-object"]>);
+
+	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<TaskInputs["delete-s3-object"]>);
+
+	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<TaskInputs["delete-s3-object"]>);
+
+	expect(s3.remove).not.toBeCalled();
+});

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`);
+});

apps/worker/src/tasks/sync-post/processor.test.ts

@@ -1,5 +1,10 @@
 import processor from "./processor.ts";
-import { createJob, type TaskInputs, Tasks } from "@playfulprogramming/bullmq";
+import {
+	createJob,
+	scheduleS3ObjectDeletion,
+	type TaskInputs,
+	Tasks,
+} from "@playfulprogramming/bullmq";
 import type { Job } from "bullmq";
 import {
 	posts,
@@ -289,12 +294,12 @@ test("Deletes a post record if it no longer exists", async () => {
 		},
 	} as unknown as Job<TaskInputs["sync-post"]>);
 
-	// Assert: Post's attachments were removed from S3 before the cascading delete
-	expect(s3.remove).toBeCalledWith(
+	// Assert: Post's attachments had their S3 removal scheduled before the cascading delete
+	expect(scheduleS3ObjectDeletion).toBeCalledWith(
 		"example-bucket",
 		"posts/example-post/attachments/notes.pdf",
 	);
-	expect(s3.remove).toBeCalledWith(
+	expect(scheduleS3ObjectDeletion).toBeCalledWith(
 		"example-bucket",
 		"posts/example-post/attachments/banner.jpeg",
 	);
@@ -1143,15 +1148,15 @@ published: "2024-01-15T00:00:00Z"
 		},
 	} as unknown as Job<TaskInputs["sync-post"]>);
 
-	// Assert: attachment no longer in the repo was removed from S3
-	expect(s3.remove).toBeCalledWith(
+	// Assert: attachment no longer in the repo had its S3 removal scheduled
+	expect(scheduleS3ObjectDeletion).toBeCalledWith(
 		"example-bucket",
 		"posts/diffing-post/attachments/old-file-sha.txt",
 	);
 
-	// Assert: changed attachment's old sha-keyed object was removed, and the
-	// new sha-keyed object was uploaded in its place
-	expect(s3.remove).toBeCalledWith(
+	// Assert: changed attachment's old sha-keyed object had its removal
+	// scheduled, and the new sha-keyed object was uploaded in its place
+	expect(scheduleS3ObjectDeletion).toBeCalledWith(
 		"example-bucket",
 		"posts/diffing-post/attachments/old-changed-sha.txt",
 	);
@@ -1163,35 +1168,38 @@ published: "2024-01-15T00:00:00Z"
 		"text/plain",
 	);
 
-	// Assert: the new object is uploaded before the old one is removed, so a
-	// failed upload can't leave the persisted row pointing at a deleted key
+	// Assert: the new object is uploaded before the old one's removal is
+	// scheduled, so a failed upload can't leave the persisted row pointing at
+	// a deleted key
 	const newKeyUploadOrder = vi
 		.mocked(s3.upload)
 		.mock.calls.findIndex(
 			(call) =>
 				call[1] === "posts/diffing-post/attachments/new-changed-sha.txt",
 		);
 	const oldKeyRemoveOrder = vi
-		.mocked(s3.remove)
+		.mocked(scheduleS3ObjectDeletion)
 		.mock.calls.findIndex(
 			(call) =>
 				call[1] === "posts/diffing-post/attachments/old-changed-sha.txt",
 		);
 	expect(
 		vi.mocked(s3.upload).mock.invocationCallOrder[newKeyUploadOrder],
 	).toBeLessThan(
-		vi.mocked(s3.remove).mock.invocationCallOrder[oldKeyRemoveOrder],
+		vi.mocked(scheduleS3ObjectDeletion).mock.invocationCallOrder[
+			oldKeyRemoveOrder
+		],
 	);
 
-	// Assert: unchanged attachment was NOT re-uploaded or removed
+	// Assert: unchanged attachment was NOT re-uploaded or scheduled for removal
 	expect(s3.upload).not.toBeCalledWith(
 		"example-bucket",
 		"posts/diffing-post/attachments/unchanged-sha.txt",
 		undefined,
 		expect.anything(),
 		expect.anything(),
 	);
-	expect(s3.remove).not.toBeCalledWith(
+	expect(scheduleS3ObjectDeletion).not.toBeCalledWith(
 		"example-bucket",
 		"posts/diffing-post/attachments/unchanged-sha.txt",
 	);
@@ -1337,7 +1345,7 @@ published: "2024-01-15T00:00:00Z"
 		expect.anything(),
 		expect.anything(),
 	);
-	expect(s3.remove).not.toBeCalled();
+	expect(scheduleS3ObjectDeletion).not.toBeCalled();
 
 	// Assert: the existing row was carried forward unchanged
 	expect(insertPostAttachmentsValues).toBeCalledWith([

apps/worker/src/tasks/sync-post/processor.ts

@@ -1,5 +1,9 @@
 import { env, PostMetaSchema } from "@playfulprogramming/common";
-import { Tasks, createJob } from "@playfulprogramming/bullmq";
+import {
+	Tasks,
+	createJob,
+	scheduleS3ObjectDeletion,
+} from "@playfulprogramming/bullmq";
 import {
 	db,
 	posts,
@@ -146,8 +150,10 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => {
 				const bucket = await s3.ensureBucket(env.S3_BUCKET);
 				await Promise.all(
 					removedAttachmentRows.map(async ({ attachmentKey }) => {
-						await s3.remove(bucket, attachmentKey);
-						console.log(`Removed attachment ${attachmentKey} from S3`);
+						await scheduleS3ObjectDeletion(bucket, attachmentKey);
+						console.log(
+							`Scheduled removal of attachment ${attachmentKey} from S3`,
+						);
 					}),
 				);
 			}
@@ -286,9 +292,9 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => {
 
 	for (const [attachmentName, row] of previousAttachmentsByName) {
 		if (!discoveredAttachmentNames.has(attachmentName)) {
-			await s3.remove(bucket, row.attachmentKey);
+			await scheduleS3ObjectDeletion(bucket, row.attachmentKey);
 			console.log(
-				`Removed attachment ${row.attachmentKey} from S3 (no longer in repo)`,
+				`Scheduled removal of attachment ${row.attachmentKey} from S3 (no longer in repo)`,
 			);
 		}
 	}
@@ -356,7 +362,7 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => {
 		console.log(`Uploaded attachment ${attachmentKey} to S3`);
 
 		if (previous !== undefined) {
-			await s3.remove(bucket, previous.attachmentKey);
+			await scheduleS3ObjectDeletion(bucket, previous.attachmentKey);
 		}
 
 		attachmentRows.push({

apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.test.ts

@@ -7,6 +7,7 @@ import {
 	urlMetadataGist,
 	urlMetadataGistFile,
 } from "@playfulprogramming/db";
+import { scheduleS3ObjectDeletion } from "@playfulprogramming/bullmq";
 
 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",
+	);
+});

apps/worker/src/tasks/url-metadata/getEmbedDataFromGist.ts

@@ -3,6 +3,7 @@ import {
 	urlMetadataGist,
 	urlMetadataGistFile,
 } from "@playfulprogramming/db";
+import { scheduleS3ObjectDeletion } from "@playfulprogramming/bullmq";
 import { s3 } from "@playfulprogramming/s3";
 import { fetchAsBot } from "../../utils/fetchAsBot.ts";
 import * as github from "@playfulprogramming/github-api";
@@ -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 {

apps/worker/test-utils/setup.ts

@@ -27,6 +27,10 @@ vi.mock("@playfulprogramming/bullmq", async () => {
 		flowProducer: { add: vi.fn() },
 		createQueue: vi.fn(),
 		createJob: vi.fn(),
+		// scheduleS3ObjectDeletion 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.
+		scheduleS3ObjectDeletion: vi.fn(),
 	};
 });
 
@@ -36,6 +40,8 @@ vi.mock("@playfulprogramming/s3", () => {
 			ensureBucket: vi.fn(() => "example-bucket"),
 			upload: vi.fn(),
 			remove: vi.fn(),
+			getLastModified: vi.fn(),
+			unmodifiedSince: vi.fn(() => true),
 		},
 	};
 });

packages/bullmq/package.json

@@ -7,11 +7,16 @@
 	"scripts": {
 		"test:eslint": "eslint ./src",
 		"test:build": "publint --strict",
+		"test": "vitest run",
 		"build": "tsc --noEmit"
 	},
 	"dependencies": {
 		"@playfulprogramming/redis": "workspace:*",
+		"@playfulprogramming/s3": "workspace:*",
 		"bullmq": "catalog:",
 		"typebox": "catalog:"
+	},
+	"devDependencies": {
+		"vitest": "catalog:"
 	}
 }

packages/bullmq/src/queues.ts

@@ -35,12 +35,14 @@ export async function createJob<T extends TasksValues>(
 	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<any>;
 	await queue.add(id, data, {
 		deduplication: {
 			id: id,
 		},
+		delay: opts?.delay,
 	});
 }