feat(audit): org-wide audit log with actor attribution and durable queue delivery

apps/alerting/src/scheduled.ts

@@ -12,6 +12,7 @@ import {
 	AlertRuntime,
 	AlertRulesService,
 	AlertsService,
+	AuditLogService,
 	AnomalyDetectionService,
 	BucketCacheService,
 	CacheBackendLive,
@@ -147,7 +148,13 @@ export const buildLayer = (env: AlertingWorkerEnv) => {
 
 	const ErrorActorsServiceLive = ErrorActorsService.layer.pipe(Layer.provide(BaseLive))
 	const ErrorIssueWorkflowServiceLive = ErrorIssueWorkflowService.layer.pipe(
-		Layer.provide(Layer.mergeAll(BaseLive, ErrorActorsServiceLive)),
+		Layer.provide(
+			Layer.mergeAll(
+				BaseLive,
+				ErrorActorsServiceLive,
+				AuditLogService.layer.pipe(Layer.provide(WarehouseQueryServiceLive)),
+			),
+		),
 	)
 	const ErrorPolicyServiceLive = ErrorPolicyService.layer.pipe(Layer.provide(BaseLive))
 

apps/api/alchemy.run.ts

@@ -249,6 +249,8 @@ const makeWorkerBindings = ({
 	vcsSyncQueueName,
 	planetScaleWebhookQueue,
 	planetScaleWebhookQueueName,
+	auditEventsQueue,
+	auditEventsQueueName,
 }: {
 	stage: MapleStage
 	mapleDb: Cloudflare.Hyperdrive.Connection | undefined
@@ -258,6 +260,8 @@ const makeWorkerBindings = ({
 	vcsSyncQueueName: string
 	planetScaleWebhookQueue: Cloudflare.Queues.Queue
 	planetScaleWebhookQueueName: string
+	auditEventsQueue: Cloudflare.Queues.Queue
+	auditEventsQueueName: string
 }) => ({
 	// Ref stages attach MAPLE_DB via `bindMapleDbRef` below.
 	...(mapleDb ? { MAPLE_DB: mapleDb } : undefined),
@@ -279,6 +283,8 @@ const makeWorkerBindings = ({
 	VCS_SYNC_QUEUE_NAME: vcsSyncQueueName,
 	PLANETSCALE_WEBHOOK_QUEUE: planetScaleWebhookQueue,
 	PLANETSCALE_WEBHOOK_QUEUE_NAME: planetScaleWebhookQueueName,
+	AUDIT_EVENTS_QUEUE: auditEventsQueue,
+	AUDIT_EVENTS_QUEUE_NAME: auditEventsQueueName,
 	// Long-running schema-apply: chunks heavy backfill migrations across durable
 	// steps so they never hit the Worker request budget. Class is exported from
 	// src/worker.ts. The first Workflow arg IS the physical workflow name; the
@@ -389,6 +395,15 @@ export const createMapleApi = ({
 		const planetScaleWebhookQueue = yield* Cloudflare.Queues.Queue("planetscale-webhooks", {
 			name: planetScaleWebhookQueueName,
 		})
+		const auditEventsQueueName = resolveWorkerName("audit-events", stage)
+		const auditEventsQueue = yield* Cloudflare.Queues.Queue("audit-events", {
+			name: auditEventsQueueName,
+		})
+		// Parking lot for audit entries that exhausted their retries. Deliberately
+		// has no consumer: an entry landing here is a lost audit record, and the
+		// point is that it survives for inspection instead of being dropped.
+		const auditEventsDlqName = resolveWorkerName("audit-events-dlq", stage)
+		yield* Cloudflare.Queues.Queue("audit-events-dlq", { name: auditEventsDlqName })
 
 		const worker = (yield* Cloudflare.Worker("api", {
 			name: resolveWorkerName("api", stage),
@@ -434,6 +449,8 @@ export const createMapleApi = ({
 					vcsSyncQueueName,
 					planetScaleWebhookQueue,
 					planetScaleWebhookQueueName,
+					auditEventsQueue,
+					auditEventsQueueName,
 				}),
 				...configuredEnv,
 				...devEnv,
@@ -463,6 +480,21 @@ export const createMapleApi = ({
 				maxWaitTimeMs: 5000,
 			},
 		})
+		// Audit entries tolerate a few seconds of delivery latency; batch wider and
+		// wait longer so one insert round-trip covers many entries.
+		yield* Cloudflare.Queues.Consumer("audit-events-consumer", {
+			queueId: auditEventsQueue.queueId,
+			scriptName: worker.workerName,
+			// `maxRetries` must stay in sync with AUDIT_EVENTS_MAX_RETRIES in
+			// audit-events-runtime.ts, which logs the drop on the final attempt.
+			deadLetterQueue: auditEventsDlqName,
+			settings: {
+				batchSize: 25,
+				maxConcurrency: 2,
+				maxRetries: 5,
+				maxWaitTimeMs: 5000,
+			},
+		})
 
 		return worker
 	})

apps/api/src/alerting.ts

@@ -4,6 +4,7 @@ export { AlertDestinationsService } from "./services/alerts/AlertDestinationsSer
 export { AlertReadModelsService } from "./services/alerts/AlertReadModelsService"
 export { AlertRulesService } from "./services/alerts/AlertRulesService"
 export { AnomalyDetectionService } from "./services/alerts/AnomalyDetectionService"
+export { AuditLogService } from "./services/audit/AuditLogService"
 export { BucketCacheService } from "@maple/query-engine/caching"
 export { CacheBackendLive } from "@/platform/CacheBackendLive"
 export { CloudflareAnalyticsService } from "./services/integrations/CloudflareAnalyticsService"

apps/api/src/audit-events-runtime.test.ts

@@ -0,0 +1,163 @@
+import { describe, expect, it } from "@effect/vitest"
+import { OrgId } from "@maple/domain/primitives"
+import type { AuditLogRow } from "@maple/domain/tinybird"
+import { WarehouseUpstreamError } from "@maple/domain/http"
+import { Effect, Layer, Schema } from "effect"
+import { processAuditEventsBatch } from "./audit-events-runtime"
+import { makeWarehouseServiceStub } from "@/routes/v2/v2-test-support"
+import { AuditLogEvent, encodeAuditLogEventSync } from "./services/audit/audit-event"
+import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService"
+
+const asOrgId = Schema.decodeUnknownSync(OrgId)
+const ORG = asOrgId("org_audit_consumer_test")
+const OTHER_ORG = asOrgId("org_audit_consumer_other")
+
+const event = (id: string, orgId: OrgId = ORG) =>
+	encodeAuditLogEventSync(
+		new AuditLogEvent({
+			orgId,
+			id: Schema.decodeUnknownSync(AuditLogEvent.fields.id)(id),
+			actorType: "user",
+			source: "dashboard",
+			action: "dashboard.created",
+			outcome: "allowed",
+			occurredAtMs: 1_700_000_000_000,
+		}),
+	)
+
+/** One queue message, recording which terminal call the consumer made on it. */
+const message = (body: unknown, attempts: number) => {
+	const calls: string[] = []
+	return {
+		message: {
+			body,
+			attempts,
+			ack: () => calls.push("ack"),
+			retry: () => calls.push("retry"),
+		},
+		calls,
+	}
+}
+
+const batchOf = (...messages: ReadonlyArray<{ readonly message: unknown }>) =>
+	({ messages: messages.map((entry) => entry.message) }) as never
+
+/** A warehouse whose `ingest` is interrupted — a deploy tearing the isolate down. */
+const interruptedWarehouse = Layer.succeed(
+	WarehouseQueryService,
+	makeWarehouseServiceStub({ ingest: () => Effect.interrupt }),
+)
+
+/** A warehouse whose `ingest` dies rather than failing — an unexpected defect. */
+const dyingWarehouse = Layer.succeed(
+	WarehouseQueryService,
+	makeWarehouseServiceStub({
+		ingest: () => Effect.die(new Error("ingest exploded")),
+	}),
+)
+
+/** A warehouse whose `ingest` records each call, or fails every call. */
+const warehouse = (fail = false) => {
+	const written: Array<{ orgId: string; rows: ReadonlyArray<AuditLogRow> }> = []
+	const layer = Layer.succeed(
+		WarehouseQueryService,
+		makeWarehouseServiceStub({
+			ingest: (tenant, _datasource, rows) =>
+				fail
+					? Effect.fail(
+							new WarehouseUpstreamError({ message: "tinybird down", pipeName: "audit_log", cause: new Error("down") }),
+						)
+					: Effect.sync(() => {
+							// SAFETY: this stub only ever receives the audit datasource's rows.
+							written.push({ orgId: tenant.orgId, rows: rows as ReadonlyArray<AuditLogRow> })
+						}),
+		}),
+	)
+	return { written, layer }
+}
+
+describe("processAuditEventsBatch", () => {
+	it.effect("writes well-formed events through ingest, one batch per org, and acks them", () =>
+		Effect.gen(function* () {
+			const store = warehouse()
+			const first = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			const second = message(event("22222222-2222-4222-8222-222222222222"), 1)
+			const other = message(event("33333333-3333-4333-8333-333333333333", OTHER_ORG), 1)
+			yield* processAuditEventsBatch(batchOf(first, second, other)).pipe(Effect.provide(store.layer))
+
+			expect(first.calls).toEqual(["ack"])
+			expect(second.calls).toEqual(["ack"])
+			expect(other.calls).toEqual(["ack"])
+			expect(store.written.map((write) => [write.orgId, write.rows.length]).sort()).toEqual([
+				[OTHER_ORG, 1],
+				[ORG, 2],
+			])
+			expect(store.written.flatMap((write) => write.rows).every((row) => row.Action === "dashboard.created")).toBe(
+				true,
+			)
+		}),
+	)
+
+	// Cloudflare routes a message to the DLQ only when the consumer retries it
+	// past `max_retries`. Acking on the final attempt would discard the entry
+	// instead, which is exactly the silent drop this branch exists to prevent.
+	it.effect("retries a failed write on the final attempt so the message reaches the DLQ", () =>
+		Effect.gen(function* () {
+			const exhausted = message(event("44444444-4444-4444-8444-444444444444"), 6)
+			yield* processAuditEventsBatch(batchOf(exhausted)).pipe(Effect.provide(warehouse(true).layer))
+			expect(exhausted.calls).toEqual(["retry"])
+		}),
+	)
+
+	it.effect("retries every message of a failed org batch while attempts remain", () =>
+		Effect.gen(function* () {
+			const a = message(event("55555555-5555-4555-8555-555555555555"), 2)
+			const b = message(event("66666666-6666-4666-8666-666666666666"), 2)
+			yield* processAuditEventsBatch(batchOf(a, b)).pipe(Effect.provide(warehouse(true).layer))
+			expect(a.calls).toEqual(["retry"])
+			expect(b.calls).toEqual(["retry"])
+		}),
+	)
+
+	// A typed failure retries; so must a defect. Catching only the failure
+	// channel would let an unexpected throw escape the consumer, and Cloudflare
+	// treats a consumer that neither acked nor retried as a retry anyway — but
+	// silently, with no log and no DLQ accounting.
+	it.effect("retries when the write dies instead of failing", () =>
+		Effect.gen(function* () {
+			const defect = message(event("77777777-7777-4777-8777-777777777777"), 2)
+			yield* processAuditEventsBatch(batchOf(defect)).pipe(Effect.provide(dyingWarehouse))
+			expect(defect.calls).toEqual(["retry"])
+		}),
+	)
+
+	// Interruption is not a failed attempt. Counting it as one would spend the
+	// message's retry budget — and eventually route it to the DLQ — for a deploy.
+	// Unacked is enough: the platform redelivers.
+	it.effect("does not count an interrupted batch as an attempt", () =>
+		Effect.gen(function* () {
+			const torn = message(event("88888888-8888-4888-8888-888888888888"), 2)
+			const exit = yield* processAuditEventsBatch(batchOf(torn)).pipe(
+				Effect.provide(interruptedWarehouse),
+				Effect.exit,
+			)
+			expect(exit._tag).toBe("Failure")
+			expect(torn.calls).toEqual([])
+		}),
+	)
+
+	// A message that cannot decode will never decode. Retrying only burns the
+	// attempts that would otherwise carry a recoverable message to the DLQ.
+	it.effect("acks a malformed message instead of retrying it forever", () =>
+		Effect.gen(function* () {
+			const store = warehouse()
+			const malformed = message({ not: "an audit event" }, 1)
+			const fine = message(event("77777777-7777-4777-8777-777777777777"), 1)
+			yield* processAuditEventsBatch(batchOf(malformed, fine)).pipe(Effect.provide(store.layer))
+
+			expect(malformed.calls).toEqual(["ack"])
+			expect(fine.calls).toEqual(["ack"])
+			expect(store.written).toHaveLength(1)
+		}),
+	)
+})