fix(infra): one warehouse env for every querying Worker, and small Worker fixes

alchemy.run.ts

@@ -271,6 +271,8 @@ const createDevProcess = (app: DevApp, route: Portless.Route) =>
 			PORT: Output.map(Output.asOutput(route.port), String),
 			PORTLESS_URL: Portless.routeUrl(app),
 			MAPLE_API_URL: Portless.routeUrl("api"),
+			// The scraper's default is a fixed port; ingest's dev script listens on its route's port.
+			MAPLE_INGEST_URL: Portless.routeUrl("ingest"),
 			// The alchemy CLI sets NODE_ENV=production and children inherit it; vite and
 			// astro read it for `import.meta.env.DEV`/`PROD`, so dev servers must override it.
 			NODE_ENV: "development",

apps/ai/src/worker.ts

@@ -15,6 +15,7 @@ import {
 import {
 	appUrlsEnv,
 	authEnv,
+	clickhouseEnv,
 	githubAppSourceEnv,
 	ingestKeyCryptoEnv,
 	merge,
@@ -43,6 +44,8 @@ import { AiObservabilityLive } from "./worker/observability"
 const configuredEnv = ({ stage, region, domains, profile }: MapleStackContext) =>
 	merge(
 		tinybirdEnv,
+		// Same upstream as api, so a tool answers from the warehouse the dashboard reads.
+		clickhouseEnv,
 		authEnv,
 		appUrlsEnv(domains),
 		selfObservabilityEnv(stage, region),

apps/alerting/src/worker.ts

@@ -20,6 +20,7 @@ import {
 	apnsEnv,
 	appUrlsEnv,
 	authEnv,
+	clickhouseEnv,
 	cloudflareOAuthEnv,
 	githubAppSourceEnv,
 	ingestKeyCryptoEnv,
@@ -56,6 +57,8 @@ export type AlertingWorkerEnv = Record<string, unknown>
 const configuredEnv = (stage: MapleStage, region: MapleRegion, domains: MapleDomains) =>
 	merge(
 		tinybirdEnv,
+		// Same upstream as api, so a tick evaluates the warehouse the dashboard reads.
+		clickhouseEnv,
 		authEnv,
 		ingestKeyCryptoEnv,
 		appUrlsEnv(domains),

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

@@ -13,6 +13,7 @@ import {
 	auditEventToRow,
 	decodeAuditLogEvent,
 } from "@maple/backend/services/audit/audit-event"
+import { AUDIT_EVENTS_MAX_RETRIES } from "./queue-retry-limits"
 
 /**
  * The consumer writes through `WarehouseQueryService.ingest`, which pins every
@@ -25,14 +26,6 @@ export const AuditEventsLive = WarehouseQueryService.layer.pipe(
 	Layer.provide(Layer.mergeAll(EventBaseLive, EdgeCacheServiceLive)),
 )
 
-/**
- * Must match `maxRetries` on the audit-events consumer in `worker.ts`.
- * Cloudflare routes the message to the DLQ after this many retries without
- * telling us; the check below is what makes the hand-off visible in logs at
- * the moment it happens.
- */
-const AUDIT_EVENTS_MAX_RETRIES = 5
-
 /**
  * Best-effort identity for the exhaustion log. The body reached us as queue
  * JSON and may be anything at all, so these read defensively rather than
@@ -54,7 +47,9 @@ interface DecodedMessage {
 
 /**
  * Retrying past the limit is what hands the message to the DLQ; acking there
- * would silently discard it instead.
+ * would silently discard it instead. Cloudflare makes that hand-off without
+ * telling us, so the final-attempt check is what puts it in the logs. The
+ * limit is the consumer's own `maxRetries` (`worker/consumers.ts`).
  */
 const retryOrExhaust = (message: Message, cause: unknown) => {
 	const isFinalAttempt = message.attempts > AUDIT_EVENTS_MAX_RETRIES

apps/api/src/cancellation-review-runtime.ts

@@ -6,6 +6,7 @@ import { EventBaseLive } from "@maple/backend/platform/DatabasePgLive"
 import type { QueueBatch } from "@maple/backend/platform/queue-batch"
 import { CancellationReviewJob } from "@maple/backend/services/cancellation-review/CancellationReviewQueue"
 import { CancellationReviewService } from "@maple/backend/services/cancellation-review/CancellationReviewService"
+import { CANCELLATION_REVIEW_MAX_RETRIES } from "./queue-retry-limits"
 
 /** Not `maple-api`: a review's minute of upstream reads would skew the request service's percentiles. */
 export const cancellationReviewTelemetry = eventTelemetry({ serviceName: "maple-cancellation-reviews" })
@@ -18,12 +19,6 @@ export const CancellationReviewLive = CancellationReviewService.layer.pipe(
 	Layer.provide(Layer.mergeAll(EventBaseLive, EdgeCacheServiceLive)),
 )
 
-/**
- * Must match `maxRetries` on the cancellation-reviews consumer
- * (`worker/consumers.ts`). The attempt past it logs the drop and acks.
- */
-const CANCELLATION_REVIEW_MAX_RETRIES = 5
-
 /**
  * 1, 2, 4, 8, 16 minutes: about half an hour for Clerk, Autumn, Postgres or
  * Slack to come back. A review waiting out another delivery's claim (a

apps/api/src/queue-retry-limits.ts

@@ -0,0 +1,8 @@
+/**
+ * Retry limits a queue consumer's settings and its batch handler must agree on: Cloudflare stops
+ * redelivering after `maxRetries` without saying so, and the handler logs the drop on that attempt.
+ * Dependency-free, so `worker/consumers.ts` reads them without loading a runtime module.
+ */
+export const AUDIT_EVENTS_MAX_RETRIES = 5
+
+export const CANCELLATION_REVIEW_MAX_RETRIES = 5

apps/api/src/resources/env.ts

@@ -9,6 +9,7 @@ import {
 	apnsEnv,
 	appUrlsEnv,
 	authEnv,
+	clickhouseEnv,
 	cloudflareOAuthEnv,
 	githubAppSourceEnv,
 	ingestKeyCryptoEnv,
@@ -28,11 +29,7 @@ export const apiConfiguredEnv = (stage: MapleStage, region: MapleRegion, domains
 	merge(
 		tinybirdEnv,
 		// ClickHouse (BYO warehouse); `tinybird` unless an org config overrides it.
-		optionalPlain("CLICKHOUSE_URL"),
-		plainWithDefault("CLICKHOUSE_PROVIDER", "tinybird"),
-		optionalPlain("CLICKHOUSE_USER"),
-		optionalPlain("CLICKHOUSE_DATABASE"),
-		optionalSecret("CLICKHOUSE_PASSWORD"),
+		clickhouseEnv,
 		// Dev-only; the runtime ignores it outside MAPLE_ENVIRONMENT=development.
 		optionalPlain("MAPLE_IGNORE_ORG_CLICKHOUSE"),
 		// Dev only: on a deploy a pin would point the whole API at one tenant. The local stack's

apps/api/src/worker/consumers.ts

@@ -5,7 +5,6 @@
  * layer graph and one Postgres socket for the batch.
  */
 import * as Cloudflare from "@alchemy.run/cloudflare"
-import { renamedFrom } from "alchemy/Rename"
 import { Effect, Layer, Stream } from "effect"
 import { DatabasePgLive } from "@maple/backend/platform/DatabasePgLive"
 import { AuditEventsDlq, AuditEventsQueue } from "@maple/infra/cloudflare"
@@ -17,13 +16,13 @@ import {
 } from "../resources/queues"
 import type { ApiPortsLayer } from "./bindings"
 import type { VcsSyncIsolateCaches } from "@maple/backend/services/integrations/vcs/vcs-sync-isolate-caches"
+import { AUDIT_EVENTS_MAX_RETRIES, CANCELLATION_REVIEW_MAX_RETRIES } from "../queue-retry-limits"
 import { runEvent } from "./events"
 import { auditEventsModule, cancellationReviewModule, planetScaleWebhookModule } from "./modules"
 import { consumeVcsSyncBatch } from "./vcs-sync"
 import { VCS_SYNC_BATCH_SIZE, VCS_SYNC_MAX_RETRIES } from "./vcs-sync-dispatch"
 
-// Consumer settings. The audit consumer's `maxRetries` must stay in sync with
-// AUDIT_EVENTS_MAX_RETRIES in audit-events-runtime.ts, which logs the drop on
+// Consumer settings. Each `maxRetries` is the constant its batch handler reads to log the drop on
 // the final attempt. The VCS one reads its constants from the batch route, which caps a request at them.
 const VCS_SYNC_CONSUMER = {
 	batchSize: VCS_SYNC_BATCH_SIZE,
@@ -33,30 +32,26 @@ const VCS_SYNC_CONSUMER = {
 } satisfies Cloudflare.Queues.MessagesProps
 const PLANETSCALE_WEBHOOKS_CONSUMER = VCS_SYNC_CONSUMER
 // A review is up to a minute of upstream reads and one model call: one per
-// invocation, few at once. The consumer sets each retry's delay; `maxRetries`
-// must match CANCELLATION_REVIEW_MAX_RETRIES in cancellation-review-runtime.ts.
+// invocation, few at once. The consumer sets each retry's delay.
 const CANCELLATION_REVIEWS_CONSUMER = {
 	batchSize: 1,
 	maxConcurrency: 2,
-	maxRetries: 5,
+	maxRetries: CANCELLATION_REVIEW_MAX_RETRIES,
 	maxWaitTime: "1 second",
 } satisfies Cloudflare.Queues.MessagesProps
 // Audit entries tolerate a few seconds of delivery latency; batch wider and
 // wait longer so one insert round-trip covers many entries.
 const auditEventsConsumer = (deadLetterQueue: string | undefined): Cloudflare.Queues.MessagesProps => ({
 	batchSize: 25,
 	maxConcurrency: 2,
-	maxRetries: 5,
+	maxRetries: AUDIT_EVENTS_MAX_RETRIES,
 	maxWaitTime: "5 seconds",
 	deadLetterQueue,
 })
 
 /**
  * Attaches the five consumers to the host at plan time and their listeners
- * at runtime. Needs `Queues.EventSourceLive`. `renamedFrom` carries the
- * consumer resources over from the ids the api factory declared them under,
- * so the deploy migrates their state rows instead of re-creating the
- * consumers; drop it once every stage has deployed past it.
+ * at runtime. Needs `Queues.EventSourceLive`.
  */
 export const registerQueueConsumers = (
 	ports: ApiPortsLayer,
@@ -70,7 +65,7 @@ export const registerQueueConsumers = (
 			Effect.flatMap(Stream.runCollect(stream), (messages) =>
 				consumeVcsSyncBatch(ports, env, vcsSyncCaches, messages),
 			),
-		).pipe(renamedFrom({ fqn: "vcs-sync-consumer" }))
+		)
 		// Same handler, own consumer: a scheduled sync's backlog drains here without holding up webhooks.
 		const vcsBackfillQueue = yield* VcsBackfillQueue
 		yield* Cloudflare.Queues.consumeQueueMessages(vcsBackfillQueue, VCS_SYNC_CONSUMER, (stream) =>
@@ -96,7 +91,7 @@ export const registerQueueConsumers = (
 							),
 						),
 				),
-		).pipe(renamedFrom({ fqn: "planetscale-webhooks-consumer" }))
+		)
 		const cancellationReviewsQueue = yield* CancellationReviewsQueue
 		yield* Cloudflare.Queues.consumeQueueMessages(
 			cancellationReviewsQueue,
@@ -136,5 +131,5 @@ export const registerQueueConsumers = (
 						),
 					),
 				),
-		).pipe(renamedFrom({ fqn: "audit-events-consumer" }))
+		)
 	})

apps/chat-bot/src/routes/webhook.ts

@@ -6,10 +6,8 @@
  * verify a signature, parse a body or know a payload: a connector that cannot
  * authenticate its own caller is a connector that has not been written yet.
  *
- * Nothing registered today delivers this way — the first connector is a socket
- * one — so this exists as the contract's other half, tested against a fake
- * connector. It is also why the Worker takes no public hostname yet; see
- * `../worker.ts`.
+ * A connector of `kind: "webhook"` is delivered this way, which is why the Worker
+ * takes a public hostname where the stage has one; see `../worker.ts`.
  */
 import { connectors as registeredConnectors } from "@maple/chat-platform/connectors"
 import { ConfigProvider, Effect, Option, Schema, Array as Arr } from "effect"

apps/electric-sync/src/electric/ElectricClient.test.ts

@@ -199,21 +199,18 @@ describe("buildUpstreamShapeUrl", () => {
 		assert.isNull(params.get("replica"))
 	})
 
-	it.each(["alert_destinations", "alert_destinations_v2"] as const)(
-		"pins %s's column projection and drops the secret columns",
-		(shape) => {
-			const columns = columnsOf(buildUrl(shape).params)
-			// PK + org scope must be present for Electric.
-			assert.include(columns, "id")
-			assert.include(columns, "org_id")
-			// The public config the browser renders is allowed…
-			assert.include(columns, "config_json")
-			// …but the encrypted secret columns must never be projected.
-			assert.notInclude(columns, "secret_ciphertext")
-			assert.notInclude(columns, "secret_iv")
-			assert.notInclude(columns, "secret_tag")
-		},
-	)
+	it("pins alert_destinations_v2's column projection and drops the secret columns", () => {
+		const columns = columnsOf(buildUrl("alert_destinations_v2").params)
+		// PK + org scope must be present for Electric.
+		assert.include(columns, "id")
+		assert.include(columns, "org_id")
+		// The public config the browser renders is allowed…
+		assert.include(columns, "config_json")
+		// …but the encrypted secret columns must never be projected.
+		assert.notInclude(columns, "secret_ciphertext")
+		assert.notInclude(columns, "secret_iv")
+		assert.notInclude(columns, "secret_tag")
+	})
 
 	it("projects only safe API-key display fields", () => {
 		const { params } = buildUrl("api_keys")
@@ -230,7 +227,7 @@ describe("buildUpstreamShapeUrl", () => {
 
 	it("never lets the client widen a column-restricted shape back to secrets", () => {
 		const columns = columnsOf(
-			buildUrl("alert_destinations", {
+			buildUrl("alert_destinations_v2", {
 				query: "columns=id%2Corg_id%2Csecret_ciphertext%2Csecret_iv%2Csecret_tag",
 			}).params,
 		)

apps/electric-sync/src/shapes/registry.test.ts

@@ -12,10 +12,11 @@ describe("isShapeName", () => {
 	it("accepts whitelisted shapes and rejects everything else", () => {
 		assert.isTrue(isSubscriptionName("dashboards"))
 		assert.isTrue(isSubscriptionName("alert_rules"))
-		assert.isTrue(isSubscriptionName("alert_destinations"))
 		assert.isTrue(isSubscriptionName("alert_destinations_v2"))
+		// Retired once every client moved to `_v2`.
+		assert.isFalse(isSubscriptionName("alert_destinations"))
 		assert.isTrue(isSubscriptionName("api_keys"))
-		// Pruned from both the whitelist and the publication (0022) once their
+		// Pruned from both the whitelist and the publication (electric_publication_prune) once their
 		// client collections were removed — they must not resolve as shapes.
 		assert.isFalse(isSubscriptionName("error_issues"))
 		assert.isFalse(isSubscriptionName("actors"))

apps/electric-sync/src/shapes/registry.ts

@@ -19,13 +19,14 @@ import { Schema, Record as Rec } from "effect"
 // requires the PK in the projection).
 //
 // Every table listed here MUST also be a member of `electric_publication_default`
-// (packages/db/drizzle/0009 + later publication migrations) — Electric runs with
+// (packages/db/drizzle/20260704212005_electric_publication + later publication
+// migrations): Electric runs with
 // ELECTRIC_MANUAL_TABLE_PUBLISHING=true and will not publish a table itself, so a
 // shape over an unpublished table never receives changes. The reverse also holds:
 // a published table with no shape here (and no client collection) is pure
 // replication cost, so drop it from the publication instead of leaving it. That
 // is why the errors vertical (error_issues / actors / open_error_incidents) and
-// scrape_target_checks are gone — see 0022_electric_publication_prune.
+// scrape_target_checks are gone (20260727105910_electric_publication_prune).
 const SUBSCRIPTIONS = {
 	dashboards: { table: "dashboards" },
 	alert_rules: { table: "alert_rules" },
@@ -58,23 +59,8 @@ const SUBSCRIPTIONS = {
 	// workspace); the encrypted webhook secrets live in separate `secret_*` columns
 	// that MUST NOT reach the browser, so the projection drops them (and the
 	// unused `created_by`/`updated_by`). The PK `id` is required in the projection.
-	alert_destinations: {
-		table: "alert_destinations",
-		columns: [
-			"id",
-			"org_id",
-			"name",
-			"type",
-			"enabled",
-			"config_json",
-			"last_tested_at",
-			"last_test_error",
-			"created_at",
-			"updated_at",
-		],
-	},
-	// `_v2` adds the auto-disable state, so the browser can say why Maple stopped
-	// delivering. The v1 shape stays for clients still on the previous build.
+	// `_v2` because `columns` gained the auto-disable state, so the browser can say
+	// why Maple stopped delivering.
 	alert_destinations_v2: {
 		table: "alert_destinations",
 		columns: [

apps/scraper/src/ApiClient.ts

@@ -41,7 +41,11 @@ const encodeUptimeReports = Schema.encodeSync(UptimeCheckReportList)
 const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown))
 const encodeJson = Schema.encodeEffect(Schema.fromJsonString(Schema.Unknown))
 
-/** Not idempotent: a timeout may follow a batch that landed, so only an explicit refusal retries. */
+/**
+ * The in-request retry covers an explicit refusal only: a timeout may follow a batch that landed.
+ * The batch reporter retries a transiently failed batch, so a batch can land twice. It can also
+ * be dropped: the reporter's buffer is bounded and its shutdown flush is timed.
+ */
 const isRefusal = (status: number | null): boolean => status === 429 || status === 503
 
 export class ApiClient extends Context.Service<ApiClient, ApiClientApi>()("@maple/scraper/ApiClient", {

apps/scraper/src/Env.ts

@@ -34,11 +34,11 @@ export interface ScraperEnvConfig {
 	readonly PORT: number
 }
 
-// Defaults target the local dev stack (`bun dev`: api on 3472, ingest on
-// 3474, the docker-compose dev token) so the scraper boots without extra
-// configuration instead of crashing the turbo dev TUI. Production overrides
-// all three (apps/scraper/alchemy.run.ts); a missing override
-// degrades to visible per-reconcile warnings, never a crash loop.
+// Defaults target a standalone run (api on 3472, ingest on 3474, the
+// docker-compose dev token) so the scraper boots without extra configuration.
+// `bun dev` passes both URLs from its routes (root alchemy.run.ts) and
+// production overrides all three (apps/scraper/alchemy.run.ts); a missing
+// override degrades to visible per-reconcile warnings, never a crash loop.
 //
 // Numeric settings ARE validated at layer build, unlike the URLs: a missing
 // URL fails loudly every reconcile, but SCRAPER_CONCURRENCY=0 builds a

apps/status/src/handler.test.ts

@@ -7,7 +7,7 @@ import {
 	makeSnapshot,
 } from "@maple/status-page/fixtures"
 import { beforeEach, describe, expect, it } from "vitest"
-import { cacheKey } from "./snapshot"
+import { CACHE_FRESH_MS, cacheKey } from "./snapshot"
 import {
 	type FakeStorage,
 	makeFakeStorage,
@@ -111,6 +111,28 @@ describe("serving a page", () => {
 		expect((await send(fake, "https://status.maple.workers.dev/s/unknown/")).status).toBe(404)
 	})
 
+	it("asks KV for the shared host once per cache window, not on every slug request", async () => {
+		const url = "https://status.maple.workers.dev/s/acme/"
+		const hostKey = StatusPageKeys.host("status.maple.workers.dev")
+		const hostReads = () => Arr.filter(fake.kvReads, (key) => key === hostKey).length
+		expect((await send(fake, url)).status).toBe(200)
+		expect((await send(fake, url)).status).toBe(200)
+		expect(hostReads()).toBe(1)
+		expect((await send(fake, url, { nowMs: FIXTURE_NOW + CACHE_FRESH_MS })).status).toBe(200)
+		expect(hostReads()).toBe(2)
+	})
+
+	it("serves a host that gains a page once its remembered miss has aged out", async () => {
+		const page = "https://status.late.example/"
+		expect((await send(fake, page)).status).toBe(404)
+		fake.kv.set(StatusPageKeys.host("status.late.example"), FIXTURE_PAGE_ID)
+		expect((await send(fake, page, { nowMs: FIXTURE_NOW + CACHE_FRESH_MS })).status).toBe(200)
+	})
+
+	it("keeps a host's own page ahead of the slug path, so it never serves another page", async () => {
+		expect((await send(fake, "https://status.acme.example/s/acme/")).status).toBe(404)
+	})
+
 	it("stops serving a renamed page under its old slug", async () => {
 		fake.kv.set(StatusPageKeys.slug("old-acme"), FIXTURE_PAGE_ID)
 		expect((await send(fake, "https://status.maple.workers.dev/s/old-acme/")).status).toBe(404)