feat(security): audit_trail_events table, list endpoint and Maple's own events

apps/ai/src/worker/http.test.ts

@@ -22,6 +22,7 @@ import { HttpServerResponse } from "effect/http"
 import { TestClock } from "effect/testing"
 import { AuditEventsQueueProducer, MapleDbConnection } from "@maple/backend/platform/bindings"
 import { AuditLogService } from "@maple/backend/services/audit/AuditLogService"
+import { AuditTrailService } from "@maple/backend/services/audit-trail/AuditTrailService"
 import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService"
 import { makeWarehouseServiceStub } from "@maple/backend/testing/warehouse-test-support"
 import { OrgId } from "@maple/primitives"
@@ -216,9 +217,13 @@ describe("the maple-ai Worker through alchemy's bridge", () => {
 			const handedOver = yield* Ref.make<ReadonlyArray<Effect.Effect<void>>>([])
 			const sent = yield* Ref.make(0)
 			const ingested = yield* Ref.make(0)
-			const warehouse = Layer.succeed(
-				WarehouseQueryService,
-				makeWarehouseServiceStub({ ingest: () => Ref.update(ingested, (n) => n + 1) }),
+			const warehouse = Layer.effect(AuditTrailService, AuditTrailService.make).pipe(
+				Layer.provideMerge(
+					Layer.succeed(
+						WarehouseQueryService,
+						makeWarehouseServiceStub({ ingest: () => Ref.update(ingested, (n) => n + 1) }),
+					),
+				),
 			)
 			const producer = Layer.succeed(AuditEventsQueueProducer, {
 				sendBatch: (messages) => Ref.update(sent, (n) => n + messages.length),

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

@@ -1,26 +1,32 @@
 import { describe, expect, it } from "@effect/vitest"
 import { OrgId } from "@maple/primitives"
-import type { AuditLogRow } from "@maple/domain/tinybird"
 import { WarehouseUpstreamError } from "@maple/domain/http"
 import { Effect, Layer, Schema, Array as Arr } from "effect"
 import { processAuditEventsBatch } from "./audit-events-runtime"
 import { makeWarehouseServiceStub } from "@/routes/v2/v2-test-support"
 import { AuditLogEvent, encodeAuditLogEventSync } from "@maple/backend/services/audit/audit-event"
 import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService"
+import { AuditTrailService } from "@maple/backend/services/audit-trail/AuditTrailService"
 
 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) =>
+const event = (
+	id: string,
+	orgId: OrgId = ORG,
+	what: { readonly action: string; readonly outcome: "allowed" | "denied" } = {
+		action: "dashboard.created",
+		outcome: "allowed",
+	},
+) =>
 	encodeAuditLogEventSync(
 		new AuditLogEvent({
 			orgId,
 			id: Schema.decodeSync(AuditLogEvent.fields.id)(id),
 			actorType: "user",
 			source: "dashboard",
-			action: "dashboard.created",
-			outcome: "allowed",
+			...what,
 			occurredAtMs: 1_700_000_000_000,
 		}),
 	)
@@ -45,42 +51,68 @@ const message = (body: unknown, attempts: number) => {
 const batchOf = (...messages: ReadonlyArray<{ readonly message: unknown }>) =>
 	({ messages: Arr.map(messages, (entry) => entry.message) }) as never
 
+/** The consumer's services over one warehouse stub: the trail writes through the same `ingest`. */
+const withTrail = (stub: ReturnType<typeof makeWarehouseServiceStub>) =>
+	Layer.effect(AuditTrailService, AuditTrailService.make).pipe(
+		Layer.provideMerge(Layer.succeed(WarehouseQueryService, stub)),
+	)
+
 /** A warehouse whose `ingest` is interrupted — a deploy tearing the isolate down. */
-const interruptedWarehouse = Layer.succeed(
-	WarehouseQueryService,
-	makeWarehouseServiceStub({ ingest: () => Effect.interrupt }),
-)
+const interruptedWarehouse = withTrail(makeWarehouseServiceStub({ ingest: () => Effect.interrupt }))
 
 /** A warehouse whose `ingest` dies rather than failing — an unexpected defect. */
-const dyingWarehouse = Layer.succeed(
-	WarehouseQueryService,
+const dyingWarehouse = withTrail(
 	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,
+const AUDIT_LOG = "audit_log"
+const AUDIT_TRAIL = "audit_trail_events"
+
+/**
+ * A warehouse whose `ingest` records each call, failing the ones to the `failing` datasources.
+ * `trailCopy` makes the write to the audit trail die or be interrupted instead.
+ */
+const warehouse = (failing: ReadonlyArray<string> = [], trailCopy?: "dies" | "interrupted") => {
+	const written: Array<{
+		orgId: string
+		datasource: string
+		rows: ReadonlyArray<unknown>
+		bestEffort: boolean
+	}> = []
+	const layer = withTrail(
 		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> })
-						}),
+			ingest: (tenant, datasource, rows, options) =>
+				datasource === AUDIT_TRAIL && trailCopy === "dies"
+					? Effect.die(new Error("trail exploded"))
+					: datasource === AUDIT_TRAIL && trailCopy === "interrupted"
+						? Effect.interrupt
+						: failing.includes(datasource)
+							? Effect.fail(
+									new WarehouseUpstreamError({
+										message: "tinybird down",
+										pipeName: datasource,
+										cause: new Error("down"),
+									}),
+								)
+							: Effect.sync(() => {
+									written.push({
+										orgId: tenant.orgId,
+										datasource,
+										rows,
+										bestEffort: options?.bestEffort === true,
+									})
+								}),
 		}),
 	)
-	return { written, layer }
+	/** `[orgId, row count]` per write to one datasource, in a stable order. */
+	const writesTo = (datasource: string) =>
+		Arr.map(
+			Arr.filter(written, (write) => write.datasource === datasource),
+			(write) => [write.orgId, write.rows.length] as const,
+		).sort()
+	return { written, writesTo, layer }
 }
 
 describe("processAuditEventsBatch", () => {
@@ -97,16 +129,107 @@ describe("processAuditEventsBatch", () => {
 			expect(first.calls).toEqual(["ack"])
 			expect(second.calls).toEqual(["ack"])
 			expect(other.calls).toEqual(["ack"])
-			expect(Arr.map(store.written, (write) => [write.orgId, write.rows.length]).sort()).toEqual([
+			expect(store.writesTo(AUDIT_LOG)).toEqual([
 				[OTHER_ORG, 1],
 				[ORG, 2],
 			])
-			expect(
-				Arr.every(
-					Arr.flatMap(store.written, (write) => write.rows),
-					(row) => row.Action === "dashboard.created",
-				),
-			).toBe(true)
+		}),
+	)
+
+	it.effect("copies changes and refusals to the audit trail, and leaves allowed reads out", () =>
+		Effect.gen(function* () {
+			const store = warehouse()
+			const change = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			const read = message(
+				event("22222222-2222-4222-8222-222222222222", ORG, {
+					action: "telemetry.read",
+					outcome: "allowed",
+				}),
+				1,
+			)
+			const refusedRead = message(
+				event("33333333-3333-4333-8333-333333333333", ORG, {
+					action: "telemetry.read",
+					outcome: "denied",
+				}),
+				1,
+			)
+			yield* processAuditEventsBatch(batchOf(change, read, refusedRead), MAX_RETRIES).pipe(
+				Effect.provide(store.layer),
+			)
+
+			expect(store.writesTo(AUDIT_LOG)).toEqual([[ORG, 3]])
+			expect(store.writesTo(AUDIT_TRAIL)).toEqual([[ORG, 2]])
+		}),
+	)
+
+	it.effect("writes nothing to the audit trail for a batch of allowed reads", () =>
+		Effect.gen(function* () {
+			const store = warehouse()
+			const read = message(
+				event("11111111-1111-4111-8111-111111111111", ORG, {
+					action: "telemetry.read",
+					outcome: "allowed",
+				}),
+				1,
+			)
+			yield* processAuditEventsBatch(batchOf(read), MAX_RETRIES).pipe(Effect.provide(store.layer))
+
+			expect(read.calls).toEqual(["ack"])
+			expect(store.writesTo(AUDIT_TRAIL)).toEqual([])
+		}),
+	)
+
+	// `audit_log` is the record. A message retried because only its copy failed would be
+	// rewritten there on every attempt and end in the DLQ for an entry that was stored.
+	it.effect("acks an entry whose audit trail copy died", () =>
+		Effect.gen(function* () {
+			const store = warehouse([], "dies")
+			const change = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			yield* processAuditEventsBatch(batchOf(change), MAX_RETRIES).pipe(Effect.provide(store.layer))
+
+			expect(change.calls).toEqual(["ack"])
+			expect(store.writesTo(AUDIT_LOG)).toEqual([[ORG, 1]])
+		}),
+	)
+
+	// A deploy landing between the two writes: unacked, so the platform redelivers the entry.
+	it.effect("leaves an entry unacked when its audit trail copy is interrupted", () =>
+		Effect.gen(function* () {
+			const store = warehouse([], "interrupted")
+			const change = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			const exit = yield* processAuditEventsBatch(batchOf(change), MAX_RETRIES).pipe(
+				Effect.provide(store.layer),
+				Effect.exit,
+			)
+
+			expect(exit._tag).toBe("Failure")
+			expect(change.calls).toEqual([])
+		}),
+	)
+
+	it.effect("acks an entry whose audit trail copy failed", () =>
+		Effect.gen(function* () {
+			const store = warehouse([AUDIT_TRAIL])
+			const change = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			yield* processAuditEventsBatch(batchOf(change), MAX_RETRIES).pipe(Effect.provide(store.layer))
+
+			expect(change.calls).toEqual(["ack"])
+			expect(store.writesTo(AUDIT_LOG)).toEqual([[ORG, 1]])
+		}),
+	)
+
+	// The copy is dropped on failure, so its failed insert must not read as a lost write.
+	it.effect("writes the audit trail copy as best effort and the entry itself as not", () =>
+		Effect.gen(function* () {
+			const store = warehouse()
+			const change = message(event("11111111-1111-4111-8111-111111111111"), 1)
+			yield* processAuditEventsBatch(batchOf(change), MAX_RETRIES).pipe(Effect.provide(store.layer))
+
+			expect(Arr.map(store.written, (write) => [write.datasource, write.bestEffort])).toEqual([
+				[AUDIT_LOG, false],
+				[AUDIT_TRAIL, true],
+			])
 		}),
 	)
 
@@ -117,7 +240,7 @@ describe("processAuditEventsBatch", () => {
 		Effect.gen(function* () {
 			const exhausted = message(event("44444444-4444-4444-8444-444444444444"), MAX_RETRIES + 1)
 			yield* processAuditEventsBatch(batchOf(exhausted), MAX_RETRIES).pipe(
-				Effect.provide(warehouse(true).layer),
+				Effect.provide(warehouse([AUDIT_LOG]).layer),
 			)
 			expect(exhausted.calls).toEqual(["retry"])
 		}),
@@ -128,7 +251,7 @@ describe("processAuditEventsBatch", () => {
 			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), MAX_RETRIES).pipe(
-				Effect.provide(warehouse(true).layer),
+				Effect.provide(warehouse([AUDIT_LOG]).layer),
 			)
 			expect(a.calls).toEqual(["retry"])
 			expect(b.calls).toEqual(["retry"])
@@ -175,7 +298,7 @@ describe("processAuditEventsBatch", () => {
 
 			expect(malformed.calls).toEqual(["ack"])
 			expect(fine.calls).toEqual(["ack"])
-			expect(store.written).toHaveLength(1)
+			expect(store.writesTo(AUDIT_LOG)).toEqual([[ORG, 1]])
 		}),
 	)
 })

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

@@ -8,6 +8,11 @@ import type { QueueBatch } from "@maple/backend/platform/queue-batch"
 import { ALERTING_SYSTEM_USER_ID, systemTenant } from "@maple/backend/services/auth/system-tenant"
 import { AUDIT_LOG_DATASOURCE } from "@maple/backend/services/audit/AuditLogService"
 import { WarehouseQueryService } from "@maple/backend/services/warehouse/WarehouseQueryService"
+import {
+	AuditTrailService,
+	type AuditTrailServiceApi,
+} from "@maple/backend/services/audit-trail/AuditTrailService"
+import { mapleAuditTrailEvents } from "@maple/backend/services/audit-trail/providers/maple"
 import {
 	type AuditLogEvent,
 	auditEventToRow,
@@ -21,7 +26,7 @@ import {
  * layer requires them, not because a write ever consults them. Its spans are
  * `maple-api`'s, through the telemetry the bridge builds into the event.
  */
-export const AuditEventsLive = WarehouseQueryService.layer.pipe(
+export const AuditEventsLive = Layer.mergeAll(WarehouseQueryService.layer, AuditTrailService.layer).pipe(
 	Layer.provide(Layer.mergeAll(EventBaseLive, EdgeCacheServiceLive)),
 )
 
@@ -72,8 +77,24 @@ const retryOrExhaust = (message: Message, cause: unknown, maxRetries: number) =>
 }
 
 /**
- * Audit events queue consumer: lowers each event to its `audit_log` row and
- * writes one batch per org through the managed ingest pipeline. The table is a
+ * Changes and refusals also go to the unified audit trail. `audit_log` is the record, so a failed
+ * copy is logged and dropped: retrying the message for it would send stored entries to the DLQ.
+ */
+const copyToTrail = (trail: AuditTrailServiceApi, orgId: OrgId, events: ReadonlyArray<AuditLogEvent>) =>
+	Effect.suspend(() => trail.write(orgId, mapleAuditTrailEvents(events), { bestEffort: true })).pipe(
+		Effect.catchCause((cause) =>
+			Cause.hasInterruptsOnly(cause)
+				? Effect.interrupt
+				: Effect.logWarning("Audit trail copy failed; audit_log kept the entries").pipe(
+						Effect.annotateLogs({ orgId, entries: events.length, error: summarizeCause(cause) }),
+					),
+		),
+	)
+
+/**
+ * Audit events queue consumer: lowers each event to its `audit_log` row (and, for a change or
+ * a refusal, its `audit_trail_events` row) and writes one batch per org through the managed
+ * ingest pipeline. The table is a
  * ReplacingMergeTree on the entry id, so queue redelivery collapses at merge
  * time; write failures retry through the queue's policy and, once exhausted,
  * land in `audit-events-dlq` rather than disappearing.
@@ -83,6 +104,7 @@ export const processAuditEventsBatch = Effect.fn("AuditEvents.processBatch")(fun
 	maxRetries: number,
 ) {
 	const warehouse = yield* WarehouseQueryService
+	const trail = yield* AuditTrailService
 	const now = yield* Clock.currentTimeMillis
 
 	const decodedPerMessage = yield* Effect.forEach(batch.messages, (message) =>
@@ -113,14 +135,16 @@ export const processAuditEventsBatch = Effect.fn("AuditEvents.processBatch")(fun
 
 	yield* Effect.forEach(
 		byOrg,
-		([orgId, group]) =>
-			warehouse
+		([orgId, group]) => {
+			const events = Arr.map(group, ({ event }) => event)
+			return warehouse
 				.ingest(
 					systemTenant(orgId, ALERTING_SYSTEM_USER_ID),
 					AUDIT_LOG_DATASOURCE,
-					Arr.map(group, ({ event }) => auditEventToRow(event, now)),
+					Arr.map(events, (event) => auditEventToRow(event, now)),
 				)
 				.pipe(
+					Effect.andThen(copyToTrail(trail, orgId, events)),
 					Effect.flatMap(() =>
 						Effect.sync(() => {
 							for (const { message } of group) message.ack()
@@ -146,7 +170,8 @@ export const processAuditEventsBatch = Effect.fn("AuditEvents.processBatch")(fun
 									},
 								),
 					),
-				),
+				)
+		},
 		{ concurrency: 3, discard: true },
 	)
 })