Bootstrap shared production bindings and refine PlanetScale scrape detection

alchemy.run.ts

@@ -1,6 +1,8 @@
 import { appendFileSync } from "node:fs"
 import * as Alchemy from "alchemy"
 import * as Cloudflare from "alchemy/Cloudflare"
+import * as Output from "alchemy/Output"
+import * as RemovalPolicy from "alchemy/RemovalPolicy"
 import * as Effect from "effect/Effect"
 import { formatMapleStage, parseMapleStage, resolveMapleDomains } from "@maple/infra/cloudflare"
 import { createAlertingWorker } from "./apps/alerting/alchemy.run.ts"
@@ -27,6 +29,70 @@ if (!process.env.CLOUDFLARE_ACCOUNT_ID && process.env.CLOUDFLARE_DEFAULT_ACCOUNT
 const resolveUrl = (domain: string | undefined, envKey: string, fallback = ""): string =>
 	domain ? `https://${domain}` : process.env[envKey]?.trim() || fallback
 
+const requireEnv = (key: string): string => {
+	const value = process.env[key]?.trim()
+	if (!value) {
+		throw new Error(`Missing required deployment env: ${key}`)
+	}
+	return value
+}
+
+const createProductionSharedResources = (stage: ReturnType<typeof parseMapleStage>) =>
+	Effect.gen(function* () {
+		// Bootstrap these account/zone-wide resources in production first. Other
+		// stages keep their existing bindings until the production state has been
+		// deployed, after which they can safely resolve these logical IDs via ref().
+		if (stage.kind !== "prd") {
+			return {}
+		}
+
+		const emailRouting = yield* Cloudflare.Email.Routing("email-routing-maple-dev", {
+			zone: "maple.dev",
+		}).pipe(RemovalPolicy.retain())
+		const sendingSubdomain = yield* Cloudflare.Email.SendingSubdomain("email-sending-noreply", {
+			zoneId: emailRouting.zoneId,
+			name: "noreply.maple.dev",
+		}).pipe(RemovalPolicy.retain())
+
+		const ingestEndpoint = (process.env.MAPLE_ENDPOINT?.trim() || "https://ingest.maple.dev").replace(
+			/\/+$/,
+			"",
+		)
+		const headers = { authorization: `Bearer ${requireEnv("MAPLE_OTEL_INGEST_KEY")}` }
+		const tracesDestination = yield* Cloudflare.Workers.ObservabilityDestination(
+			"workers-observability-traces",
+			{
+				name: "maple-workers-traces",
+				url: `${ingestEndpoint}/v1/traces`,
+				headers,
+				logpushDataset: "opentelemetry-traces",
+				enabled: true,
+			},
+		).pipe(RemovalPolicy.retain())
+		const logsDestination = yield* Cloudflare.Workers.ObservabilityDestination(
+			"workers-observability-logs",
+			{
+				name: "maple-workers-logs",
+				url: `${ingestEndpoint}/v1/logs`,
+				headers,
+				logpushDataset: "opentelemetry-logs",
+				enabled: true,
+			},
+		).pipe(RemovalPolicy.retain())
+
+		// The binding descriptor is lazy so enabling the sending subdomain is an
+		// explicit dependency of both Workers that send transactional email.
+		const emailBinding = sendingSubdomain.enabled.pipe(
+			Output.mapEffect(() =>
+				Cloudflare.Email.SendEmail("email", {
+					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
+				}),
+			),
+		)
+
+		return { emailBinding, logsDestination, tracesDestination }
+	})
+
 export default Alchemy.Stack(
 	"maple",
 	{
@@ -40,6 +106,7 @@ export default Alchemy.Stack(
 	Effect.gen(function* () {
 		const stage = parseMapleStage(yield* Alchemy.Stage)
 		const domains = resolveMapleDomains(stage)
+		const shared = yield* createProductionSharedResources(stage)
 
 		const apiUrl = resolveUrl(domains.api, "MAPLE_API_BASE_URL")
 		const chatUrl = resolveUrl(domains.chat, "MAPLE_CHAT_BASE_URL")
@@ -51,8 +118,18 @@ export default Alchemy.Stack(
 		// Deploy the API RPC surface before switching chat-flue to it. The reverse
 		// CHAT_FLUE binding is attached after chat deploys, which breaks the resource
 		// cycle without an HTTP fallback or a placeholder Worker.
-		const { worker: api, db: mapleDb } = yield* createMapleApi({ stage, domains })
-		const chatFlue = yield* createChatFlueWorker({ stage, domains, mapleApiRpc: api })
+		const { worker: api, db: mapleDb } = yield* createMapleApi({
+			stage,
+			domains,
+			emailBinding: shared.emailBinding,
+		})
+		const chatFlue = yield* createChatFlueWorker({
+			stage,
+			domains,
+			mapleApiRpc: api,
+			logsDestination: shared.logsDestination,
+			tracesDestination: shared.tracesDestination,
+		})
 		yield* api.bind("CHAT_FLUE", {
 			bindings: [{ type: "service", name: "CHAT_FLUE", service: chatFlue.workerName }],
 		})
@@ -74,7 +151,12 @@ export default Alchemy.Stack(
 
 		const localUi = yield* createLocalUiWorker({ stage, domains })
 
-		const alerting = yield* createAlertingWorker({ stage, domains, mapleDb })
+		const alerting = yield* createAlertingWorker({
+			stage,
+			domains,
+			mapleDb,
+			emailBinding: shared.emailBinding,
+		})
 
 		const summary = {
 			stage: formatMapleStage(stage),

apps/alerting/alchemy.run.ts

@@ -1,4 +1,5 @@
 import path from "node:path"
+import type { Input } from "alchemy"
 import * as Cloudflare from "alchemy/Cloudflare"
 import * as Effect from "effect/Effect"
 import * as Redacted from "effect/Redacted"
@@ -33,9 +34,11 @@ export interface CreateAlertingWorkerOptions {
 	domains: MapleDomains
 	/** Managed per-branch Hyperdrive from the api factory; undefined on ref stages (stg/prd). */
 	mapleDb: Cloudflare.Hyperdrive.Connection | undefined
+	/** Shared production binding derived from the retained sending subdomain. */
+	emailBinding?: Input<Cloudflare.Email.SendEmail>
 }
 
-export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOptions) =>
+export const createAlertingWorker = ({ stage, mapleDb, emailBinding }: CreateAlertingWorkerOptions) =>
 	Effect.gen(function* () {
 		const hyperdriveRefId = resolveHyperdriveRefId(stage)
 		// Cross-script binding to the AI triage Workflow hosted by the api worker —
@@ -64,9 +67,11 @@ export const createAlertingWorker = ({ stage, mapleDb }: CreateAlertingWorkerOpt
 				// Ref stages attach MAPLE_DB via worker.bind below.
 				...(mapleDb ? { MAPLE_DB: mapleDb } : {}),
 				AI_TRIAGE_WORKFLOW: aiTriageWorkflow,
-				EMAIL: Cloudflare.Email.SendEmail("email", {
-					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
-				}),
+				EMAIL:
+					emailBinding ??
+					Cloudflare.Email.SendEmail("email", {
+						allowedSenderAddresses: ["notifications@noreply.maple.dev"],
+					}),
 				TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"),
 				TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")),
 				MAPLE_AUTH_MODE: process.env.MAPLE_AUTH_MODE?.trim() || "self_hosted",

apps/api/alchemy.run.ts

@@ -1,4 +1,5 @@
 import path from "node:path"
+import type { Input } from "alchemy"
 import * as Cloudflare from "alchemy/Cloudflare"
 import type { Rpc } from "alchemy/Rpc"
 import * as Effect from "effect/Effect"
@@ -34,12 +35,14 @@ const optionalSecret = (key: string): Record<string, Redacted.Redacted<string>>
 export interface CreateMapleApiOptions {
 	stage: MapleStage
 	domains: MapleDomains
+	/** Shared production binding derived from the retained sending subdomain. */
+	emailBinding?: Input<Cloudflare.Email.SendEmail>
 }
 
 /** Alchemy resource type carried across the chat-flue service binding. */
 export type MapleApiWorker = Cloudflare.Worker & Rpc<MapleApiRpcShape>
 
-export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
+export const createMapleApi = ({ stage, domains, emailBinding }: CreateMapleApiOptions) =>
 	Effect.gen(function* () {
 		// MAPLE_DB Hyperdrive comes in two flavors:
 		//
@@ -137,9 +140,11 @@ export const createMapleApi = ({ stage, domains }: CreateMapleApiOptions) =>
 				VCS_SYNC_QUEUE: vcsSyncQueue,
 				CLICKHOUSE_SCHEMA_APPLY_WORKFLOW: schemaApplyWorkflow,
 				AI_TRIAGE_WORKFLOW: aiTriageWorkflow,
-				EMAIL: Cloudflare.Email.SendEmail("email", {
-					allowedSenderAddresses: ["notifications@noreply.maple.dev"],
-				}),
+				EMAIL:
+					emailBinding ??
+					Cloudflare.Email.SendEmail("email", {
+						allowedSenderAddresses: ["notifications@noreply.maple.dev"],
+					}),
 				TINYBIRD_HOST: requireEnv("TINYBIRD_HOST"),
 				TINYBIRD_TOKEN: Redacted.make(requireEnv("TINYBIRD_TOKEN")),
 				...optionalPlain("CLICKHOUSE_URL"),

apps/api/src/services/PlanetScaleConnectionService.test.ts

@@ -68,6 +68,17 @@ const stubPlanetScaleApi = (options?: {
 	 * unknown/bad service tokens) 403, only `token tok_good:*` passes.
 	 */
 	readonly denyBearerMetrics?: boolean
+	/** http_sd payload served by the control-plane SD endpoint (any auth). */
+	readonly sdGroups?: ReadonlyArray<{
+		targets: ReadonlyArray<string>
+		labels?: Record<string, string>
+	}>
+	/**
+	 * Prod-faithful data-plane split: discovered metrics hosts (anything off
+	 * api.planetscale.com) 403 everything but the service-token scheme — the
+	 * behavior that made OAuth-auth'd scrapes fail while the SD probe passed.
+	 */
+	readonly denyDataPlaneBearer?: boolean
 }) => {
 	const organizations = options?.organizations ?? [{ id: "psorg_1", name: "acme" }]
 	const stub = (async (input: string | URL | Request, init?: RequestInit) => {
@@ -92,6 +103,18 @@ const stubPlanetScaleApi = (options?: {
 		if (denied) {
 			return new Response("{}", { status: denied[1], headers: { "content-type": "application/json" } })
 		}
+		if (!requestUrl.includes("api.planetscale.com") && options?.denyDataPlaneBearer) {
+			const authorization = headers.get("authorization") ?? ""
+			return authorization.startsWith("token ")
+				? new Response("up 1\n", { status: 200 })
+				: new Response("forbidden", { status: 403 })
+		}
+		if (options?.sdGroups && requestUrl.includes("api.planetscale.com") && requestUrl.includes("/metrics")) {
+			return new Response(JSON.stringify(options.sdGroups), {
+				status: 200,
+				headers: { "content-type": "application/json" },
+			})
+		}
 		if (options?.denyBearerMetrics && requestUrl.includes("/metrics")) {
 			const authorization = headers.get("authorization") ?? ""
 			if (!authorization.startsWith("token tok_good:")) {
@@ -313,6 +336,83 @@ describe("PlanetScaleConnectionService", () => {
 		)
 	})
 
+	it.effect("pauses metrics when the data plane rejects the bearer despite a passing SD probe", () => {
+		const testDb = createTestDb(trackedDbs)
+		const calls: Array<{ url: string; authorization: string | null }> = []
+		// The prod bug: api.planetscale.com's SD endpoint 2xx'd the OAuth bearer,
+		// so the target auto-enabled — but the discovered metrics.psdb.cloud hosts
+		// only accept service tokens and 403'd every actual scrape.
+		const stub = stubPlanetScaleApi({
+			calls,
+			denyDataPlaneBearer: true,
+			sdGroups: [
+				{
+					targets: ["branch-1.metrics.psdb.cloud:443"],
+					labels: { __metrics_path__: "/metrics", planetscale_database_branch_id: "branch-1" },
+				},
+			],
+		})
+
+		return Effect.gen(function* () {
+			const service = yield* PlanetScaleConnectionService
+			const orgId = asOrgId("org_1")
+
+			yield* storeGrant(orgId)
+			const bound = yield* service.finalizeOrgSelection(orgId, { organization: "acme" })
+
+			// Binding succeeds, but scraping is paused until a service token arrives.
+			assert.isTrue(bound.connected)
+			assert.strictEqual(bound.metricsAuth, "missing")
+			assert.isFalse(bound.scrapeTarget!.enabled)
+			assert.isFalse(bound.detectedPermissions?.readMetricsEndpoints)
+
+			// The verdict came from an actual data-plane scrape probe with the
+			// bearer. (Match on the host: fetch normalizes away the :443 port.)
+			assert.isTrue(
+				calls.some(
+					(call) =>
+						call.url.includes("branch-1.metrics.psdb.cloud") &&
+						call.authorization === "Bearer ps-access-token",
+				),
+			)
+		}).pipe(
+			Effect.provideService(FetchHttpClient.Fetch, stub),
+			Effect.provide(Layer.mergeAll(makeLayer(testDb), Layer.succeed(FetchHttpClient.Fetch, stub))),
+		)
+	})
+
+	it.effect("keeps oauth metrics enabled when the data-plane scrape accepts the bearer", () => {
+		const testDb = createTestDb(trackedDbs)
+		const calls: Array<{ url: string; authorization: string | null }> = []
+		// Data-plane host answers 200 to the bearer (the stub's default) — the
+		// probe must not pause a working OAuth-auth'd target.
+		const stub = stubPlanetScaleApi({
+			calls,
+			sdGroups: [
+				{
+					targets: ["branch-1.metrics.psdb.cloud:443"],
+					labels: { __metrics_path__: "/metrics", planetscale_database_branch_id: "branch-1" },
+				},
+			],
+		})
+
+		return Effect.gen(function* () {
+			const service = yield* PlanetScaleConnectionService
+			const orgId = asOrgId("org_1")
+
+			yield* storeGrant(orgId)
+			const bound = yield* service.finalizeOrgSelection(orgId, { organization: "acme" })
+
+			assert.strictEqual(bound.metricsAuth, "oauth")
+			assert.isTrue(bound.scrapeTarget!.enabled)
+			assert.isTrue(bound.detectedPermissions?.readMetricsEndpoints)
+			assert.isTrue(calls.some((call) => call.url.includes("branch-1.metrics.psdb.cloud")))
+		}).pipe(
+			Effect.provideService(FetchHttpClient.Fetch, stub),
+			Effect.provide(Layer.mergeAll(makeLayer(testDb), Layer.succeed(FetchHttpClient.Fetch, stub))),
+		)
+	})
+
 	it.effect("classifies transient metrics probe failures as upstream errors", () => {
 		const testDb = createTestDb(trackedDbs)
 		const stub = stubPlanetScaleApi({ deny: { "/metrics": 503 } })