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

apps/api/package.json

@@ -34,12 +34,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,41 @@
 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 } from "./services/integrations/planetscale/webhook-events"
 import type { PlanetScaleWebhookJob } from "./services/integrations/planetscale/PlanetScaleWebhookQueue"
 
 const trackedDbs: TestDb[] = []
 
 afterEach(() => cleanupTestDbs(trackedDbs))
 
-const job: PlanetScaleWebhookJob = {
-	kind: "planetscale-webhook",
-	orgId: "org_1",
-	connectionId: "connection_1",
-	payload: {
+const orgId = Schema.decodeUnknownSync(OrgId)("org_1")
+
+const makeJob = (
+	payload: PlanetScaleWebhookJob["payload"] = {
 		event: "branch.out_of_memory",
 		organization: "acme",
 		database: "shop",
 		resource: { name: "main" },
 	},
+): PlanetScaleWebhookJob => ({
+	kind: "planetscale-webhook",
+	orgId,
+	connectionId: "connection_1",
+	payload,
 	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 +84,32 @@ describe("PlanetScale webhook queue consumer", () => {
 		}).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: job.payload,
+			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("acknowledges terminal malformed jobs", () => {
 		const testDb = createTestDb(trackedDbs)
 		const delivery = makeBatch({ kind: "not-a-planetscale-job" })
@@ -86,10 +126,7 @@ describe("PlanetScale webhook queue consumer", () => {
 
 	it.effect("writes a lifecycle event to the timeline but not to the issue hub", () => {
 		const testDb = createTestDb(trackedDbs)
-		const delivery = makeBatch({
-			...job,
-			payload: { ...job.payload, event: "branch.ready" },
-		})
+		const delivery = makeBatch(makeJob({ ...job.payload, event: "branch.ready" }))
 		return Effect.gen(function* () {
 			yield* processPlanetScaleWebhookBatch(delivery.batch)
 			assert.isTrue(delivery.acknowledged())
@@ -138,14 +175,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: {
+		const delivery = makeBatch(
+			makeJob({
 				...job.payload,
 				event: "deploy_request.schema_applied",
 				resource: { number: 42 },
-			},
-		})
+			}),
+		)
 		return Effect.gen(function* () {
 			yield* processPlanetScaleWebhookBatch(delivery.batch)
 			const event = yield* Effect.promise(() =>

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

@@ -10,9 +10,10 @@ import {
 	deployRequestNumber,
 	insertPlanetScaleEvent,
 	planetScaleBranchName,
+	projectPlanetScaleWebhookEvent,
 	upsertPlanetScaleIssue,
 } from "./services/integrations/planetscale/webhook-events"
-import { PlanetScaleWebhookJob } from "./services/integrations/planetscale/PlanetScaleWebhookQueue"
+import { PlanetScaleWebhookQueueMessage } from "./services/integrations/planetscale/PlanetScaleWebhookQueue"
 
 const telemetry = MapleCloudflareSDK.make({
 	serviceName: "maple-api",
@@ -32,7 +33,7 @@ export const buildPlanetScaleWebhookLayer = (_env: Record<string, unknown>) => {
 
 export const flushPlanetScaleWebhookTelemetry = (env: Record<string, unknown>) => telemetry.flush(env)
 
-const decodeJob = Schema.decodeUnknownEffect(PlanetScaleWebhookJob)
+const decodeJob = Schema.decodeUnknownEffect(PlanetScaleWebhookQueueMessage)
 
 export const processPlanetScaleWebhookBatch = (batch: MessageBatch<unknown>) =>
 	Effect.forEach(
@@ -54,9 +55,29 @@ export const processPlanetScaleWebhookBatch = (batch: MessageBatch<unknown>) =>
 							),
 						),
 					onSuccess: (job) => {
-						const classified = classifyPlanetScaleEvent(job.payload.event)
+						// Old jobs can remain in Cloudflare Queue across a deploy. Rebuild the
+						// event from the durable legacy fields instead of malformed-acking them.
+						const event =
+							job.event ??
+							projectPlanetScaleWebhookEvent({
+								orgId: job.orgId,
+								connectionId: job.connectionId,
+								payload: job.payload,
+								receivedAt: job.receivedAt,
+							})
+						const eventData =
+							typeof event.data === "object" &&
+							event.data !== null &&
+							!Array.isArray(event.data)
+								? (event.data as { readonly [key: string]: unknown })
+								: null
+						const eventName =
+							typeof eventData?.event === "string" ? eventData.event : job.payload.event
+						const classified = classifyPlanetScaleEvent(eventName)
 						const annotateJob = Effect.annotateCurrentSpan({
 							orgId: job.orgId,
+							"maple.event.id": event.id,
+							"maple.event.type": event.type,
 							"maple.planetscale.connection_id": job.connectionId,
 							"maple.planetscale.webhook.event": job.payload.event,
 						})

apps/api/src/routes/v1/planetscale-webhook.http.test.ts

@@ -187,11 +187,13 @@ describe("PlanetScaleWebhookRouter", () => {
 				assert.strictEqual(jobs[0]?.orgId, "org_1")
 				assert.strictEqual(jobs[0]?.connectionId, CONNECTION_ID)
 				assert.strictEqual(jobs[0]?.payload.event, "branch.out_of_memory")
+				assert.strictEqual(jobs[0]?.event.type, "dev.maple.planetscale.webhook.received.v1")
+				assert.strictEqual(jobs[0]?.event.tenantid, "org_1")
 			}).pipe(Effect.ensuring(Effect.promise(dispose)))
 		}).pipe(Effect.provide(testDb.layer))
 	})
 
-	it.effect("enqueues lifecycle events too, and still drops genuinely unknown ones", () => {
+	it.effect("enqueues every verified factual event before downstream classification", () => {
 		const testDb = createTestDb(trackedDbs)
 		const jobs: PlanetScaleWebhookJob[] = []
 		return Effect.gen(function* () {
@@ -260,15 +262,16 @@ describe("PlanetScaleWebhookRouter", () => {
 				assert.strictEqual(branchReady.status, 202)
 				assert.strictEqual(jobs.length, 2)
 
-				// Forward-compatibility must not become "enqueue everything": an
-				// event neither side knows is acknowledged and dropped.
+				// Unknown provider facts also enter the typed event layer. The current
+				// issue/timeline consumer may ignore them, but other consumers can opt in.
 				const unknown = yield* post({
 					event: "branch.some_future_event",
 					organization: "acme",
 					database: "shop",
 				})
 				assert.strictEqual(unknown.status, 202)
-				assert.strictEqual(jobs.length, 2)
+				assert.strictEqual(jobs.length, 3)
+				assert.strictEqual(jobs[2]?.payload.event, "branch.some_future_event")
 			}).pipe(Effect.ensuring(Effect.promise(dispose)))
 		}).pipe(Effect.provide(testDb.layer))
 	})

apps/api/src/routes/v1/planetscale-webhook.http.ts

@@ -9,6 +9,7 @@ import { Env } from "@/platform/Env"
 import {
 	classifyPlanetScaleEvent,
 	decodePlanetScaleWebhookPayload,
+	projectPlanetScaleWebhookEvent,
 	verifyPlanetScaleSignature,
 } from "@/services/integrations/planetscale/webhook-events"
 import { PlanetScaleWebhookQueue } from "@/services/integrations/planetscale/PlanetScaleWebhookQueue"
@@ -163,17 +164,33 @@ export const PlanetScaleWebhookRouter = HttpRouter.use((router) =>
 				return textResponse("ok", 200)
 			}
 
-			// Both issue-worthy and timeline-only events go through the queue: the
-			// durable retry is what makes a missed deploy marker recoverable.
-			if (classified.action === "issue" || classified.action === "timeline") {
+			// Every verified factual event is normalized and projected before the
+			// durable queue boundary. The queued CloudEvent is the stable contract;
+			// payload remains temporarily for parity with the existing consumers.
+			{
 				const now = yield* Clock.currentTimeMillis
+				const orgId = decodeOrgIdSync(connection.orgId)
+				const event = yield* Effect.try({
+					try: () =>
+						projectPlanetScaleWebhookEvent({
+							orgId,
+							connectionId,
+							payload,
+							receivedAt: now,
+						}),
+					catch: () =>
+						new PlanetScaleWebhookUnavailable({
+							message: "Webhook event projection unavailable",
+						}),
+				})
 				const enqueued = yield* webhookQueue
 					.send({
 						kind: "planetscale-webhook",
-						orgId: decodeOrgIdSync(connection.orgId),
+						orgId,
 						connectionId,
 						payload,
 						receivedAt: now,
+						event,
 					})
 					.pipe(
 						Effect.tapError((error) =>
@@ -198,10 +215,6 @@ export const PlanetScaleWebhookRouter = HttpRouter.use((router) =>
 						event: payload.event,
 					}),
 				)
-			} else {
-				yield* Effect.logInfo("PlanetScale webhook lifecycle event acknowledged").pipe(
-					Effect.annotateLogs({ orgId: connection.orgId, event: payload.event }),
-				)
 			}
 
 			yield* Effect.annotateCurrentSpan({

apps/api/src/services/alerts/AlertDestinationDelivery.ts

@@ -7,6 +7,7 @@ import {
 	type AlertIncidentId,
 	type AlertRuleId,
 } from "@maple/domain/http"
+import { projectAlertLifecycleEvent } from "@maple/alerting-core"
 import type { AlertDestinationRow } from "@maple/db"
 import { Effect } from "effect"
 import { parseBase64Aes256GcmKey } from "@/platform/Crypto"
@@ -131,8 +132,26 @@ export const makeAlertDestinationDelivery = (options: {
 			{ sendEmail, resolveSlackBotToken: options.resolveSlackBotToken },
 		)
 
-	const buildPayload = (context: AlertDeliveryPayloadContext) =>
+	const buildPayload = (context: AlertDeliveryPayloadContext, tenantId: string) =>
 		({
+			event: projectAlertLifecycleEvent({
+				tenantId,
+				ruleId: context.ruleId,
+				ruleName: context.ruleName,
+				incidentId: context.incidentId,
+				eventType: context.eventType,
+				incidentStatus: context.incidentStatus,
+				groupKey: context.groupKey,
+				signalType: context.signalType,
+				severity: context.severity,
+				comparator: context.comparator,
+				threshold: context.threshold,
+				thresholdUpper: context.thresholdUpper,
+				windowMinutes: context.windowMinutes,
+				value: context.value,
+				sampleCount: context.sampleCount,
+				occurredAtMs: context.sentAtMs,
+			}),
 			eventType: context.eventType,
 			incidentId: context.incidentId,
 			incidentStatus: context.incidentStatus,
@@ -157,6 +176,7 @@ export const makeAlertDestinationDelivery = (options: {
 			chatUrl: buildAlertChatUrl(options.appBaseUrl, context),
 			sentAt: new Date(context.sentAtMs).toISOString(),
 		}) satisfies {
+			readonly event: ReturnType<typeof projectAlertLifecycleEvent>
 			readonly eventType: AlertDeliveryPayloadContext["eventType"]
 			readonly incidentId: AlertIncidentId | null
 			readonly incidentStatus: AlertDeliveryPayloadContext["incidentStatus"]
@@ -181,7 +201,7 @@ export const makeAlertDestinationDelivery = (options: {
 			secretConfig: enrichedSecret,
 			...context,
 		}
-		const payload = buildPayload(fullContext)
+		const payload = buildPayload(fullContext, destinationRow.orgId)
 		return yield* dispatchDelivery(fullContext, JSON.stringify(payload))
 	})