From 89ee1c5a21afb8759bf76d69191d9ddb77b20789 Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Mon, 13 Jul 2026 07:33:16 -0700 Subject: [PATCH 1/3] feat(worker): add cleanup task for unreferenced S3 attachments Sweeps S3 objects under posts/*/attachments/* on a repeatable interval and removes any with no matching post_attachments row, catching attachments orphaned by interrupted sync-post jobs. --- apps/worker/src/index.ts | 21 +++++- .../cleanup-attachments/processor.test.ts | 71 +++++++++++++++++++ .../tasks/cleanup-attachments/processor.ts | 32 +++++++++ apps/worker/test-utils/setup.ts | 1 + packages/bullmq/src/tasks/types.ts | 3 + packages/s3/src/utils.ts | 26 +++++++ 6 files changed, 153 insertions(+), 1 deletion(-) create mode 100644 apps/worker/src/tasks/cleanup-attachments/processor.test.ts create mode 100644 apps/worker/src/tasks/cleanup-attachments/processor.ts 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..f69502c5 --- /dev/null +++ b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts @@ -0,0 +1,71 @@ +import processor from "./processor.ts"; +import { db, postAttachments } from "@playfulprogramming/db"; +import { s3 } from "@playfulprogramming/s3"; + +test("Removes an attachment from S3 when no post_attachments row references it", async () => { + vi.mocked(s3.list).mockResolvedValue([ + "posts/example-post/en/content.md", + "posts/example-post/attachments/referenced-sha.pdf", + "posts/example-post/attachments/orphaned-sha.jpeg", + ]); + + 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.mocked(s3.list).mockResolvedValue([ + "posts/example-post/attachments/referenced-sha.pdf", + ]); + + 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.mocked(s3.list).mockResolvedValue(["posts/example-post/en/content.md"]); + + 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.mocked(s3.list).mockResolvedValue([ + "posts/example-post/attachments/orphaned-sha.jpeg", + ]); + + 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", + ); +}); 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..8449e925 --- /dev/null +++ b/apps/worker/src/tasks/cleanup-attachments/processor.ts @@ -0,0 +1,32 @@ +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/"; + +export default createProcessor(Tasks.CLEANUP_ATTACHMENTS, async () => { + const bucket = await s3.ensureBucket(env.S3_BUCKET); + + const objectKeys = await s3.list(bucket, ATTACHMENTS_PREFIX); + const attachmentKeys = objectKeys.filter((key) => + key.includes("/attachments/"), + ); + + if (attachmentKeys.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 attachmentKeys) { + 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..b594b288 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]: object; } export type TaskOutputsValues = TaskOutputs[TasksValues]; diff --git a/packages/s3/src/utils.ts b/packages/s3/src/utils.ts index 7c4eed77..a95fcd64 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,31 @@ export async function remove(bucket: string, key: string) { await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key })); } +export async function list(bucket: string, prefix: string): Promise { + const keys: string[] = []; + 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) keys.push(object.Key); + } + + continuationToken = response.IsTruncated + ? response.NextContinuationToken + : undefined; + } while (continuationToken); + + return keys; +} + export async function matchesEtag( bucket: string, key: string, From 76e8422f59d49b142e62c4b63b99c9570af584fd Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Mon, 13 Jul 2026 08:00:50 -0700 Subject: [PATCH 2/3] feat(worker): add grace period to unreferenced attachment cleanup sync-post uploads an attachment to S3 before its post_attachments row commits, so a very recent orphan-looking object may just be mid-flight. S3's list now returns lastModified alongside each key, and the cleanup task skips anything younger than a one-hour grace period. --- .../cleanup-attachments/processor.test.ts | 62 +++++++++++++++++-- .../tasks/cleanup-attachments/processor.ts | 19 ++++-- packages/s3/src/utils.ts | 18 ++++-- 3 files changed, 83 insertions(+), 16 deletions(-) diff --git a/apps/worker/src/tasks/cleanup-attachments/processor.test.ts b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts index f69502c5..9a347c13 100644 --- a/apps/worker/src/tasks/cleanup-attachments/processor.test.ts +++ b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts @@ -2,11 +2,27 @@ 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([ - "posts/example-post/en/content.md", - "posts/example-post/attachments/referenced-sha.pdf", - "posts/example-post/attachments/orphaned-sha.jpeg", + { + 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({ @@ -27,8 +43,13 @@ test("Removes an attachment from S3 when no post_attachments row references it", }); test("Leaves an attachment alone when a post_attachments row still references it", async () => { + vi.setSystemTime(NOW); + vi.mocked(s3.list).mockResolvedValue([ - "posts/example-post/attachments/referenced-sha.pdf", + { + key: "posts/example-post/attachments/referenced-sha.pdf", + lastModified: OUTSIDE_GRACE_PERIOD, + }, ]); vi.mocked(db.select).mockReturnValue({ @@ -45,7 +66,14 @@ test("Leaves an attachment alone when a post_attachments row still references it }); test("Does nothing when there are no attachment objects in S3", async () => { - vi.mocked(s3.list).mockResolvedValue(["posts/example-post/en/content.md"]); + vi.setSystemTime(NOW); + + vi.mocked(s3.list).mockResolvedValue([ + { + key: "posts/example-post/en/content.md", + lastModified: OUTSIDE_GRACE_PERIOD, + }, + ]); await processor({} as never); @@ -54,8 +82,13 @@ test("Does nothing when there are no attachment objects in S3", async () => { }); test("Queries the full attachment table with no per-post filter", async () => { + vi.setSystemTime(NOW); + vi.mocked(s3.list).mockResolvedValue([ - "posts/example-post/attachments/orphaned-sha.jpeg", + { + key: "posts/example-post/attachments/orphaned-sha.jpeg", + lastModified: OUTSIDE_GRACE_PERIOD, + }, ]); const from = vi.fn().mockResolvedValue([]); @@ -69,3 +102,20 @@ test("Queries the full attachment table with no per-post filter", async () => { "posts/example-post/attachments/orphaned-sha.jpeg", ); }); + +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 index 8449e925..8a17c61c 100644 --- a/apps/worker/src/tasks/cleanup-attachments/processor.ts +++ b/apps/worker/src/tasks/cleanup-attachments/processor.ts @@ -5,16 +5,23 @@ 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 objectKeys = await s3.list(bucket, ATTACHMENTS_PREFIX); - const attachmentKeys = objectKeys.filter((key) => - key.includes("/attachments/"), - ); + 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 (attachmentKeys.length === 0) return; + if (candidateKeys.length === 0) return; const referencedRows = await db .select({ attachmentKey: postAttachments.attachmentKey }) @@ -23,7 +30,7 @@ export default createProcessor(Tasks.CLEANUP_ATTACHMENTS, async () => { referencedRows.map((row) => row.attachmentKey), ); - for (const key of attachmentKeys) { + for (const key of candidateKeys) { if (referencedKeys.has(key)) continue; await s3.remove(bucket, key); diff --git a/packages/s3/src/utils.ts b/packages/s3/src/utils.ts index a95fcd64..d19d7f36 100644 --- a/packages/s3/src/utils.ts +++ b/packages/s3/src/utils.ts @@ -71,8 +71,16 @@ export async function remove(bucket: string, key: string) { await client.send(new DeleteObjectCommand({ Bucket: bucket, Key: key })); } -export async function list(bucket: string, prefix: string): Promise { - const keys: string[] = []; +export interface S3Object { + key: string; + lastModified: Date; +} + +export async function list( + bucket: string, + prefix: string, +): Promise { + const objects: S3Object[] = []; let continuationToken: string | undefined; do { @@ -85,7 +93,9 @@ export async function list(bucket: string, prefix: string): Promise { ); for (const object of response.Contents ?? []) { - if (object.Key) keys.push(object.Key); + if (object.Key && object.LastModified) { + objects.push({ key: object.Key, lastModified: object.LastModified }); + } } continuationToken = response.IsTruncated @@ -93,7 +103,7 @@ export async function list(bucket: string, prefix: string): Promise { : undefined; } while (continuationToken); - return keys; + return objects; } export async function matchesEtag( From ca598c657a169779af5cb537747adfb2c400c5b7 Mon Sep 17 00:00:00 2001 From: Brian Bornino Date: Mon, 13 Jul 2026 08:21:41 -0700 Subject: [PATCH 3/3] fix(bullmq): correct CLEANUP_ATTACHMENTS output type to void TaskOutputs[Tasks.CLEANUP_ATTACHMENTS] was typed as object but the processor returns Promise, breaking the build. Also adds a test locking in that an S3 removal failure rejects the job instead of being swallowed. --- .../cleanup-attachments/processor.test.ts | 20 +++++++++++++++++++ packages/bullmq/src/tasks/types.ts | 2 +- 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/apps/worker/src/tasks/cleanup-attachments/processor.test.ts b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts index 9a347c13..9ab6f2de 100644 --- a/apps/worker/src/tasks/cleanup-attachments/processor.test.ts +++ b/apps/worker/src/tasks/cleanup-attachments/processor.test.ts @@ -103,6 +103,26 @@ test("Queries the full attachment table with no per-post filter", async () => { ); }); +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); diff --git a/packages/bullmq/src/tasks/types.ts b/packages/bullmq/src/tasks/types.ts index b594b288..b171ae2b 100644 --- a/packages/bullmq/src/tasks/types.ts +++ b/packages/bullmq/src/tasks/types.ts @@ -47,7 +47,7 @@ export interface TaskOutputs { [Tasks.URL_METADATA]: UrlMetadataOutput; [Tasks.POST_IMAGES]: PostImageOutput; [Tasks.GRANT_AUTHOR_ACHIEVEMENTS]: GrantAuthorAchievementsOutput; - [Tasks.CLEANUP_ATTACHMENTS]: object; + [Tasks.CLEANUP_ATTACHMENTS]: void; } export type TaskOutputsValues = TaskOutputs[TasksValues];