feat(insights): harden investigations, revenue, and delivery

.env.example

@@ -9,6 +9,9 @@ BULLMQ_REDIS_URL="redis://localhost:6379"
 INSIGHTS_PORT="4002"
 INSIGHTS_BULLMQ_REDIS_URL=""
 INSIGHTS_DISPATCH_INTERVAL_MS="300000"
+INSIGHTS_SCHEDULED_DISPATCH_ENABLED="false"
+# Comma-separated organization IDs for canaries; use * only after rollout.
+INSIGHTS_SCHEDULED_ORGANIZATION_IDS=""
 INSIGHTS_MAINTENANCE_INTERVAL_MS="300000"
 INSIGHTS_STALE_ITEM_MS="900000"
 INSIGHTS_WORKER_CONCURRENCY="2"

.github/workflows/ci.yml

@@ -43,8 +43,6 @@ jobs:
     name: Check Types
     runs-on: blacksmith-4vcpu-ubuntu-2404
     timeout-minutes: 15
-    env:
-      CLICKHOUSE_READONLY_URL: ${{ secrets.CLICKHOUSE_READONLY_URL }}
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
@@ -61,8 +59,12 @@ jobs:
       - name: ClickHouse types match DDL
         run: bun run --cwd packages/db ch:check
       - name: ClickHouse schema matches cluster
-        if: env.CLICKHOUSE_READONLY_URL != ''
-        run: bun run --cwd packages/db ch:verify
+        env:
+          CLICKHOUSE_READONLY_URL: ${{ secrets.CLICKHOUSE_READONLY_URL }}
+        run: |
+          if [[ -n "$CLICKHOUSE_READONLY_URL" ]]; then
+            bun run --cwd packages/db ch:verify
+          fi
 
   test:
     name: Test
@@ -127,10 +129,12 @@ jobs:
         ports:
           - 8123:8123
         options: >-
+          --ulimit nofile=262144:262144
           --health-cmd "clickhouse-client --query 'SELECT 1'"
           --health-interval 10s
           --health-timeout 5s
-          --health-retries 5
+          --health-start-period 30s
+          --health-retries 12
 
     steps:
       - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
@@ -158,6 +162,8 @@ jobs:
         run: bun run --cwd packages/db clickhouse:init
       - name: Test
         env:
+          CLICKHOUSE_INTEGRATION_TESTS: "true"
+          CLICKHOUSE_URL: http://default:@localhost:8123
           NODE_ENV: test
         run: bun run test
       - name: Insights integration

apps/api/src/billing/autumn-webhook-replay.test.ts

