feat(worker): add cleanup task for unreferenced S3 attachments

apps/worker/src/index.ts

@@ -1,6 +1,6 @@
 import { createHealthcheck } from "./createHealthcheck.ts";
 import { createWorker } from "./createWorker.ts";
-import { Tasks } from "@playfulprogramming/bullmq";
+import { Tasks, createQueue } from "@playfulprogramming/bullmq";
 
 createWorker(Tasks.POST_IMAGES, "./tasks/post-images/processor.ts");
 createWorker(Tasks.URL_METADATA, "./tasks/url-metadata/processor.ts");
@@ -12,5 +12,24 @@ createWorker(
 	Tasks.GRANT_AUTHOR_ACHIEVEMENTS,
 	"./tasks/grant-author-achievements/processor.ts",
 );
+createWorker(
+	Tasks.CLEANUP_ATTACHMENTS,
+	"./tasks/cleanup-attachments/processor.ts",
+);
 createWorker(Tasks.DELETE_S3_OBJECT, "./tasks/delete-s3-object/processor.ts");
 createHealthcheck();
+
+// Repeatable job: BullMQ dedupes repeatable schedulers by name + repeat
+// options, so re-registering this on every worker restart is a no-op rather
+// than creating duplicate schedules.
+const CLEANUP_ATTACHMENTS_INTERVAL_MS = 24 * 60 * 60 * 1000;
+
+createQueue(Tasks.CLEANUP_ATTACHMENTS)
+	.add(
+		Tasks.CLEANUP_ATTACHMENTS,
+		{},
+		{ repeat: { every: CLEANUP_ATTACHMENTS_INTERVAL_MS } },
+	)
+	.catch((err) =>
+		console.error("Failed to schedule cleanup-attachments job:", err),
+	);

apps/worker/src/tasks/cleanup-attachments/processor.test.ts

@@ -0,0 +1,129 @@
+import processor from "./processor.ts";
+import { db, attachments } from "@playfulprogramming/db";
+import { s3 } from "@playfulprogramming/s3";
+import { lt } from "drizzle-orm";
+import type * as DrizzleOrm from "drizzle-orm";
+
+vi.mock("drizzle-orm", async (importOriginal) => {
+	const actual = await importOriginal<typeof DrizzleOrm>();
+	return { ...actual, lt: vi.fn(actual.lt) };
+});
+
+const NOW = new Date("2025-05-05T12:00:00Z");
+
+const deleteAttachmentReturning = db
+	.delete(attachments)
+	.where(expect.anything()).returning;
+const insertAttachmentValues = db.insert(attachments).values;
+const insertAttachmentOnConflictDoUpdate = db
+	.insert(attachments)
+	.values(expect.anything()).onConflictDoUpdate;
+
+test("Removes an attachment returned by the delete query from S3", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(deleteAttachmentReturning)
+		.mockResolvedValueOnce([
+			{
+				attachmentKey: "posts/example-post/attachments/orphaned-sha.jpeg",
+				sha: "orphaned-sha",
+				width: 100,
+				height: 100,
+			},
+		])
+		.mockResolvedValueOnce([]);
+
+	await processor({} as never);
+
+	expect(s3.remove).toHaveBeenCalledWith(
+		"example-bucket",
+		"posts/example-post/attachments/orphaned-sha.jpeg",
+	);
+	expect(s3.remove).toHaveBeenCalledTimes(1);
+});
+
+test("Uses a one-hour-old cutoff for the lastModified staleness check", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(deleteAttachmentReturning).mockResolvedValueOnce([]);
+
+	await processor({} as never);
+
+	expect(lt).toHaveBeenCalledWith(
+		attachments.lastModified,
+		new Date(NOW.getTime() - 60 * 60 * 1000),
+	);
+});
+
+test("Does nothing when the delete query returns no rows", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(deleteAttachmentReturning).mockResolvedValueOnce([]);
+
+	await processor({} as never);
+
+	expect(s3.remove).not.toHaveBeenCalled();
+});
+
+test("Keeps deleting and removing until the delete query returns no more rows", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(deleteAttachmentReturning)
+		.mockResolvedValueOnce([
+			{
+				attachmentKey: "posts/example-post/attachments/first.jpeg",
+				sha: "first",
+				width: 1,
+				height: 1,
+			},
+		])
+		.mockResolvedValueOnce([
+			{
+				attachmentKey: "posts/example-post/attachments/second.jpeg",
+				sha: "second",
+				width: 1,
+				height: 1,
+			},
+		])
+		.mockResolvedValueOnce([]);
+
+	await processor({} as never);
+
+	expect(s3.remove).toHaveBeenNthCalledWith(
+		1,
+		"example-bucket",
+		"posts/example-post/attachments/first.jpeg",
+	);
+	expect(s3.remove).toHaveBeenNthCalledWith(
+		2,
+		"example-bucket",
+		"posts/example-post/attachments/second.jpeg",
+	);
+	expect(s3.remove).toHaveBeenCalledTimes(2);
+});
+
+test("Re-inserts the row and fails the job when S3 removal rejects, rather than leaking the object untracked", async () => {
+	vi.setSystemTime(NOW);
+
+	const orphan = {
+		attachmentKey: "posts/example-post/attachments/orphaned-sha.jpeg",
+		sha: "orphaned-sha",
+		width: 100,
+		height: 100,
+	};
+	vi.mocked(deleteAttachmentReturning).mockResolvedValueOnce([orphan]);
+
+	const s3Error = new Error("S3 removal failed");
+	vi.mocked(s3.remove).mockRejectedValue(s3Error);
+
+	await expect(processor({} as never)).rejects.toThrow(s3Error);
+
+	expect(insertAttachmentValues).toHaveBeenCalledWith({
+		...orphan,
+		lastModified: expect.any(Date),
+	});
+	expect(insertAttachmentOnConflictDoUpdate).toHaveBeenCalledWith({
+		target: attachments.attachmentKey,
+		set: { lastModified: expect.any(Date) },
+	});
+});

