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,4 +12,23 @@ createWorker(
 	Tasks.GRANT_AUTHOR_ACHIEVEMENTS,
 	"./tasks/grant-author-achievements/processor.ts",
 );
+createWorker(
+	Tasks.CLEANUP_ATTACHMENTS,
+	"./tasks/cleanup-attachments/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,141 @@
+import processor from "./processor.ts";
+import { db, postAttachments } from "@playfulprogramming/db";
+import { s3 } from "@playfulprogramming/s3";
+
+const NOW = new Date("2025-05-05T12:00:00Z");
+const ONE_HOUR_MS = 60 * 60 * 1000;
+const OUTSIDE_GRACE_PERIOD = new Date(NOW.getTime() - ONE_HOUR_MS - 1000);
+const INSIDE_GRACE_PERIOD = new Date(NOW.getTime() - 30 * 60 * 1000);
+
+test("Removes an attachment from S3 when no post_attachments row references it", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/en/content.md",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+		{
+			key: "posts/example-post/attachments/referenced-sha.pdf",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+		{
+			key: "posts/example-post/attachments/orphaned-sha.jpeg",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	vi.mocked(db.select).mockReturnValue({
+		from: vi.fn().mockResolvedValue([
+			{
+				attachmentKey: "posts/example-post/attachments/referenced-sha.pdf",
+			},
+		]),
+	} as never);
+
+	await processor({} as never);
+
+	expect(s3.remove).toBeCalledWith(
+		"example-bucket",
+		"posts/example-post/attachments/orphaned-sha.jpeg",
+	);
+	expect(s3.remove).toBeCalledTimes(1);
+});
+
+test("Leaves an attachment alone when a post_attachments row still references it", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/attachments/referenced-sha.pdf",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	vi.mocked(db.select).mockReturnValue({
+		from: vi.fn().mockResolvedValue([
+			{
+				attachmentKey: "posts/example-post/attachments/referenced-sha.pdf",
+			},
+		]),
+	} as never);
+
+	await processor({} as never);
+
+	expect(s3.remove).not.toBeCalled();
+});
+
+test("Does nothing when there are no attachment objects in S3", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/en/content.md",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	await processor({} as never);
+
+	expect(db.select).not.toBeCalled();
+	expect(s3.remove).not.toBeCalled();
+});
+
+test("Queries the full attachment table with no per-post filter", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/attachments/orphaned-sha.jpeg",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	const from = vi.fn().mockResolvedValue([]);
+	vi.mocked(db.select).mockReturnValue({ from } as never);
+
+	await processor({} as never);
+
+	expect(from).toBeCalledWith(postAttachments);
+	expect(s3.remove).toBeCalledWith(
+		"example-bucket",
+		"posts/example-post/attachments/orphaned-sha.jpeg",
+	);
+});
+
+test("Fails the job when an S3 removal rejects, rather than continuing past the error", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/attachments/orphaned-sha.jpeg",
+			lastModified: OUTSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	vi.mocked(db.select).mockReturnValue({
+		from: vi.fn().mockResolvedValue([]),
+	} as never);
+
+	const s3Error = new Error("S3 removal failed");
+	vi.mocked(s3.remove).mockRejectedValue(s3Error);
+
+	await expect(processor({} as never)).rejects.toThrow(s3Error);
+});
+
+test("Leaves an unreferenced attachment alone when it's within the grace period", async () => {
+	vi.setSystemTime(NOW);
+
+	vi.mocked(s3.list).mockResolvedValue([
+		{
+			key: "posts/example-post/attachments/fresh-sha.jpeg",
+			lastModified: INSIDE_GRACE_PERIOD,
+		},
+	]);
+
+	await processor({} as never);
+
+	// The grace period filters it out before the post_attachments query even runs
+	expect(db.select).not.toBeCalled();
+	expect(s3.remove).not.toBeCalled();
+});

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