@@ -0,0 +1,136 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const state = vi.hoisted(() => ({ error: vi.fn(), warn: vi.fn() }));
+
+vi.mock("evlog", () => ({ log: { error: state.error, warn: state.warn } }));
+vi.mock("@/routes/webhooks/autumn", () => ({
+	replayDeferredAutumnWebhooks: vi.fn(async () => ({
+		completed: 0,
+		deadLettered: 0,
+		deferred: 0,
+		failed: [],
+	})),
+}));
+vi.mock("@/routes/webhooks/autumn-inbox", () => ({
+	deleteCompletedAutumnWebhooks: vi.fn(async () => 0),
+	deleteDeadLetterAutumnWebhooks: vi.fn(async () => 0),
+	listUnalertedAutumnWebhookDeadLetters: vi.fn(async () => []),
+	markAutumnWebhookDeadLettersAlerted: vi.fn(async () => 0),
+}));
+
+import { startAutumnWebhookReplayLoop } from "./autumn-webhook-replay";
+import { runAutumnWebhookMaintenance } from "./autumn-webhook-replay";
+import { replayDeferredAutumnWebhooks } from "@/routes/webhooks/autumn";
+import {
+	deleteCompletedAutumnWebhooks,
+	deleteDeadLetterAutumnWebhooks,
+	listUnalertedAutumnWebhookDeadLetters,
+	markAutumnWebhookDeadLettersAlerted,
+} from "@/routes/webhooks/autumn-inbox";
+
+beforeEach(() => {
+	vi.useFakeTimers();
+	state.error.mockClear();
+	state.warn.mockClear();
+	vi.mocked(deleteCompletedAutumnWebhooks).mockClear();
+	vi.mocked(deleteDeadLetterAutumnWebhooks).mockClear();
+	vi.mocked(listUnalertedAutumnWebhookDeadLetters).mockClear();
+	vi.mocked(markAutumnWebhookDeadLettersAlerted).mockClear();
+	vi.mocked(replayDeferredAutumnWebhooks).mockClear();
+});
+
+afterEach(() => {
+	vi.useRealTimers();
+});
+
+describe("Autumn webhook replay loop", () => {
+	it("runs bounded replay and retention maintenance in one shot", async () => {
+		await expect(runAutumnWebhookMaintenance()).resolves.toEqual({
+			completed: 0,
+			deadLettered: 0,
+			deadLetters: 0,
+			deferred: 0,
+			deleted: 0,
+			failed: [],
+		});
+		expect(replayDeferredAutumnWebhooks).toHaveBeenCalledWith(100);
+		expect(deleteCompletedAutumnWebhooks).toHaveBeenCalledWith({ limit: 100 });
+		expect(deleteDeadLetterAutumnWebhooks).toHaveBeenCalledWith({ limit: 100 });
+	});
+
+	it("reports item-level replay failures that remain queued", async () => {
+		vi.mocked(replayDeferredAutumnWebhooks).mockResolvedValueOnce({
+			completed: 0,
+			deadLettered: 0,
+			deferred: 0,
+			failed: ["msg-failed"],
+		});
+
+		await runAutumnWebhookMaintenance();
+
+		expect(state.warn).toHaveBeenCalledWith(
+			expect.objectContaining({
+				component: "autumn_webhook_replay",
+				failed_count: 1,
+			})
+		);
+	});
+
+	it("alerts once for newly dead-lettered webhooks before retention", async () => {
+		vi.mocked(listUnalertedAutumnWebhookDeadLetters).mockResolvedValueOnce([
+			{
+				attempts: 12,
+				deadLetteredAt: new Date("2026-07-01T00:00:00.000Z"),
+				errorMessage: "provider unavailable",
+				id: "msg-dead",
+				type: "balances.limit_reached",
+			},
+		]);
+
+		await runAutumnWebhookMaintenance();
+
+		expect(state.error).toHaveBeenCalledWith(
+			expect.objectContaining({
+				component: "autumn_webhook_replay",
+				dead_letter_count: 1,
+				dead_letter_ids: ["msg-dead"],
+			})
+		);
+		expect(markAutumnWebhookDeadLettersAlerted).toHaveBeenCalledWith([
+			"msg-dead",
+		]);
+	});
+
+	it("runs immediately, repeats every minute, and stops deterministically", async () => {
+		const maintenance = vi.fn(async () => undefined);
+		const loop = startAutumnWebhookReplayLoop(maintenance);
+
+		await loop.run();
+		expect(maintenance).toHaveBeenCalledTimes(1);
+		await vi.advanceTimersByTimeAsync(60_000);
+		expect(maintenance).toHaveBeenCalledTimes(2);
+
+		loop.stop();
+		await vi.advanceTimersByTimeAsync(120_000);
+		expect(maintenance).toHaveBeenCalledTimes(2);
+	});
+
+	it("logs maintenance failures without rejecting or stopping the loop", async () => {
+		const maintenance = vi.fn(async () => {
+			throw new Error("database unavailable");
+		});
+		const loop = startAutumnWebhookReplayLoop(maintenance);
+
+		await expect(loop.run()).resolves.toBeUndefined();
+		expect(state.error).toHaveBeenCalledWith(
+			expect.objectContaining({
+				component: "autumn_webhook_replay",
+				error_message: "database unavailable",
+			})
+		);
+
+		await vi.advanceTimersByTimeAsync(60_000);
+		expect(maintenance).toHaveBeenCalledTimes(2);
+		loop.stop();
+	});
+});

apps/api/src/billing/autumn-webhook-replay.ts