apps/worker/src/tasks/cleanup-attachments/processor.ts

@@ -0,0 +1,79 @@
+import { env } from "@playfulprogramming/common";
+import { Tasks } from "@playfulprogramming/bullmq";
+import { db, attachments, postAttachments } from "@playfulprogramming/db";
+import { s3 } from "@playfulprogramming/s3";
+import { and, eq, inArray, lt, notExists } from "drizzle-orm";
+import { createProcessor } from "../../createProcessor.ts";
+
+const GRACE_PERIOD_MS = 60 * 60 * 1000;
+
+export default createProcessor(
+	Tasks.CLEANUP_ATTACHMENTS,
+	async (_job, { signal }) => {
+		const bucket = await s3.ensureBucket(env.S3_BUCKET);
+		const staleBefore = new Date(Date.now() - GRACE_PERIOD_MS);
+
+		while (!signal.aborted) {
+			// sync-post uploads an attachment to S3 and inserts its attachments row
+			// before the corresponding post_attachments row commits, so a very
+			// recent row may just be mid-flight rather than truly orphaned - skip
+			// anything younger than the grace period. Safety against a
+			// concurrently-committing post_attachments insert doesn't come from
+			// this being a single SQL statement - it comes from the
+			// post_attachments -> attachments foreign key (onDelete: "cascade" in
+			// posts.ts): inserting a post_attachments row takes a lock on the
+			// referenced attachments row, which serializes against this DELETE. If
+			// that FK is ever dropped, this safety goes with it.
+			const candidateKey = db
+				.select({ attachmentKey: attachments.attachmentKey })
+				.from(attachments)
+				.where(
+					and(
+						notExists(
+							db
+								.select({ attachmentKey: postAttachments.attachmentKey })
+								.from(postAttachments)
+								.where(
+									eq(postAttachments.attachmentKey, attachments.attachmentKey),
+								),
+						),
+						lt(attachments.lastModified, staleBefore),
+					),
+				)
+				.limit(1);
+
+			const [deleted] = await db
+				.delete(attachments)
+				.where(inArray(attachments.attachmentKey, candidateKey))
+				.returning({
+					attachmentKey: attachments.attachmentKey,
+					sha: attachments.sha,
+					width: attachments.width,
+					height: attachments.height,
+				});
+
+			if (!deleted) return;
+
+			try {
+				await s3.remove(bucket, deleted.attachmentKey);
+			} catch (err) {
+				// The row is already claimed/deleted, so a failed removal here would
+				// otherwise permanently lose track of the S3 object - nothing would
+				// ever find it again. Re-insert it (with a fresh lastModified) so the
+				// next scheduled run picks it back up, then fail the job as a whole.
+				await db
+					.insert(attachments)
+					.values({ ...deleted, lastModified: new Date() })
+					.onConflictDoUpdate({
+						target: attachments.attachmentKey,
+						set: { lastModified: new Date() },
+					});
+				throw err;
+			}
+
+			console.log(
+				`Removed unreferenced attachment ${deleted.attachmentKey} from S3`,
+			);
+		}
+	},
+);

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

