feat(eventing): add typed signal-to-event projection architecture

apps/api/package.json

@@ -36,12 +36,14 @@
 		"@effect/platform-bun": "catalog:effect",
 		"@maple-dev/clickhouse-builder": "workspace:*",
 		"@maple-dev/effect-sdk": "workspace:*",
+		"@maple/alerting-core": "workspace:*",
 		"@maple/auth": "workspace:*",
 		"@maple/cache": "workspace:*",
 		"@maple/db": "workspace:*",
 		"@maple/domain": "workspace:*",
 		"@maple/effect-cloudflare": "workspace:*",
 		"@maple/email": "workspace:*",
+		"@maple/eventing-core": "workspace:*",
 		"@maple/infra": "workspace:*",
 		"@maple/llm": "workspace:*",
 		"@maple/query-engine": "workspace:*",

apps/api/src/planetscale-webhook-runtime.test.ts

@@ -1,27 +1,44 @@
 import type { MessageBatch } from "@cloudflare/workers-types"
 import { afterEach, assert, describe, it } from "@effect/vitest"
-import { Effect, Layer } from "effect"
+import { OrgId } from "@maple/domain/http"
+import { Effect, Layer, Schema } from "effect"
 import { Database, DatabaseError } from "@/platform/DatabaseLive"
 import { cleanupTestDbs, createTestDb, queryFirstRow, type TestDb } from "@/platform/test-pglite"
 import { processPlanetScaleWebhookBatch } from "./planetscale-webhook-runtime"
+import {
+	projectPlanetScaleWebhookEvent,
+	type PlanetScaleWebhookPayload,
+} from "./services/integrations/planetscale/webhook-events"
 import type { PlanetScaleWebhookJob } from "./services/integrations/planetscale/PlanetScaleWebhookQueue"
 
 const trackedDbs: TestDb[] = []
 
 afterEach(() => cleanupTestDbs(trackedDbs))
 
-const job: PlanetScaleWebhookJob = {
+const orgId = Schema.decodeUnknownSync(OrgId)("org_1")
+
+const basePayload: PlanetScaleWebhookPayload = {
+	timestamp: 1,
+	event: "branch.out_of_memory",
+	organization: "acme",
+	database: "shop",
+	resource: { name: "main" },
+}
+
+const makeJob = (payload: PlanetScaleWebhookPayload = basePayload): PlanetScaleWebhookJob => ({
 	kind: "planetscale-webhook",
-	orgId: "org_1",
+	orgId,
 	connectionId: "connection_1",
-	payload: {
-		event: "branch.out_of_memory",
-		organization: "acme",
-		database: "shop",
-		resource: { name: "main" },
-	},
 	receivedAt: 1_000,
-}
+	event: projectPlanetScaleWebhookEvent({
+		orgId,
+		connectionId: "connection_1",
+		payload,
+		receivedAt: 1_000,
+	}),
+})
+
+const job = makeJob()
 
 const makeBatch = (body: unknown) => {
 	let acknowledged = false
@@ -70,6 +87,120 @@ describe("PlanetScale webhook queue consumer", () => {
 		}).pipe(Effect.provide(testDb.layer))
 	})
 
