diff --git a/alchemy.run.ts b/alchemy.run.ts index 4415dd10c3..ea2c9e57c4 100644 --- a/alchemy.run.ts +++ b/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", diff --git a/apps/ai/src/worker.ts b/apps/ai/src/worker.ts index 7d34085f76..e2fa0a4736 100644 --- a/apps/ai/src/worker.ts +++ b/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), diff --git a/apps/alerting/src/worker.ts b/apps/alerting/src/worker.ts index 67e44a2a14..893fa3c296 100644 --- a/apps/alerting/src/worker.ts +++ b/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 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), diff --git a/apps/api/src/audit-events-runtime.ts b/apps/api/src/audit-events-runtime.ts index 3c3418061e..b1f9aa4165 100644 --- a/apps/api/src/audit-events-runtime.ts +++ b/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 diff --git a/apps/api/src/cancellation-review-runtime.ts b/apps/api/src/cancellation-review-runtime.ts index 164e85f336..f0459eab2d 100644 --- a/apps/api/src/cancellation-review-runtime.ts +++ b/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 diff --git a/apps/api/src/queue-retry-limits.ts b/apps/api/src/queue-retry-limits.ts new file mode 100644 index 0000000000..e1178a062a --- /dev/null +++ b/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 diff --git a/apps/api/src/resources/env.ts b/apps/api/src/resources/env.ts index fe5110bc1a..64650ad2ca 100644 --- a/apps/api/src/resources/env.ts +++ b/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 diff --git a/apps/api/src/worker/consumers.ts b/apps/api/src/worker/consumers.ts index e47cf5a780..7846c68d51 100644 --- a/apps/api/src/worker/consumers.ts +++ b/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,12 +32,11 @@ 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 @@ -46,17 +44,14 @@ const CANCELLATION_REVIEWS_CONSUMER = { 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" })) + ) }) diff --git a/apps/chat-bot/src/routes/webhook.ts b/apps/chat-bot/src/routes/webhook.ts index 2b6b076508..b69998d623 100644 --- a/apps/chat-bot/src/routes/webhook.ts +++ b/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" diff --git a/apps/electric-sync/src/electric/ElectricClient.test.ts b/apps/electric-sync/src/electric/ElectricClient.test.ts index 41a99eca75..bf5fa96405 100644 --- a/apps/electric-sync/src/electric/ElectricClient.test.ts +++ b/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, ) diff --git a/apps/electric-sync/src/shapes/registry.test.ts b/apps/electric-sync/src/shapes/registry.test.ts index caad0635f6..42c7eaf5ec 100644 --- a/apps/electric-sync/src/shapes/registry.test.ts +++ b/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")) diff --git a/apps/electric-sync/src/shapes/registry.ts b/apps/electric-sync/src/shapes/registry.ts index 0771b497c4..2974eddb83 100644 --- a/apps/electric-sync/src/shapes/registry.ts +++ b/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: [ diff --git a/apps/scraper/src/ApiClient.ts b/apps/scraper/src/ApiClient.ts index 10b9c6fb99..26805c13c8 100644 --- a/apps/scraper/src/ApiClient.ts +++ b/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()("@maple/scraper/ApiClient", { diff --git a/apps/scraper/src/Env.ts b/apps/scraper/src/Env.ts index 3b2d1ec926..18b2200964 100644 --- a/apps/scraper/src/Env.ts +++ b/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 diff --git a/apps/status/src/handler.test.ts b/apps/status/src/handler.test.ts index 5ede6e4d14..2147d0f9b4 100644 --- a/apps/status/src/handler.test.ts +++ b/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) diff --git a/apps/status/src/snapshot.ts b/apps/status/src/snapshot.ts index f97a86691c..2652141bb9 100644 --- a/apps/status/src/snapshot.ts +++ b/apps/status/src/snapshot.ts @@ -48,9 +48,13 @@ const decodeFound = ( ): Option.Option> => raw === null ? Option.none() : Option.map(decode(raw), (value) => ({ raw, value })) +/** Cached in place of a value when KV answered that the key does not exist. */ +const KV_MISS = "missing" + /** * Cache (fresh) -> KV -> R2 -> cache (any age). An undecodable value is a miss, and a value read - * from KV or R2 refreshes the cache. With `kvMissIsFinal`, a KV `null` ends the search. + * from KV or R2 refreshes the cache. With `kvMissIsFinal`, a KV `null` ends the search, and with + * `rememberMiss` that answer is cached for as long as a hit is. */ const readThrough = ( env: StatusEnv, @@ -59,6 +63,7 @@ const readThrough = ( readonly kvKey: string readonly r2Key: string | null readonly kvMissIsFinal: boolean + readonly rememberMiss: boolean readonly decode: (raw: string) => Option.Option }, ): Effect.Effect> => @@ -66,8 +71,9 @@ const readThrough = ( const { storage } = env const cached = yield* storage.cacheGet(input.cacheKey).pipe(Effect.orElseSucceed(() => null)) const fromCache = decodeFound(cached?.value ?? null, input.decode) - if (cached !== null && Option.isSome(fromCache) && env.nowMs - cached.storedAt < CACHE_FRESH_MS) { - return found(fromCache.value.value) + if (cached !== null && env.nowMs - cached.storedAt < CACHE_FRESH_MS) { + if (Option.isSome(fromCache)) return found(fromCache.value.value) + if (input.rememberMiss && cached.value === KV_MISS) return MISSING } const remember = (hit: Found) => storage @@ -77,7 +83,15 @@ const readThrough = ( const kv = yield* Effect.result(storage.kvGet(input.kvKey)) const fromKv = Result.isSuccess(kv) ? decodeFound(kv.success, input.decode) : Option.none() if (Option.isSome(fromKv)) return yield* remember(fromKv.value) - if (input.kvMissIsFinal && Result.isSuccess(kv) && kv.success === null) return MISSING + if (input.kvMissIsFinal && Result.isSuccess(kv) && kv.success === null) { + // Never over a cached value: that copy is the fallback for when KV and R2 both fail. + if (input.rememberMiss && Option.isNone(fromCache)) { + yield* storage + .cachePut(input.cacheKey, { value: KV_MISS, storedAt: env.nowMs }) + .pipe(Effect.ignore) + } + return MISSING + } const r2 = input.r2Key === null ? null : yield* Effect.result(storage.r2Get(input.r2Key)) const fromR2 = @@ -93,13 +107,17 @@ const decodePageId = (raw: string): Option.Option => { return /^[0-9a-f-]{36}$/u.test(id) ? Option.some(id) : Option.none() } -/** The page a hostname serves: KV `host:`, R2 `hosts/` when KV fails, then any cached answer. */ +/** + * The page a hostname serves: KV `host:`, R2 `hosts/` when KV fails, then any cached answer. + * A miss is remembered: the shared host has no page, and every `/s/` request asks it first. + */ export const pageIdForHost = (env: StatusEnv, hostname: string): Effect.Effect> => readThrough(env, { cacheKey: cacheKey("host", Str.toLowerCase(hostname)), kvKey: StatusPageKeys.host(hostname), r2Key: StatusPageKeys.hostObject(hostname), kvMissIsFinal: true, + rememberMiss: true, decode: decodePageId, }) @@ -110,6 +128,7 @@ export const pageIdForSlug = (env: StatusEnv, slug: string): Effect.Effect (lookup._tag === "found" ? Option.some(lookup.value) : Option.none()))) diff --git a/apps/status/src/test-support.ts b/apps/status/src/test-support.ts index 14c32ebff8..2e562c45df 100644 --- a/apps/status/src/test-support.ts +++ b/apps/status/src/test-support.ts @@ -39,8 +39,14 @@ export const makeFakeStorage = () => { const unlisted = new Set() // Runs once inside the next `list`, to land a concurrent write while it is in flight. const duringList: ListHook = { run: null } + // Every key a KV `get` asked for, in order. + const kvReads = new Array() const storage: StatusStorage = { - kvGet: (key) => guard("kv", key, () => kv.get(key) ?? null), + kvGet: (key) => + guard("kv", key, () => { + kvReads.push(key) + return kv.get(key) ?? null + }), kvPut: (key, value) => guard("kv-put", key, () => void kv.set(key, value)), kvList: (prefix, limit) => guard("kv", prefix, () => { @@ -66,7 +72,7 @@ export const makeFakeStorage = () => { } // Shared across requests, so two posts get distinct entry ids. const ids = { next: 0 } - return { storage, kv, r2, cache, failing, limits, unlisted, duringList, ids } + return { storage, kv, kvReads, r2, cache, failing, limits, unlisted, duringList, ids } } export type FakeStorage = ReturnType diff --git a/docs/electric-sync.md b/docs/electric-sync.md index a693e6b1a3..760196f66b 100644 --- a/docs/electric-sync.md +++ b/docs/electric-sync.md @@ -61,8 +61,7 @@ writes: endpoint captures the Postgres txid on the mutating statement | `alert_rules` | alert_rules | none | | `alert_rule_states` | alert_rule_states | none | | `alert_incidents` | alert_incidents | none | -| `alert_destinations` | alert_destinations | columns: drops the encrypted `secret_*` (previous builds only) | -| `alert_destinations_v2` | alert_destinations | columns: as v1, plus `disabled_at` / `disabled_reason` | +| `alert_destinations_v2` | alert_destinations | columns: drops the encrypted `secret_*` | | `api_keys` | api_keys | columns: drops `key_hash` / `metadata_json` | | `investigation_v2` | investigations | scoped to one `id` (client supplies the value as `$2`); columns | diff --git a/docs/infra.md b/docs/infra.md index 8cde419eb7..e33a7d52d1 100644 --- a/docs/infra.md +++ b/docs/infra.md @@ -237,11 +237,10 @@ What each kind of Worker keeps beside the module: - **Queues** (`api`): `Cloudflare.Queues.consumeQueueMessages(queue, settings, handler)` in `impl`, under `Queues.EventSourceLive`, yields the `Consumer` resource at plan time and the listener at runtime (`src/worker/consumers.ts`, over the declarations in - `src/resources/queues.ts`). The consumers carry `renamedFrom({ fqn })` with the ids the - retired api factory declared them under (`vcs-sync-consumer`, …). Alchemy migrates the - state rows, so the deploy plans a noop instead of re-creating each consumer. Without it, - the delete of the old row would have removed the physical consumer the new row had - adopted. Drop the decoration once every stage has deployed past it. + `src/resources/queues.ts`). Renaming a consumer's id needs `renamedFrom({ fqn })` from + `alchemy/Rename` until every stage has deployed past it: alchemy then migrates the state + row and plans a noop. Without it, the delete of the old row removes the physical consumer + the new row had adopted. - **Background telemetry** (`api`): queue batches and cron ticks run under their own SDK instance (`eventTelemetry` in `@maple/infra/worker-telemetry`, provided around the event), so `maple-vcs-sync` and `maple-planetscale-webhooks` keep their own service names. diff --git a/packages/backend/src/platform/EmailService.ts b/packages/backend/src/platform/EmailService.ts index f261841620..325b0fcbf1 100644 --- a/packages/backend/src/platform/EmailService.ts +++ b/packages/backend/src/platform/EmailService.ts @@ -41,7 +41,7 @@ export class EmailService extends Context.Service // duplicate copies of every cron-driven email. The alchemy configs no // longer attach EMAIL outside prd; this guard covers any binding that // still reaches a non-prod worker (`alchemy dev`, manual deploys). - const emailAllowed = env.MAPLE_ENVIRONMENT === "production" || env.MAPLE_EMAIL_ALLOW_NONPROD + const emailAllowed = env.MAPLE_ENVIRONMENT === "production" const isConfigured = Option.isSome(sender) && emailAllowed const send = Effect.fn("EmailService.send")(function* ( @@ -62,7 +62,7 @@ export class EmailService extends Context.Service if (!emailAllowed) { return yield* new EmailDeliveryError({ - message: `Email suppressed: sends are disabled in ${env.MAPLE_ENVIRONMENT} (set MAPLE_EMAIL_ALLOW_NONPROD=true to override)`, + message: `Email suppressed: sends are disabled in ${env.MAPLE_ENVIRONMENT}`, }) } diff --git a/packages/backend/src/platform/Env.ts b/packages/backend/src/platform/Env.ts index 1571dcb1bc..fb4c3ba431 100644 --- a/packages/backend/src/platform/Env.ts +++ b/packages/backend/src/platform/Env.ts @@ -54,8 +54,6 @@ export interface EnvConfig { readonly MAPLE_ENVIRONMENT: string /** The regional instance this worker belongs to, derived by the stack from the stage (`prd-eu` → `eu`). */ readonly MAPLE_REGION: MapleRegion - /** Escape hatch: allow real email sends outside production (e.g. a deliberate test run on a dev stage). */ - readonly MAPLE_EMAIL_ALLOW_NONPROD: boolean /** Route every org to the managed warehouse; honoured only in development. */ readonly MAPLE_IGNORE_ORG_CLICKHOUSE: boolean readonly CLERK_SECRET_KEY: Option.Option @@ -247,9 +245,6 @@ const envConfig = Config.all({ MAPLE_API_BASE_URL: stringWithDefault("MAPLE_API_BASE_URL", "http://127.0.0.1:3472"), MAPLE_ENVIRONMENT: stringWithDefault("MAPLE_ENVIRONMENT", "development"), MAPLE_REGION: stringWithDefault("MAPLE_REGION", "us").pipe(Config.map(parseMapleRegion)), - MAPLE_EMAIL_ALLOW_NONPROD: stringWithDefault("MAPLE_EMAIL_ALLOW_NONPROD", "false").pipe( - Config.map((value) => value === "true"), - ), MAPLE_IGNORE_ORG_CLICKHOUSE: stringWithDefault("MAPLE_IGNORE_ORG_CLICKHOUSE", "false").pipe( Config.map((value) => value === "1" || value === "true"), ), diff --git a/packages/infra/src/env.ts b/packages/infra/src/env.ts index ecd9442fe6..d3ceb04829 100644 --- a/packages/infra/src/env.ts +++ b/packages/infra/src/env.ts @@ -107,6 +107,18 @@ export const tinybirdEnv: Config.Config = merge( optionalPlain("TINYBIRD_RAW_SQL_JWT_RPS_LIMIT"), ) +/** + * The managed ClickHouse read endpoint; `WarehouseQueryService` reads through it when + * `CLICKHOUSE_URL` is set. Shared by every Worker that queries, so they all read one warehouse. + */ +export const clickhouseEnv: Config.Config = merge( + optionalPlain("CLICKHOUSE_URL"), + plainWithDefault("CLICKHOUSE_PROVIDER", "tinybird"), + optionalPlain("CLICKHOUSE_USER"), + optionalPlain("CLICKHOUSE_DATABASE"), + optionalSecret("CLICKHOUSE_PASSWORD"), +) + /** Ingest-key envelope encryption + lookup HMAC. Required wherever ingest keys are read. */ export const ingestKeyCryptoEnv: Config.Config = merge( requireSecretEntry("MAPLE_INGEST_KEY_ENCRYPTION_KEY"),