@@ -0,0 +1,39 @@
+import { env } from "@playfulprogramming/common";
+import { Tasks } from "@playfulprogramming/bullmq";
+import { db, postAttachments } from "@playfulprogramming/db";
+import { s3 } from "@playfulprogramming/s3";
+import { createProcessor } from "../../createProcessor.ts";
+
+const ATTACHMENTS_PREFIX = "posts/";
+const GRACE_PERIOD_MS = 60 * 60 * 1000;
+
+export default createProcessor(Tasks.CLEANUP_ATTACHMENTS, async () => {
+	const bucket = await s3.ensureBucket(env.S3_BUCKET);
+
+	const objects = await s3.list(bucket, ATTACHMENTS_PREFIX);
+	const now = Date.now();
+
+	// sync-post uploads an attachment to S3 before its post_attachments row is
+	// committed, so a very recent object may just be mid-flight rather than
+	// truly orphaned - skip anything younger than the grace period.
+	const candidateKeys = objects
+		.filter((object) => object.key.includes("/attachments/"))
+		.filter((object) => now - object.lastModified.getTime() > GRACE_PERIOD_MS)
+		.map((object) => object.key);
+
+	if (candidateKeys.length === 0) return;
+
+	const referencedRows = await db
+		.select({ attachmentKey: postAttachments.attachmentKey })
+		.from(postAttachments);
+	const referencedKeys = new Set(
+		referencedRows.map((row) => row.attachmentKey),
+	);
+
+	for (const key of candidateKeys) {
+		if (referencedKeys.has(key)) continue;
+
+		await s3.remove(bucket, key);
+		console.log(`Removed unreferenced attachment ${key} from S3`);
+	}
+});

apps/worker/test-utils/setup.ts

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

packages/bullmq/src/tasks/types.ts

@@ -20,6 +20,7 @@ export const Tasks = {
 	URL_METADATA: "url-metadata",
 	POST_IMAGES: "post-images",
 	GRANT_AUTHOR_ACHIEVEMENTS: "grant-author-achievements",
+	CLEANUP_ATTACHMENTS: "cleanup-attachments",
 } as const;
 
 export type TasksKeys = keyof typeof Tasks;
@@ -33,6 +34,7 @@ export interface TaskInputs {
 	[Tasks.URL_METADATA]: UrlMetadataInput;
 	[Tasks.POST_IMAGES]: PostImageInput;
 	[Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsInput;
+	[Tasks.CLEANUP_ATTACHMENTS]: object;
 }
 
 export type TaskInputsValues = TaskInputs[TasksValues];
@@ -45,6 +47,7 @@ export interface TaskOutputs {
 	[Tasks.URL_METADATA]: UrlMetadataOutput;
 	[Tasks.POST_IMAGES]: PostImageOutput;
 	[Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsOutput;
+	[Tasks.CLEANUP_ATTACHMENTS]: void;
 }
 
 export type TaskOutputsValues = TaskOutputs[TasksValues];

packages/s3/src/utils.ts

@@ -5,6 +5,7 @@ import {
 	DeleteObjectCommand,
 	GetObjectCommand,
 	HeadObjectCommand,
+	ListObjectsV2Command,
 	NoSuchKey,
 	PutBucketPolicyCommand,
 } from "@aws-sdk/client-s3";
@@ -70,6 +71,41 @@ export async function remove(bucket: string, key: string) {
 	await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key }));
 }
 
+export interface S3Object {
+	key: string;
+	lastModified: Date;
+}
+
+export async function list(
+	bucket: string,
+	prefix: string,
+): Promise<S3Object[]> {
+	const objects: S3Object[] = [];
+	let continuationToken: string | undefined;
+
+	do {
+		const response = await client.send(
+			new ListObjectsV2Command({
+				Bucket: bucket,
+				Prefix: prefix,
+				ContinuationToken: continuationToken,
+			}),
+		);
+
+		for (const object of response.Contents ?? []) {
+			if (object.Key && object.LastModified) {
+				objects.push({ key: object.Key, lastModified: object.LastModified });
+			}
+		}
+
+		continuationToken = response.IsTruncated
+			? response.NextContinuationToken
+			: undefined;
+	} while (continuationToken);
+
+	return objects;
+}
+
 export async function matchesEtag(
 	bucket: string,
 	key: string,