@@ -0,0 +1,103 @@
+import { log } from "evlog";
+import {
+	deleteCompletedAutumnWebhooks,
+	deleteDeadLetterAutumnWebhooks,
+	listUnalertedAutumnWebhookDeadLetters,
+	markAutumnWebhookDeadLettersAlerted,
+} from "@/routes/webhooks/autumn-inbox";
+import { replayDeferredAutumnWebhooks } from "@/routes/webhooks/autumn";
+
+const REPLAY_INTERVAL_MS = 60_000;
+const REPLAY_BATCH_SIZE = 100;
+
+export interface AutumnWebhookReplayLoop {
+	run(): Promise<void>;
+	stop(): void;
+}
+
+export async function runAutumnWebhookMaintenance(): Promise<{
+	completed: number;
+	deadLettered: number;
+	deadLetters: number;
+	deferred: number;
+	deleted: number;
+	failed: string[];
+}> {
+	const replay = await replayDeferredAutumnWebhooks(REPLAY_BATCH_SIZE);
+	if (replay.failed.length > 0) {
+		log.warn({
+			service: "api",
+			component: "autumn_webhook_replay",
+			message: "Autumn webhook replay batch had failures",
+			failed_count: replay.failed.length,
+		});
+	}
+
+	const deadLetters = await listUnalertedAutumnWebhookDeadLetters({
+		limit: REPLAY_BATCH_SIZE,
+	});
+	if (deadLetters.length > 0) {
+		log.error({
+			service: "api",
+			component: "autumn_webhook_replay",
+			dead_letter_count: deadLetters.length,
+			dead_letter_ids: deadLetters.map((row) => row.id),
+			error_message: "Autumn webhooks exhausted replay attempts",
+		});
+		await markAutumnWebhookDeadLettersAlerted(deadLetters.map((row) => row.id));
+	}
+
+	const deletedRows = await Promise.all([
+		deleteCompletedAutumnWebhooks({ limit: REPLAY_BATCH_SIZE }),
+		deleteDeadLetterAutumnWebhooks({ limit: REPLAY_BATCH_SIZE }),
+	]);
+	return {
+		...replay,
+		deadLetters: deadLetters.length,
+		deleted: deletedRows.reduce((total, count) => total + count, 0),
+	};
+}
+
+export function startAutumnWebhookReplayLoop(
+	maintenance: () => Promise<unknown> = runAutumnWebhookMaintenance
+): AutumnWebhookReplayLoop {
+	let active: Promise<void> | null = null;
+	let stopped = false;
+
+	const run = (): Promise<void> => {
+		if (stopped) {
+			return Promise.resolve();
+		}
+		if (active) {
+			return active;
+		}
+		active = Promise.resolve()
+			.then(maintenance)
+			.then(() => undefined)
+			.catch((error) => {
+				log.error({
+					service: "api",
+					component: "autumn_webhook_replay",
+					error_message: error instanceof Error ? error.message : String(error),
+				});
+			})
+			.finally(() => {
+				active = null;
+			});
+		return active;
+	};
+
+	run().catch(() => undefined);
+	const timer = setInterval(() => {
+		run().catch(() => undefined);
+	}, REPLAY_INTERVAL_MS);
+	timer.unref?.();
+
+	return {
+		run,
+		stop: () => {
+			stopped = true;
+			clearInterval(timer);
+		},
+	};
+}

apps/api/src/bootstrap/shutdown.ts

@@ -15,12 +15,12 @@ export function warmPostgresPool() {
 	);
 }
 
-export function registerShutdownHooks() {
-	process.on("SIGINT", () => shutdownApi("SIGINT"));
-	process.on("SIGTERM", () => shutdownApi("SIGTERM"));
+export function registerShutdownHooks(beforeShutdown?: () => void) {
+	process.on("SIGINT", () => shutdownApi("SIGINT", beforeShutdown));
+	process.on("SIGTERM", () => shutdownApi("SIGTERM", beforeShutdown));
 }
 
-async function shutdownApi(signal: string) {
+async function shutdownApi(signal: string, beforeShutdown?: () => void) {
 	if (shuttingDown) {
 		log.info({
 			lifecycle: "shutdown",
@@ -31,6 +31,7 @@ async function shutdownApi(signal: string) {
 	}
 
 	shuttingDown = true;
+	beforeShutdown?.();
 	const timeout = setTimeout(() => {
 		log.error({
 			lifecycle: "shutdown",

apps/api/src/http/errors.test.ts

@@ -83,7 +83,29 @@ describe("handleAppError", () => {
 			requestId: "req_test_validation",
 		});
 		expect(payload.details).toEqual([
-			expect.objectContaining({ field: "body.name" }),
+			{ field: "body.name", message: "Invalid value" },
 		]);
 	});
+
+	it("does not reflect Elysia validation values or messages in production", async () => {
+		process.env.NODE_ENV = "production";
+		const response = handleAppError({
+			code: "VALIDATION",
+			requestId: "req_test_reflection",
+			error: new ValidationError(
+				"body",
+				t.Object({ profile: t.Object({ email: t.String() }) }),
+				{ profile: { email: 867_530_900 } }
+			),
+		});
+		const payload = await readPayload(response);
+		const serialized = JSON.stringify(payload);
+
+		expect(payload.details).toEqual([
+			{ field: "body.profile.email", message: "Invalid value" },
+		]);
+		expect(serialized).not.toContain("867530900");
+		expect(serialized).not.toContain("Expected");
+		expect(serialized).not.toContain("found");
+	});
 });