konnosDashboard
log-fanout.ts507 lines · main
1import { newId } from '@briven/shared';
2import { and, eq, isNull, lt, sql } from 'drizzle-orm';
3
4import { getDb } from '../db/client.js';
5import {
6 auditLogs,
7 functionLogs,
8 projects,
9 webhookDeliveries,
10 webhookOutboundDeliveries,
11 type NewFunctionLog,
12 type ProjectTier,
13} from '../db/schema.js';
14import { getRedis } from '../lib/redis.js';
15import { log } from '../lib/logger.js';
16
17const BATCH_MAX = 500;
18const BATCH_WINDOW_MS = 1_000;
19const BLOCK_MS = 5_000;
20const RECONNECT_BACKOFF_MS = [500, 1_000, 2_500, 5_000, 10_000];
21
22/**
23 * Subscribes to every project's `logs:{projectId}` stream and fans
24 * entries out into the meta-DB `function_logs` table for durable storage
25 * and dashboard queries. Best-effort — a crashed worker may drop recent
26 * entries, which is acceptable for debug logs (audit logs use a separate
27 * pathway with stronger guarantees).
28 *
29 * Single worker per api process; Phase 3+ can split into a dedicated
30 * worker service if throughput demands.
31 */
32export function startLogFanoutWorker(): void {
33 void runLoop().catch((err: unknown) => {
34 log.error('log_fanout_worker_crashed', {
35 message: err instanceof Error ? err.message : String(err),
36 });
37 });
38}
39
40async function runLoop(): Promise<void> {
41 let attempt = 0;
42 let lastId = '$'; // start from NEW entries on fresh boot
43 while (true) {
44 const sharedRedis = getRedis();
45 if (!sharedRedis) {
46 // No redis = nothing to fan out. Sleep a minute, reconsider config.
47 await sleep(60_000);
48 continue;
49 }
50 const blocking = sharedRedis.duplicate();
51 try {
52 log.info('log_fanout_connected');
53 attempt = 0;
54 while (true) {
55 // Discover all active project streams. XREAD needs explicit stream
56 // names; Redis can't wildcard. Per project this is one SCAN cycle
57 // every BATCH_WINDOW_MS — cheap for <10k projects.
58 const streams = await scanStreams(sharedRedis);
59 if (streams.length === 0) {
60 await sleep(BATCH_WINDOW_MS);
61 continue;
62 }
63 const keys = streams.flat();
64 const ids = streams.map(() => lastId);
65
66 const reply = (await blocking.xread(
67 'COUNT',
68 BATCH_MAX,
69 'BLOCK',
70 BLOCK_MS,
71 'STREAMS',
72 ...keys,
73 ...ids,
74 )) as Array<[string, Array<[string, string[]]>]> | null;
75
76 if (!reply) continue;
77
78 const rows: NewFunctionLog[] = [];
79 for (const [key, entries] of reply) {
80 const projectId = key.slice('logs:'.length);
81 for (const [id, flat] of entries) {
82 const fields = parseFields(flat);
83 if (fields.kind !== 'invocation') continue;
84 rows.push(toRow(projectId, fields));
85 lastId = id;
86 }
87 }
88 if (rows.length > 0) {
89 await persist(rows);
90 }
91 }
92 } catch (err) {
93 log.warn('log_fanout_disconnected', {
94 message: err instanceof Error ? err.message : String(err),
95 });
96 blocking.disconnect();
97 const delay = RECONNECT_BACKOFF_MS[Math.min(attempt, RECONNECT_BACKOFF_MS.length - 1)]!;
98 attempt += 1;
99 await sleep(delay);
100 }
101 }
102}
103
104async function scanStreams(redis: ReturnType<typeof getRedis>): Promise<string[][]> {
105 if (!redis) return [];
106 const out: string[][] = [];
107 let cursor = '0';
108 do {
109 const [next, batch] = (await redis.scan(cursor, 'MATCH', 'logs:*', 'COUNT', 200)) as [
110 string,
111 string[],
112 ];
113 for (const key of batch) {
114 // Exclude subscriber counters and other non-stream keys.
115 if (key.startsWith('logs:subscribers:')) continue;
116 out.push([key]);
117 }
118 cursor = next;
119 } while (cursor !== '0');
120 return out;
121}
122
123async function persist(rows: NewFunctionLog[]): Promise<void> {
124 try {
125 await getDb().insert(functionLogs).values(rows);
126 } catch (err) {
127 log.error('log_fanout_insert_failed', {
128 count: rows.length,
129 message: err instanceof Error ? err.message : String(err),
130 });
131 }
132}
133
134function toRow(projectId: string, f: Record<string, string>): NewFunctionLog {
135 return {
136 id: newId('fn'),
137 projectId,
138 deploymentId: f.deploymentId ?? '',
139 invocationId: f.invocationId ?? '',
140 functionName: (f.functionName ?? '').slice(0, 128),
141 status: (f.status ?? 'err').slice(0, 8),
142 durationMs: (f.durationMs ?? '0').slice(0, 12),
143 touchedTables: (f.touchedTables ?? '').split(',').filter(Boolean),
144 userLogsJson: safeJson(f.logs) ?? [],
145 errCode: f.errCode ?? null,
146 errMessage: f.errMessage ?? null,
147 };
148}
149
150function safeJson(raw: string | undefined): unknown {
151 if (!raw) return null;
152 try {
153 return JSON.parse(raw);
154 } catch {
155 return null;
156 }
157}
158
159function parseFields(flat: string[]): Record<string, string> {
160 const out: Record<string, string> = {};
161 for (let i = 0; i < flat.length; i += 2) {
162 const k = flat[i];
163 const v = flat[i + 1];
164 if (typeof k === 'string' && typeof v === 'string') out[k] = v;
165 }
166 return out;
167}
168
169/** Days of function-log retention per tier. Used by `pruneOldFunctionLogs`. */
170export const RETENTION_DAYS_BY_TIER: Record<ProjectTier, number> = {
171 free: 7,
172 pro: 30,
173 team: 90,
174};
175
176/**
177 * Retention: drop `function_log` rows older than the per-tier cutoff. The
178 * `projects.tier` column is the denormalised tier; we group projects by
179 * tier (only three groups) and issue one DELETE per tier with that
180 * tier's cutoff. A project that's been soft-deleted is treated as `free`
181 * (shortest retention) — its data is on the way out anyway.
182 *
183 * Returns the total deleted-row count summed across tiers.
184 */
185export async function pruneOldFunctionLogs(): Promise<number> {
186 const db = getDb();
187 let total = 0;
188 const now = Date.now();
189
190 for (const [tier, days] of Object.entries(RETENTION_DAYS_BY_TIER) as Array<
191 [ProjectTier, number]
192 >) {
193 const cutoff = new Date(now - days * 86_400_000);
194 // Subquery: every project on this tier that isn't soft-deleted.
195 // Soft-deleted projects fall through to the `free` (shortest) branch.
196 const tierProjects = db
197 .select({ id: projects.id })
198 .from(projects)
199 .where(and(eq(projects.tier, tier), isNull(projects.deletedAt)));
200
201 const res = await db
202 .delete(functionLogs)
203 .where(
204 and(
205 lt(functionLogs.createdAt, cutoff),
206 sql`${functionLogs.projectId} IN ${tierProjects}`,
207 ),
208 )
209 .returning({ id: functionLogs.id });
210 total += res.length;
211 if (res.length > 0) {
212 log.info('function_logs_pruned', {
213 tier,
214 days,
215 count: res.length,
216 cutoff: cutoff.toISOString(),
217 });
218 }
219 }
220
221 // Catch-all: soft-deleted projects + any projects whose tier somehow
222 // doesn't match free|pro|team (data corruption hedge). Apply the
223 // shortest retention.
224 const freeCutoff = new Date(now - RETENTION_DAYS_BY_TIER.free * 86_400_000);
225 const knownTiers = sql.raw(
226 Object.keys(RETENTION_DAYS_BY_TIER).map((t) => `'${t}'`).join(','),
227 );
228 const orphans = db
229 .select({ id: projects.id })
230 .from(projects)
231 .where(sql`${projects.deletedAt} IS NOT NULL OR ${projects.tier} NOT IN (${knownTiers})`);
232 const orphanRes = await db
233 .delete(functionLogs)
234 .where(
235 and(
236 lt(functionLogs.createdAt, freeCutoff),
237 sql`${functionLogs.projectId} IN ${orphans}`,
238 ),
239 )
240 .returning({ id: functionLogs.id });
241 total += orphanRes.length;
242 if (orphanRes.length > 0) {
243 log.info('function_logs_pruned', {
244 tier: 'orphan',
245 days: RETENTION_DAYS_BY_TIER.free,
246 count: orphanRes.length,
247 cutoff: freeCutoff.toISOString(),
248 });
249 }
250
251 return total;
252}
253
254const RETENTION_TICK_MS = 6 * 60 * 60 * 1000;
255
256export function startLogRetentionCron(): void {
257 const run = (): void => {
258 void pruneOldFunctionLogs().catch((err: unknown) => {
259 log.warn('function_logs_prune_failed', {
260 message: err instanceof Error ? err.message : String(err),
261 });
262 });
263 };
264 // Run once shortly after boot (so a restart doesn't skip a whole day if
265 // the previous tick fired a few minutes before), then every 6h.
266 setTimeout(run, 30_000);
267 setInterval(run, RETENTION_TICK_MS);
268}
269
270/* ─── audit-log retention (privacy policy §5) ───────────────────────── */
271// Privacy policy commits to a 13-month audit retention window. Anything
272// older gets deleted on a daily sweep. Volume is low (a few hundred rows
273// per day at our current scale) so we don't need batched deletes; the
274// single statement runs in ms.
275
276export const AUDIT_RETENTION_DAYS = 30 * 13; // 13 months ≈ 390 days
277
278/**
279 * Cutoff date for audit-log retention based on `now`. Extracted so the
280 * boundary logic is unit-testable without standing up the DB.
281 */
282export function auditRetentionCutoff(nowMs: number): Date {
283 return new Date(nowMs - AUDIT_RETENTION_DAYS * 86_400_000);
284}
285
286export async function pruneOldAuditLogs(): Promise<number> {
287 const db = getDb();
288 const cutoff = auditRetentionCutoff(Date.now());
289 const res = await db
290 .delete(auditLogs)
291 .where(lt(auditLogs.createdAt, cutoff))
292 .returning({ id: auditLogs.id });
293 if (res.length > 0) {
294 log.info('audit_logs_pruned', {
295 count: res.length,
296 retentionDays: AUDIT_RETENTION_DAYS,
297 cutoff: cutoff.toISOString(),
298 });
299 }
300 return res.length;
301}
302
303// 24h between runs. Audit volume is low so checking more often wastes
304// db round-trips for no operational benefit.
305const AUDIT_RETENTION_TICK_MS = 24 * 60 * 60 * 1000;
306
307export function startAuditRetentionCron(): void {
308 const run = (): void => {
309 void pruneOldAuditLogs().catch((err: unknown) => {
310 log.warn('audit_logs_prune_failed', {
311 message: err instanceof Error ? err.message : String(err),
312 });
313 });
314 };
315 // 60s after boot (small offset from function-logs cron at 30s so a
316 // fresh boot doesn't fire both simultaneously), then daily.
317 setTimeout(run, 60_000);
318 setInterval(run, AUDIT_RETENTION_TICK_MS);
319}
320
321/* ─── webhook-delivery retention ────────────────────────────────────── */
322// webhook_deliveries grows one row per inbound POST. A leaked endpoint
323// can amplify that into a real cost without bounded retention. We reuse
324// RETENTION_DAYS_BY_TIER (same scheme as function_logs) — short enough
325// to keep the table small, long enough for operators to debug yesterday's
326// failed signatures.
327
328export async function pruneOldWebhookDeliveries(): Promise<number> {
329 const db = getDb();
330 let total = 0;
331 const now = Date.now();
332
333 for (const [tier, days] of Object.entries(RETENTION_DAYS_BY_TIER) as Array<
334 [ProjectTier, number]
335 >) {
336 const cutoff = new Date(now - days * 86_400_000);
337 const tierProjects = db
338 .select({ id: projects.id })
339 .from(projects)
340 .where(and(eq(projects.tier, tier), isNull(projects.deletedAt)));
341
342 const res = await db
343 .delete(webhookDeliveries)
344 .where(
345 and(
346 lt(webhookDeliveries.createdAt, cutoff),
347 sql`${webhookDeliveries.projectId} IN ${tierProjects}`,
348 ),
349 )
350 .returning({ id: webhookDeliveries.id });
351 total += res.length;
352 if (res.length > 0) {
353 log.info('webhook_deliveries_pruned', {
354 tier,
355 days,
356 count: res.length,
357 cutoff: cutoff.toISOString(),
358 });
359 }
360 }
361
362 // Catch-all for soft-deleted / unknown-tier projects. Same shortest-
363 // retention treatment as function_logs.
364 const freeCutoff = new Date(now - RETENTION_DAYS_BY_TIER.free * 86_400_000);
365 const knownTiers = sql.raw(
366 Object.keys(RETENTION_DAYS_BY_TIER)
367 .map((t) => `'${t}'`)
368 .join(','),
369 );
370 const orphans = db
371 .select({ id: projects.id })
372 .from(projects)
373 .where(sql`${projects.deletedAt} IS NOT NULL OR ${projects.tier} NOT IN (${knownTiers})`);
374 const orphanRes = await db
375 .delete(webhookDeliveries)
376 .where(
377 and(
378 lt(webhookDeliveries.createdAt, freeCutoff),
379 sql`${webhookDeliveries.projectId} IN ${orphans}`,
380 ),
381 )
382 .returning({ id: webhookDeliveries.id });
383 total += orphanRes.length;
384 if (orphanRes.length > 0) {
385 log.info('webhook_deliveries_pruned', {
386 tier: 'orphan',
387 days: RETENTION_DAYS_BY_TIER.free,
388 count: orphanRes.length,
389 cutoff: freeCutoff.toISOString(),
390 });
391 }
392
393 return total;
394}
395
396const WEBHOOK_DELIVERIES_RETENTION_TICK_MS = 6 * 60 * 60 * 1000;
397
398export function startWebhookDeliveriesRetentionCron(): void {
399 const run = (): void => {
400 void pruneOldWebhookDeliveries().catch((err: unknown) => {
401 log.warn('webhook_deliveries_prune_failed', {
402 message: err instanceof Error ? err.message : String(err),
403 });
404 });
405 };
406 // 90s after boot — offset from the other retention crons so a fresh
407 // boot doesn't fire all three simultaneously and hold connection
408 // pool capacity that the api request path also needs.
409 setTimeout(run, 90_000);
410 setInterval(run, WEBHOOK_DELIVERIES_RETENTION_TICK_MS);
411}
412
413/* ─── outbound-delivery retention ───────────────────────────────────── */
414// webhook_outbound_deliveries grows under retry storms — each event
415// emission inserts one row per subscriber, and a failing endpoint can
416// keep a row pending for ~30 min while attempts back off. Same per-tier
417// cutoff as function_logs + inbound deliveries (free=7d, pro=30d,
418// team=90d). Terminal rows (ok/failed/cancelled) prune by created_at;
419// pending rows are spared regardless of age — they're still on the
420// dispatcher's retry path until terminal.
421
422export async function pruneOldOutboundWebhookDeliveries(): Promise<number> {
423 const db = getDb();
424 let total = 0;
425 const now = Date.now();
426
427 for (const [tier, days] of Object.entries(RETENTION_DAYS_BY_TIER) as Array<
428 [ProjectTier, number]
429 >) {
430 const cutoff = new Date(now - days * 86_400_000);
431 const tierProjects = db
432 .select({ id: projects.id })
433 .from(projects)
434 .where(and(eq(projects.tier, tier), isNull(projects.deletedAt)));
435 const res = await db
436 .delete(webhookOutboundDeliveries)
437 .where(
438 and(
439 lt(webhookOutboundDeliveries.createdAt, cutoff),
440 sql`${webhookOutboundDeliveries.projectId} IN ${tierProjects}`,
441 sql`${webhookOutboundDeliveries.status} <> 'pending'`,
442 ),
443 )
444 .returning({ id: webhookOutboundDeliveries.id });
445 total += res.length;
446 if (res.length > 0) {
447 log.info('outbound_deliveries_pruned', {
448 tier,
449 days,
450 count: res.length,
451 cutoff: cutoff.toISOString(),
452 });
453 }
454 }
455
456 // Orphan sweep — soft-deleted projects + unknown tiers, shortest cutoff.
457 const freeCutoff = new Date(now - RETENTION_DAYS_BY_TIER.free * 86_400_000);
458 const knownTiers = sql.raw(
459 Object.keys(RETENTION_DAYS_BY_TIER)
460 .map((t) => `'${t}'`)
461 .join(','),
462 );
463 const orphans = db
464 .select({ id: projects.id })
465 .from(projects)
466 .where(sql`${projects.deletedAt} IS NOT NULL OR ${projects.tier} NOT IN (${knownTiers})`);
467 const orphanRes = await db
468 .delete(webhookOutboundDeliveries)
469 .where(
470 and(
471 lt(webhookOutboundDeliveries.createdAt, freeCutoff),
472 sql`${webhookOutboundDeliveries.projectId} IN ${orphans}`,
473 sql`${webhookOutboundDeliveries.status} <> 'pending'`,
474 ),
475 )
476 .returning({ id: webhookOutboundDeliveries.id });
477 total += orphanRes.length;
478 if (orphanRes.length > 0) {
479 log.info('outbound_deliveries_pruned', {
480 tier: 'orphan',
481 days: RETENTION_DAYS_BY_TIER.free,
482 count: orphanRes.length,
483 cutoff: freeCutoff.toISOString(),
484 });
485 }
486 return total;
487}
488
489const OUTBOUND_RETENTION_TICK_MS = 6 * 60 * 60 * 1000;
490
491export function startOutboundWebhookDeliveriesRetentionCron(): void {
492 const run = (): void => {
493 void pruneOldOutboundWebhookDeliveries().catch((err: unknown) => {
494 log.warn('outbound_deliveries_prune_failed', {
495 message: err instanceof Error ? err.message : String(err),
496 });
497 });
498 };
499 // 105s after boot — offset from the inbound retention cron (90s) so
500 // the two retention sweeps don't race for the same connection pool.
501 setTimeout(run, 105_000);
502 setInterval(run, OUTBOUND_RETENTION_TICK_MS);
503}
504
505function sleep(ms: number): Promise<void> {
506 return new Promise((resolve) => setTimeout(resolve, ms));
507}