@@ -29,6 +29,9 @@ const selectPreviousAuthors = db
 const insertPostReturning = db
 	.insert(posts)
 	.values(expect.anything()).returning;
+const insertAttachmentOnConflictDoUpdate = db
+	.insert(attachments)
+	.values(expect.anything()).onConflictDoUpdate;
 
 test("Syncs a standalone post successfully", async () => {
 	const postId = ":test-post-uuid:";
@@ -700,14 +703,20 @@ published: "2024-01-15T00:00:00Z"
 		sha: "notes-sha",
 		width: null,
 		height: null,
+		lastModified: expect.any(Date),
 	});
 	expect(db.insert(attachments).values).toHaveBeenCalledWith({
 		attachmentKey: "posts/attachment-post/attachments/banner-sha.jpeg",
 		sha: "banner-sha",
 		width: 1,
 		height: 1,
+		lastModified: expect.any(Date),
 	});
 	expect(db.insert(attachments).values).toHaveBeenCalledTimes(2);
+	expect(insertAttachmentOnConflictDoUpdate).toHaveBeenCalledWith({
+		target: attachments.attachmentKey,
+		set: { lastModified: expect.any(Date) },
+	});
 
 	expect(db.insert(postAttachments).values).toHaveBeenCalledExactlyOnceWith([
 		{
@@ -913,6 +922,7 @@ published: "2024-01-15T00:00:00Z"
 		sha: "new-changed-sha",
 		width: null,
 		height: null,
+		lastModified: expect.any(Date),
 	});
 	expect(db.insert(attachments).values).toHaveBeenCalledTimes(1);
 

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

@@ -310,15 +310,24 @@ export default createProcessor(Tasks.SYNC_POST, async (job, { signal }) => {
 		);
 		console.log(`Uploaded attachment ${attachmentKey} to S3`);
 
+		// onConflictDoUpdate (not DoNothing) so a conflicting insert - e.g. the
+		// same sha reused across posts/branches - still refreshes lastModified.
+		// Otherwise a stale timestamp could let the cleanup sweep delete this row
+		// in the narrow window before its new post_attachments reference commits.
+		const attachmentLastModified = new Date();
 		await db
 			.insert(attachments)
 			.values({
 				attachmentKey,
 				sha,
 				width,
 				height,
+				lastModified: attachmentLastModified,
 			})
-			.onConflictDoNothing();
+			.onConflictDoUpdate({
+				target: attachments.attachmentKey,
+				set: { lastModified: attachmentLastModified },
+			});
 
 		attachmentRows.push({
 			attachmentKey,

apps/worker/test-utils/setup.ts

@@ -44,6 +44,7 @@ vi.mock("@playfulprogramming/s3", () => {
 			ensureBucket: vi.fn(() => "example-bucket"),
 			upload: vi.fn(),
 			remove: vi.fn(),
+			list: vi.fn(() => []),
 			getLastModified: vi.fn(),
 			unmodifiedSince: vi.fn(() => true),
 		},

packages/bullmq/src/tasks/types.ts

@@ -24,6 +24,7 @@ export const Tasks = {
 	URL_METADATA: "url-metadata",
 	POST_IMAGES: "post-images",
 	GRANT_AUTHOR_ACHIEVEMENTS: "grant-author-achievements",
+	CLEANUP_ATTACHMENTS: "cleanup-attachments",
 	DELETE_S3_OBJECT: "delete-s3-object",
 } as const;
 
@@ -38,6 +39,7 @@ export interface TaskInputs {
 	[Tasks.URL_METADATA]: UrlMetadataInput;
 	[Tasks.POST_IMAGES]: PostImageInput;
 	[Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsInput;
+	[Tasks.CLEANUP_ATTACHMENTS]: object;
 	[Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectInput;
 }
 
@@ -51,6 +53,7 @@ export interface TaskOutputs {
 	[Tasks.URL_METADATA]: UrlMetadataOutput;
 	[Tasks.POST_IMAGES]: PostImageOutput;
 	[Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsOutput;
+	[Tasks.CLEANUP_ATTACHMENTS]: void;
 	[Tasks.DELETE_S3_OBJECT]: DeleteS3ObjectOutput;
 }
 

packages/db/drizzle/20260807133839_sudden_micromacro/migration.sql

@@ -0,0 +1 @@
+ALTER TABLE "attachments" ADD COLUMN "last_modified" timestamp with time zone DEFAULT now() NOT NULL;