konnosDashboard
storage-janitor.ts248 lines · main
1import { and, inArray, isNotNull, lt } from 'drizzle-orm';
2
3import { getDb } from '../db/client.js';
4import { projectFiles } from '../db/schema.js';
5import { env } from '../env.js';
6import { listObjects, presignS3Url } from '../lib/s3-presign.js';
7import { log } from '../lib/logger.js';
8
9/**
10 * Storage orphan janitor. Soft-deleted `project_files` rows expect their
11 * MinIO object to be deleted at soft-delete time (services/storage.ts
12 * deleteFile path), but if that synchronous DELETE failed (network blip,
13 * minio outage, transient 5xx) the object lingers forever — storage
14 * leak in proportion to the failure rate × time.
15 *
16 * Every JANITOR_TICK_MS the worker:
17 * 1. selects up to BATCH_SIZE rows where deleted_at < now - 1h (the
18 * 1h grace gives the synchronous delete time to succeed or fail
19 * visibly before the janitor steps in).
20 * 2. for each: signs a DELETE against MinIO and fires it. MinIO's
21 * DELETE is idempotent — 204 whether the object existed or not —
22 * so a no-op retry is cheap. Any 5xx is logged and the row gets
23 * another chance on the next tick.
24 *
25 * What this DOESN'T cover: objects in MinIO whose row got cascade-
26 * deleted via project FK (hard-delete of a project removes its
27 * project_files rows but not the underlying objects). That requires a
28 * bucket LIST + reconcile path, which is a separate slice — the
29 * sigv4 helper only signs PUT/GET/DELETE today.
30 */
31
32const JANITOR_TICK_MS = 6 * 60 * 60 * 1000;
33const BATCH_SIZE = 200;
34const GRACE_MS = 60 * 60 * 1000; // 1h after deleted_at before we sweep
35
36// Orphan reconcile pass: how long an unreferenced object can survive
37// before we delete it. 7d gives anybody mid-upload time to commit the
38// row (or recover from a crash) without losing data.
39const ORPHAN_GRACE_MS = 7 * 24 * 60 * 60 * 1000;
40const ORPHAN_PAGE_SIZE = 500;
41const ORPHAN_MAX_PAGES = 20; // cap walk at 10k objects per tick
42
43let timer: ReturnType<typeof setInterval> | null = null;
44let inflight = false;
45
46interface StorageEnv {
47 endpoint: string;
48 region: string;
49 bucket: string;
50 accessKey: string;
51 secretKey: string;
52}
53
54function storageEnv(): StorageEnv | null {
55 if (
56 !env.BRIVEN_MINIO_ENDPOINT ||
57 !env.BRIVEN_MINIO_ACCESS_KEY ||
58 !env.BRIVEN_MINIO_SECRET_KEY
59 ) {
60 return null;
61 }
62 return {
63 endpoint: env.BRIVEN_MINIO_ENDPOINT,
64 region: env.BRIVEN_MINIO_REGION ?? 'us-east-1',
65 bucket: env.BRIVEN_MINIO_BUCKET ?? 'briven',
66 accessKey: env.BRIVEN_MINIO_ACCESS_KEY,
67 secretKey: env.BRIVEN_MINIO_SECRET_KEY,
68 };
69}
70
71async function tick(): Promise<void> {
72 if (inflight) {
73 log.warn('storage_janitor_tick_skipped_inflight');
74 return;
75 }
76 const cfg = storageEnv();
77 if (!cfg) return; // storage not configured — silent no-op
78 inflight = true;
79 try {
80 await softDeletePass(cfg);
81 await orphanReconcilePass(cfg);
82 } catch (err) {
83 log.error('storage_janitor_tick_failed', {
84 message: err instanceof Error ? err.message : String(err),
85 });
86 } finally {
87 inflight = false;
88 }
89}
90
91/**
92 * Pass 1: rows with deleted_at set but the MinIO object may still exist
93 * (synchronous delete failed silently at soft-delete time). Retry the
94 * DELETE; idempotent so a no-op retry is cheap.
95 */
96async function softDeletePass(cfg: StorageEnv): Promise<void> {
97 const cutoff = new Date(Date.now() - GRACE_MS);
98 const db = getDb();
99 const rows = await db
100 .select()
101 .from(projectFiles)
102 .where(and(isNotNull(projectFiles.deletedAt), lt(projectFiles.deletedAt, cutoff)))
103 .limit(BATCH_SIZE);
104 if (rows.length === 0) return;
105 log.info('storage_janitor_soft_delete_pass', { candidateCount: rows.length });
106
107 let purged = 0;
108 let failed = 0;
109 await Promise.all(
110 rows.map(async (row) => {
111 const ok = await purgeObject(cfg, row.objectKey);
112 if (ok) purged += 1;
113 else failed += 1;
114 }),
115 );
116 if (purged > 0 || failed > 0) {
117 log.info('storage_janitor_soft_delete_done', { purged, failed });
118 }
119}
120
121/**
122 * Pass 2: bucket-walk reconcile. project_files rows can disappear via
123 * FK cascade when a project gets hard-deleted, leaving objects with no
124 * row pointing at them. We list the bucket, check each object key
125 * against project_files, and delete objects that have no matching row
126 * AND were last modified more than ORPHAN_GRACE_MS ago.
127 *
128 * Capped at ORPHAN_MAX_PAGES per tick — operators who somehow have >10k
129 * orphans should grep `storage_janitor_orphan_*` logs and run a manual
130 * cleanup rather than wait for 24 hours of janitor ticks.
131 */
132async function orphanReconcilePass(cfg: StorageEnv): Promise<void> {
133 const orphanCutoff = new Date(Date.now() - ORPHAN_GRACE_MS);
134 const db = getDb();
135 let startAfter: string | undefined;
136 let pagesWalked = 0;
137 let totalOrphans = 0;
138
139 for (;;) {
140 if (pagesWalked >= ORPHAN_MAX_PAGES) {
141 log.warn('storage_janitor_orphan_page_cap_hit', { pagesWalked, totalOrphans });
142 break;
143 }
144 let result;
145 try {
146 result = await listObjects({
147 endpoint: cfg.endpoint,
148 region: cfg.region,
149 bucket: cfg.bucket,
150 prefix: 'projects/',
151 maxKeys: ORPHAN_PAGE_SIZE,
152 startAfter,
153 accessKey: cfg.accessKey,
154 secretKey: cfg.secretKey,
155 });
156 } catch (err) {
157 log.warn('storage_janitor_orphan_list_failed', {
158 err: err instanceof Error ? err.message : String(err),
159 });
160 return;
161 }
162 pagesWalked += 1;
163
164 // Only consider objects past the grace window — fresh uploads
165 // might still be mid-commit.
166 const candidates = result.objects.filter((o) => o.lastModified < orphanCutoff);
167 if (candidates.length > 0) {
168 const keys = candidates.map((o) => o.key);
169 const matched = await db
170 .select({ objectKey: projectFiles.objectKey })
171 .from(projectFiles)
172 .where(inArray(projectFiles.objectKey, keys));
173 const matchedSet = new Set(matched.map((r) => r.objectKey));
174 const orphans = candidates.filter((o) => !matchedSet.has(o.key));
175 if (orphans.length > 0) {
176 log.info('storage_janitor_orphans_detected', {
177 page: pagesWalked,
178 count: orphans.length,
179 });
180 await Promise.all(
181 orphans.map(async (o) => {
182 const ok = await purgeObject(cfg, o.key);
183 if (ok) totalOrphans += 1;
184 }),
185 );
186 }
187 }
188
189 if (!result.isTruncated || !result.nextStartAfter) break;
190 startAfter = result.nextStartAfter;
191 }
192
193 if (totalOrphans > 0) {
194 log.info('storage_janitor_orphan_done', { totalOrphans, pagesWalked });
195 }
196}
197
198async function purgeObject(cfg: StorageEnv, objectKey: string): Promise<boolean> {
199 const url = presignS3Url({
200 endpoint: cfg.endpoint,
201 region: cfg.region,
202 bucket: cfg.bucket,
203 key: objectKey,
204 method: 'DELETE',
205 accessKey: cfg.accessKey,
206 secretKey: cfg.secretKey,
207 expiresIn: 60,
208 });
209 try {
210 const res = await fetch(url, { method: 'DELETE', signal: AbortSignal.timeout(10_000) });
211 if (res.status === 204 || res.status === 404) return true;
212 log.warn('storage_janitor_object_not_purged', { objectKey, status: res.status });
213 return false;
214 } catch (err) {
215 log.warn('storage_janitor_object_delete_failed', {
216 objectKey,
217 message: err instanceof Error ? err.message : String(err),
218 });
219 return false;
220 }
221}
222
223export function startStorageJanitor(): void {
224 if (timer) return;
225 if (!env.BRIVEN_DATABASE_URL) {
226 log.warn('storage_janitor_skipped_no_db');
227 return;
228 }
229 // 120s after boot — last in the worker startup cascade (schedule 45,
230 // webhook-retention 90, outbound-retention 105, janitor 120). Keeps
231 // the boot-time connection-pool burst smooth.
232 setTimeout(() => {
233 void tick();
234 timer = setInterval(() => {
235 void tick();
236 }, JANITOR_TICK_MS);
237 }, 120_000).unref?.();
238 log.info('storage_janitor_armed', { tickMs: JANITOR_TICK_MS, batch: BATCH_SIZE });
239}
240
241export function stopStorageJanitor(): void {
242 if (timer) {
243 clearInterval(timer);
244 timer = null;
245 }
246}
247
248export const _internals = { tick, purgeObject };