diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index 5114b521..720af9bc 100644 --- a/apps/worker/src/index.ts +++ b/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), + ); diff --git a/apps/worker/src/tasks/cleanup-attachments/processor.test.ts b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts new file mode 100644 index 00000000..9ab6f2de --- /dev/null +++ b/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(); +}); diff --git a/apps/worker/src/tasks/cleanup-attachments/processor.ts b/apps/worker/src/tasks/cleanup-attachments/processor.ts new file mode 100644 index 00000000..8a17c61c --- /dev/null +++ b/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`); + } +}); diff --git a/apps/worker/test-utils/setup.ts b/apps/worker/test-utils/setup.ts index de6a3618..783a4564 100644 --- a/apps/worker/test-utils/setup.ts +++ b/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(() => []), }, }; }); diff --git a/packages/bullmq/src/tasks/types.ts b/packages/bullmq/src/tasks/types.ts index 465ff1f3..b171ae2b 100644 --- a/packages/bullmq/src/tasks/types.ts +++ b/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]; diff --git a/packages/s3/src/utils.ts b/packages/s3/src/utils.ts index 7c4eed77..d19d7f36 100644 --- a/packages/s3/src/utils.ts +++ b/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 { + 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,