+	it.effect("applies an issue event exactly once across duplicate queue deliveries", () => {
+		const testDb = createTestDb(trackedDbs)
+		const first = makeBatch(job)
+		const duplicate = makeBatch(job)
+		return Effect.gen(function* () {
+			yield* processPlanetScaleWebhookBatch(first.batch)
+			yield* Effect.promise(() =>
+				testDb.pglite.exec(
+					"UPDATE error_issues SET workflow_state = 'done', resolved_at = '2026-08-20T00:00:00Z'",
+				),
+			)
+			yield* processPlanetScaleWebhookBatch(duplicate.batch)
+			assert.isTrue(first.acknowledged())
+			assert.isTrue(duplicate.acknowledged())
+			const issue = yield* Effect.promise(() =>
+				queryFirstRow<{ occurrence_count: number; workflow_state: string }>(
+					testDb,
+					"SELECT occurrence_count, workflow_state FROM error_issues WHERE org_id = $1",
+					["org_1"],
+				),
+			)
+			assert.strictEqual(issue?.occurrence_count, 1)
+			assert.strictEqual(issue?.workflow_state, "done")
+			const history = yield* Effect.promise(() =>
+				queryFirstRow<{ count: number }>(
+					testDb,
+					"SELECT count(*)::int AS count FROM error_issue_events WHERE org_id = $1",
+					["org_1"],
+				),
+			)
+			assert.strictEqual(history?.count, 1)
+		}).pipe(Effect.provide(testDb.layer))
+	})
+
+	it.effect("recovers exactly once after the timeline commits but the issue transaction fails", () => {
+		const testDb = createTestDb(trackedDbs)
+		const failed = makeBatch(job)
+		const retry = makeBatch(job)
+		return Effect.gen(function* () {
+			yield* Effect.promise(() =>
+				testDb.pglite.exec(`CREATE FUNCTION reject_planetscale_issue_event() RETURNS trigger AS $$
+					BEGIN RAISE EXCEPTION 'forced issue event failure'; END;
+					$$ LANGUAGE plpgsql;
+					CREATE TRIGGER reject_planetscale_issue_event
+					BEFORE INSERT ON error_issue_events
+					FOR EACH ROW EXECUTE FUNCTION reject_planetscale_issue_event();`),
+			)
+			yield* processPlanetScaleWebhookBatch(failed.batch)
+			assert.isTrue(failed.retried())
+			yield* Effect.promise(() =>
+				testDb.pglite.exec(`DROP TRIGGER reject_planetscale_issue_event ON error_issue_events;
+					DROP FUNCTION reject_planetscale_issue_event();`),
+			)
+			yield* processPlanetScaleWebhookBatch(retry.batch)
+			assert.isTrue(retry.acknowledged())
+			const counts = yield* Effect.promise(() =>
+				queryFirstRow<{ timeline: number; issues: number; receipts: number }>(
+					testDb,
+					`SELECT
+						(SELECT count(*)::int FROM planetscale_events) AS timeline,
+						(SELECT count(*)::int FROM error_issues) AS issues,
+						(SELECT count(*)::int FROM planetscale_issue_receipts) AS receipts`,
+				),
+			)
+			assert.deepStrictEqual(counts, { timeline: 1, issues: 1, receipts: 1 })
+		}).pipe(Effect.provide(testDb.layer))
+	})
+
+	it.effect("processes the exact pre-event-envelope queue body during rolling upgrades", () => {
+		const testDb = createTestDb(trackedDbs)
+		const legacyJob = {
+			kind: "planetscale-webhook",
+			orgId,
+			connectionId: "connection_1",
+			payload: basePayload,
+			receivedAt: 1_000,
+		}
+		const delivery = makeBatch(legacyJob)
+		return Effect.gen(function* () {
+			yield* processPlanetScaleWebhookBatch(delivery.batch)
+			assert.isTrue(delivery.acknowledged())
+			assert.isFalse(delivery.retried())
+			const row = yield* Effect.promise(() =>
+				queryFirstRow<{ workflow_state: string; occurrence_count: number }>(
+					testDb,
+					"SELECT workflow_state, occurrence_count FROM error_issues WHERE org_id = $1",
+					["org_1"],
+				),
+			)
+			assert.strictEqual(row?.workflow_state, "triage")
+			assert.strictEqual(row?.occurrence_count, 1)
+		}).pipe(Effect.provide(testDb.layer))
+	})
+
+	it.effect("terminally acknowledges timestamp-less legacy queue bodies", () => {
+		const testDb = createTestDb(trackedDbs)
+		const delivery = makeBatch({
+			kind: "planetscale-webhook",
+			orgId,
+			connectionId: "connection_1",
+			payload: { ...basePayload, timestamp: null },
+			receivedAt: 1_000,
+		})
+		return processPlanetScaleWebhookBatch(delivery.batch).pipe(
+			Effect.tap(() =>
+				Effect.sync(() => {
+					assert.isTrue(delivery.acknowledged())
+					assert.isFalse(delivery.retried())
+				}),
+			),
+			Effect.provide(testDb.layer),
+		)
+	})
+
 	it.effect("acknowledges terminal malformed jobs", () => {
 		const testDb = createTestDb(trackedDbs)
 		const delivery = makeBatch({ kind: "not-a-planetscale-job" })
@@ -84,12 +215,26 @@ describe("PlanetScale webhook queue consumer", () => {
 		)
 	})
 
-	it.effect("writes a lifecycle event to the timeline but not to the issue hub", () => {
+	it.effect("terminally acknowledges schema-valid jobs with contradictory event identity", () => {
 		const testDb = createTestDb(trackedDbs)
 		const delivery = makeBatch({
 			...job,
-			payload: { ...job.payload, event: "branch.ready" },
+			event: { ...job.event, tenantid: Schema.decodeUnknownSync(OrgId)("org_2") },
 		})
+		return processPlanetScaleWebhookBatch(delivery.batch).pipe(
+			Effect.tap(() =>
+				Effect.sync(() => {
+					assert.isTrue(delivery.acknowledged())
+					assert.isFalse(delivery.retried())
+				}),
+			),
+			Effect.provide(testDb.layer),
+		)
+	})
+
+	it.effect("writes a lifecycle event to the timeline but not to the issue hub", () => {
+		const testDb = createTestDb(trackedDbs)
+		const delivery = makeBatch(makeJob({ ...basePayload, event: "branch.ready" }))
 		return Effect.gen(function* () {
 			yield* processPlanetScaleWebhookBatch(delivery.batch)
 			assert.isTrue(delivery.acknowledged())
@@ -138,14 +283,13 @@ describe("PlanetScale webhook queue consumer", () => {
 
 	it.effect("carries the deploy-request number so redelivery dedupes", () => {
 		const testDb = createTestDb(trackedDbs)
-		const delivery = makeBatch({
-			...job,
-			payload: {
-				...job.payload,
+		const delivery = makeBatch(
+			makeJob({
+				...basePayload,
 				event: "deploy_request.schema_applied",
 				resource: { number: 42 },
-			},
-		})
+			}),
+		)
 		return Effect.gen(function* () {
 			yield* processPlanetScaleWebhookBatch(delivery.batch)
 			const event = yield* Effect.promise(() =>