The requests that belong to nobody

for r in "GET /healthz" "GET /v1/webhooks" "GET /v1/does-not-exist" \
         "GET /v1/channels/abc123/messages" "POST /internal/dispatch/expand"; do
  set -- $r
  printf "%-6s %-34s -> %s\n" "$1" "$2" \
    "$(curl -s -o /dev/null -w '%{http_code}' -X "$1" "localhost:4000$2")"
done

export interface PlatformPrincipal {
  kind: "platform";
  /** Which internal service presented it, for logs. Never the credential. */
  service: string;
  /** Present and always undefined, so the environment-scoped providers that read
   *  `principal?.environmentId` keep compiling AND keep getting nothing. */
  environmentId?: undefined;
}

consumer
  .apply(
    RequestContextMiddleware,
    RequestLogMiddleware,
    AuthenticateMiddleware,
    RateLimitMiddleware,
  )
  .forRoutes("{*path}");

export function operationsFor(method: string, path: string): LimitedOperation[] {
  if (!path.startsWith(PUBLIC_PREFIX)) return [];
  if (method === "POST" && SEND_PATH.test(path)) return ["rest", "send"];
  return ["rest"];
}

@@ -82,6 +82,144 @@ export function shape(raw: unknown): AttemptRow | null {
     status: isNumber(e.status) ? e.status : null,
     error: isString(e.error) ? e.error : null,
     latency_ms: e.latency_ms,
     outcome: e.outcome,
   };
 }
+
+/** The api's record, as it arrives on `analytics.api.request.{env|_none}`. */
+export interface RequestEvent {
+  type: string;
+  request_id: string;
+  ts: string;
+  method: string;
+  status: number;
+  latency_ms: number;
+  principal_kind: string;
+  refused_at: string;
+  /** Absent when the router did not run -- a 404 matched nothing, and a middleware refusal
+   *  ended the response before routing. Two different facts, separated by `refused_at`. */
+  endpoint?: string;
+  /** Absent when the request resolved to no tenant. Never null and never a sentinel. */
+  environment_id?: string;
+  /** Absent unless the rate limiter refused. */
+  limited_operation?: string;
+}
+
+/** One row of `relay_analytics.api_requests`, keyed by column name. */
+export interface RequestRow {
+  environment_id: string | null;
+  ts: string;
+  request_id: string;
+  endpoint: string | null;
+  method: string;
+  status: number;
+  latency_ms: number;
+  principal_kind: string;
+  refused_at: string;
+  limited_operation: string | null;
+}
+
+/** Shape one API request record, or return null if it will never be valid.
+ *
+ * `type` IS READ AND THEN DROPPED. The wire record has eleven fields and the table has ten:
+ * `type` is the router's discriminator and has no column. Forwarding it lands
+ * `Code: 117. Unknown field found while parsing JSONEachRow format: type` -- loud, because
+ * `input_format_skip_unknown_fields=0` is set, and the one place in this design where a
+ * spread fails loudly rather than open. The row is still built by naming every field:
+ * relying on a server setting to catch a shaping mistake is relying on it to be configured.
+ *
+ * ABSENT IS NULL, NOT "". The columns are `LowCardinality(Nullable(String))` because
+ * `LowCardinality(String)` cannot hold the difference -- measured, an absent field and an
+ * explicit empty string both land as ''. A 404 has no endpoint; a route named "" does not
+ * exist. */
+export function shapeRequest(raw: unknown): RequestRow | null {
+  if (typeof raw !== "object" || raw === null) return null;
+  const e = raw as Partial<RequestEvent>;
+
+  if (
+    !isString(e.request_id) ||
+    !isString(e.ts) ||
+    !isString(e.method) ||
+    !isNumber(e.status) ||
+    !isNumber(e.latency_ms) ||
+    !isString(e.principal_kind) ||
+    !isString(e.refused_at)
+  ) {
+    return null;
+  }
+
+  return {
+    environment_id: isString(e.environment_id) ? e.environment_id : null,
+    ts: e.ts,
+    request_id: e.request_id,
+    endpoint: isString(e.endpoint) ? e.endpoint : null,
+    method: e.method,
+    status: e.status,
+    latency_ms: e.latency_ms,
+    principal_kind: e.principal_kind,
+    refused_at: e.refused_at,
+    limited_operation: isString(e.limited_operation) ? e.limited_operation : null,
+  };
+}
+
+// ---------------------------------------------------------------------------
+// ROUTING (chapter 4.4). The stream carries more than one record type now.
+//
+// `shape` above is the ATTEMPT shaper, and it answers `null` for anything else — which the
+// consumer reads as "terminate". So until this chapter, publishing a second record type onto
+// `ANALYTICS` destroyed it: one error line carrying a stream sequence, `m.term()`, and it
+// never comes back. Measured at 049 phase 1, with an attempt record beside it as the control:
+//
+//     pass 1: written 1  malformed 1
+//     pass 2: written 0  malformed 0
+//     stream still holds 2 of 2 · consumer num_pending 0 · ack_pending 0
+//
+// BOTH INSTRUMENTS REPORT NOTHING WRONG. The stream says the record is there, because
+// `retention: Limits` keeps a terminated message; the consumer says there is nothing pending.
+// The only trace is the error line, and it carries a sequence and no type.
+//
+// So the decision "what is this record" is made BEFORE anything tries to turn it into a row,
+// and "not mine" stops being the same answer as "malformed". One is a record this consumer
+// does not write and must leave alone; the other will never parse and must not come back.
+// ---------------------------------------------------------------------------
+
+/** The wire's own discriminator. R11: the PAYLOAD says what the payload is, not the subject —
+ *  a router that parses subjects has to be right about tokens too, and a malformed token
+ *  publishes a subject one level deeper that no intended filter matches. */
+export const API_REQUEST_TYPE = "api.request";
+
+export type Shaped =
+  | { kind: "attempt"; row: AttemptRow }
+  | { kind: "request"; row: RequestRow }
+  | { kind: "malformed" }
+  | { kind: "unclaimed"; type: string };
+
+/** Decide what a record is.
+ *
+ * AN ABSENT `type` MEANS ATTEMPT, AND THAT IS A COMPATIBILITY RULE RATHER THAN A DEFAULT.
+ * Chapter 3.20's publisher has no `type` field and never will for the records already on the
+ * stream — 36 of them at this chapter's tag, 0 carrying one. A reader of anything durable
+ * cannot require a field its writer did not have; 043 paid for that lesson when a required
+ * `attachments` terminated every in-flight `message.created` written by the previous binary.
+ *
+ * A RECOGNISED TYPE WITH MISSING FIELDS IS STILL MALFORMED. Widening "not mine" must not
+ * swallow the parse arm: 048's rule is retry forever on transport, terminate at parse, and
+ * the poison case is what makes the unbounded redelivery safe. */
+export function route(raw: unknown): Shaped {
+  if (typeof raw !== "object" || raw === null) return { kind: "malformed" };
+
+  const type = (raw as { type?: unknown }).type;
+  if (type === undefined) {
+    const row = shape(raw);
+    return row === null ? { kind: "malformed" } : { kind: "attempt", row };
+  }
+  if (type === API_REQUEST_TYPE) {
+    const row = shapeRequest(raw);
+    return row === null ? { kind: "malformed" } : { kind: "request", row };
+  }
+
+  // Anything else is somebody's record and not this consumer's. Leaving it costs the stream's
+  // retention window; terminating it costs the record. Those are not comparable, and a
+  // consumer that does not recognise a type is the party with the least information.
+  return { kind: "unclaimed", type: typeof type === "string" ? type : String(type) };
+}

@@ -9,27 +9,21 @@
 // INCONVENIENCE. That runtime exists because "a future consumer forgets to dedupe → double
 // webhooks / double metering", mitigated by "a consumer template with dedup built in" -- and
 // the dedup it has built in is `claimEvent`, a PostgreSQL transaction. Constitution III
 // keeps the operational and analytical paths apart, so the template written to describe this
 // consumer is the one thing this consumer may not use. Deduplication happens in the table
 // instead, on the record's own key.
-import { AckPolicy, connect, type NatsConnection } from "nats";
+import { AckPolicy, connect } from "nats";
 
 import { ALL_ANALYTICS_SUBJECT, ANALYTICS_STREAM } from "@relay/protocol";
-import { createLogger, type Logger } from "@relay/service-kit";
+import { createLogger } from "@relay/service-kit";
 
-import { createClickHouse, type ClickHouse } from "./clickhouse.js";
-import { shape, type AttemptRow } from "./shape.js";
+import { createClickHouse } from "./clickhouse.js";
+import { BATCH_MS, DURABLE, ingestOnce } from "./ingest.js";
 
 const DEFAULT_NATS_URL = "nats://localhost:4222";
-export const DURABLE = "analytics-ingester";
-
-// DR-11 publishes 2 s or 10,000 rows, and it takes BOTH because each bound fails alone: a
-// row count never flushes for a quiet tenant, and an interval has no ceiling under load.
-export const BATCH_ROWS = 10_000;
-export const BATCH_MS = 2_000;
 
 const ACK_WAIT_NS = 30 * 1_000_000_000;
 
 /** NO REDELIVERY LIMIT, AND THAT IS A DECISION ABOUT A DIFFERENT FAILURE.
  *
  * Both existing consumers set one -- MAX_DELIVER = 5 on the api's runtime, 10 on the
@@ -46,76 +40,12 @@ const ACK_WAIT_NS = 30 * 1_000_000_000;
  * So the queue's seven-day retention is the only bound. That makes the poison case
  * load-bearing rather than tidy: a payload that will never parse is terminated at the parse,
  * or it comes back until the retention expires. Retry forever on transport, terminate at
  * parse. One rule, two arms. */
 const MAX_DELIVER = -1;
 
-export interface IngestResult {
-  written: number;
-  malformed: number;
-}
-
-export async function ingestOnce({
-  nc,
-  store,
-  logger,
-  batchRows = BATCH_ROWS,
-  batchMs = BATCH_MS,
-  stream = ANALYTICS_STREAM,
-  durable = DURABLE,
-}: {
-  nc: NatsConnection;
-  store: ClickHouse;
-  logger: Logger;
-  batchRows?: number;
-  batchMs?: number;
-  /** The stream and durable are parameters so a test can use its own rather than
-   *  publishing probe records into the platform's. The defaults are the real ones. */
-  stream?: string;
-  durable?: string;
-}): Promise<IngestResult> {
-  const js = nc.jetstream();
-  const consumer = await js.consumers.get(stream, durable);
-
-  const rows: AttemptRow[] = [];
-  const pending: Array<{ ack: () => void }> = [];
-  let malformed = 0;
-
-  const messages = await consumer.fetch({ max_messages: batchRows, expires: batchMs });
-  for await (const m of messages) {
-    let parsed: unknown;
-    try {
-      parsed = JSON.parse(new TextDecoder().decode(m.data));
-    } catch {
-      parsed = null;
-    }
-    const row = shape(parsed);
-    if (row === null) {
-      // NAMED BY ITS SEQUENCE, NEVER BY ITS CONTENTS. The record carries `error` -- up to
-      // 2000 characters of a third-party endpoint's response, capable of echoing back
-      // anything -- and constitution VI keeps secrets, tokens and message content out of
-      // logs. The record survives in the stream for the retention window, so the sequence
-      // is enough to go and fetch it deliberately, which is the difference between an
-      // investigation and a leak.
-      malformed += 1;
-      logger.log("error", "ingester.malformed_record", { stream_sequence: m.seq });
-      m.term();
-      continue;
-    }
-    rows.push(row);
-    pending.push({ ack: () => m.ack() });
-  }
-
-  // ACKNOWLEDGE ONLY AFTER THE INSERT RETURNS. A record that was not written is not
-  // acknowledged, which is what makes the store being unreachable a delay rather than a loss.
-  await store.insert(rows);
-  for (const p of pending) p.ack();
-
-  return { written: rows.length, malformed };
-}
-
 export async function main(): Promise<void> {
   const logger = createLogger("ingester");
   const url = process.env["RELAY_NATS_URL"] ?? DEFAULT_NATS_URL;
   const once = process.argv.includes("--once");
 
   const nc = await connect({ servers: url });
@@ -135,15 +65,25 @@ export async function main(): Promise<void> {
   };
   process.once("SIGTERM", stop);
   process.once("SIGINT", stop);
 
   do {
     try {
-      const { written, malformed } = await ingestOnce({ nc, store, logger });
-      if (written > 0 || malformed > 0) {
-        logger.log("info", "ingester.batch", { written, malformed });
+      const r = await ingestOnce({ nc, store, logger });
+      // `unclaimed` is in the line because a record nobody claims is redelivered until the
+      // stream's retention expires, and the count is the only way anyone finds out that is
+      // happening. The two `written` figures are separate because one number cannot say
+      // which table moved.
+      if (r.written > 0 || r.malformed > 0 || r.unclaimed > 0) {
+        logger.log("info", "ingester.batch", {
+          written: r.written,
+          attempts: r.writtenAttempts,
+          requests: r.writtenRequests,
+          malformed: r.malformed,
+          unclaimed: r.unclaimed,
+        });
       }
     } catch (error) {
       // The store is unreachable, or the insert was refused. Nothing was acknowledged, so
       // the broker will offer these records again -- forever, bounded only by retention.
       logger.log("error", "ingester.batch_failed", { error: String(error) });
       await new Promise((r) => setTimeout(r, BATCH_MS));

@@ -1,12 +1,13 @@
 // The write side. Node's own `fetch` against the HTTP interface -- no client package, which
 // is what keeps `grep -c clickhouse pnpm-lock.yaml` at 0 by design rather than by luck.
-import type { AttemptRow } from "./shape.js";
+import type { AttemptRow, RequestRow } from "./shape.js";
 
 const DB = "relay_analytics";
-const TABLE = "webhook_attempts";
+const ATTEMPTS = "webhook_attempts";
+const REQUESTS = "api_requests";
 
 // TWO SETTINGS, TWO DIFFERENT FAILURES, AND NEITHER IS OPTIONAL.
 //
 // `input_format_skip_unknown_fields=0` turns a RENAMED field into `Code: 117` instead of a
 // silent default. Its server default is 1, which is exactly why the failure it prevents was
 // invisible: the insert succeeds and the column takes the epoch.
@@ -17,13 +18,17 @@ const TABLE = "webhook_attempts";
 //
 // Neither covers an ABSENT field. The table's `CHECK ts_is_real` does.
 const SETTINGS = "input_format_skip_unknown_fields=0&date_time_input_format=best_effort";
 
 export interface ClickHouse {
   insert(rows: AttemptRow[]): Promise<void>;
+  /** The second table (chapter 4.4). A separate call rather than a `table` parameter: the two
+   *  row shapes are different types and the compiler should say so at the call site. */
+  insertRequests(rows: RequestRow[]): Promise<void>;
   count(): Promise<number>;
+  countRequests(): Promise<number>;
 }
 
 export function createClickHouse({
   host = process.env["RELAY_CLICKHOUSE_HOST"] ?? "localhost",
   port = process.env["RELAY_CLICKHOUSE_HTTP_PORT"] ?? "8123",
   user = process.env["RELAY_CLICKHOUSE_USER"] ?? "relay",
@@ -49,17 +54,31 @@ export function createClickHouse({
     // keyed on (environment_id, ts, delivery_id, attempt), so a re-inserted record collapses
     // regardless of how it was batched -- which a token cannot do, because JetStream batch
     // boundaries are not stable across a redelivery.
     async insert(rows: AttemptRow[]): Promise<void> {
       if (rows.length === 0) return;
       await post(
-        `INSERT INTO ${DB}.${TABLE} FORMAT JSONEachRow`,
+        `INSERT INTO ${DB}.${ATTEMPTS} FORMAT JSONEachRow`,
+        rows.map((r) => JSON.stringify(r)).join("\n"),
+      );
+    },
+    // THE EMPTY GUARD IS LOAD-BEARING NOW, WHERE IT WAS TIDINESS BEFORE. One fetch feeds two
+    // tables, and the two producers differ by about two orders of magnitude -- so most
+    // batches carry requests and no attempts. Without this, every one of them would post an
+    // empty INSERT: a round trip and a part that never had to exist.
+    async insertRequests(rows: RequestRow[]): Promise<void> {
+      if (rows.length === 0) return;
+      await post(
+        `INSERT INTO ${DB}.${REQUESTS} FORMAT JSONEachRow`,
         rows.map((r) => JSON.stringify(r)).join("\n"),
       );
     },
     // Reads take FINAL. The duplicate is physically present until a merge collapses it, so a
     // bare count over-counts every redelivery -- by a plausible number.
     async count(): Promise<number> {
-      return Number(await post(`SELECT count() FROM ${DB}.${TABLE} FINAL`, ""));
+      return Number(await post(`SELECT count() FROM ${DB}.${ATTEMPTS} FINAL`, ""));
+    },
+    async countRequests(): Promise<number> {
+      return Number(await post(`SELECT count() FROM ${DB}.${REQUESTS} FINAL`, ""));
     },
   };
 }

@@ -294,12 +294,43 @@ export function webhookAttemptSubject(environmentId: string): string {
     WEBHOOK_ATTEMPT_ACTION.domain,
     WEBHOOK_ATTEMPT_ACTION.action,
     environmentId,
   );
 }
 
+/** The API request log's action (FR-ANL-07, chapter 4.4). */
+export const API_REQUEST_ACTION = { domain: "api", action: "request" };
+
+export function apiRequestSubject(environmentId: string): string {
+  return analyticsSubjectFor(
+    API_REQUEST_ACTION.domain,
+    API_REQUEST_ACTION.action,
+    environmentId,
+  );
+}
+
+/** The token for a request that resolved to no tenant.
+ *
+ * A SEPARATE FUNCTION, NOT A RELAXED ARGUMENT TO `analyticsSubjectFor`. That validator
+ * refuses a non-UUID because an environment id becomes a dot-delimited subject token, and
+ * the refusal is what keeps one tenant's records out of another tenant's filter. A validator
+ * with an escape hatch is a validator with a hole, and the hole is measurable: publishing a
+ * token of `no.tenant` produces a FIVE-token subject that the four-token wildcard does not
+ * match, and `*` publishes a subject containing a literal asterisk. Neither fails at publish
+ * time. A malformed token does not reach the wrong tenant -- it goes where no intended filter
+ * reaches, which is the quiet direction.
+ *
+ * `_none` cannot be a UUID, so no exact per-tenant filter matches it. Measured over a stream
+ * holding six candidate tokens plus one real tenant's record: tenant A's exact filter saw 1,
+ * tenant B's saw 0. */
+export const NO_TENANT_TOKEN = "_none";
+
+export function apiRequestSubjectWithoutTenant(): string {
+  return [ANALYTICS_SUBJECT_PREFIX, API_REQUEST_ACTION.domain, API_REQUEST_ACTION.action, NO_TENANT_TOKEN].join(".");
+}
+
 // ---------------------------------------------------------------------------
 // The dispatch contract (constitution IV).
 //
 // The dispatcher owns no database. "Only the API service writes to PostgreSQL…
 // Other services obtain writes and backfill reads via the API service's internal
 // endpoints." These are those endpoints, and they live here for the reason

@@ -83,12 +83,32 @@ import {
 
 export interface Environment {
   id: string;
   kind: "development" | "production";
 }
 
+/** Set a per-environment rate-limit override (FR-RTL-04).
+ *
+ * The columns are nullable and null means "no override" rather than zero -- refuse-everything
+ * has to stay expressible. Written here rather than at a call site because the query engine
+ * lives in this layer only (constitution I, ADR-16), and a test reaching for `sql` directly
+ * is the lint rule firing rather than a shortcut.
+ */
+export async function setEnvironmentLimits(
+  db: Db,
+  {
+    environmentId,
+    restPerMinute,
+  }: { environmentId: string; restPerMinute: number | null },
+): Promise<void> {
+  await db
+    .update(environments)
+    .set({ restLimitPerMinute: restPerMinute })
+    .where(eq(environments.id, environmentId));
+}
+
 export async function createEnvironment(
   db: Db,
   { name, kind = "development" }: { name: string; kind?: Environment["kind"] },
 ): Promise<Environment> {
   const organisationId = randomUUID();
   const applicationId = randomUUID();

@@ -9,12 +9,13 @@ import type { RequestWithPrincipal } from "../auth/principal";
 import { LOGGER } from "../logger";
 import { remaining, resetAt, windowStart } from "./bucket";
 import { clientAddress } from "./client-address";
 import { COUNTER_STORE, LIMITS_DB } from "./limits.module";
 import { authFailureThreshold, WINDOW_MS, type LimitedOperation } from "./policy";
 import { authKey, counterKey, type CounterStore } from "./store";
+import { LIMITED_OPERATION, REFUSED_AT } from "../request-log/event";
 
 /** Read once per call site so a test that freezes time sees one instant. */
 const now0 = (): number => Date.now();
 
 // The tenant limiter (FR-RTL-01…04).
 //
@@ -116,12 +117,16 @@ export class RateLimitMiddleware implements NestMiddleware {
       const address = clientAddress(req);
       const count = await this.store.increment(
         authKey(address, windowStart(now0(), WINDOW_MS)) + ":signup",
         now0(),
       );
       if (count !== null && count > authFailureThreshold()) {
+        // Its own family. `signup` is not a `DEFAULT_LIMITS` key -- that union is rest, send
+        // and connect, and `connect` is the gateway's, counted elsewhere entirely.
+        (req as unknown as Record<symbol, unknown>)[REFUSED_AT] = "middleware";
+        (req as unknown as Record<symbol, unknown>)[LIMITED_OPERATION] = "signup";
         res.statusCode = 429;
         res.setHeader("Retry-After", "60");
         res.setHeader("content-type", "application/json");
         res.end(
           JSON.stringify({
             code: "rate_limited",
@@ -207,12 +212,22 @@ export class RateLimitMiddleware implements NestMiddleware {
     } else {
       res.setHeader("X-RateLimit-Remaining", String(nearest.remaining));
       res.setHeader("X-RateLimit-Reset", String(nearest.resetSeconds));
     }
 
     if (refusal !== undefined) {
+      // STAMPED FOR THE REQUEST LOG, and it is `refusal.operation` -- the ONE operation that
+      // tripped -- not `operationsFor`'s array, which says what was COUNTED. The line below
+      // already narrows it to name the failure to the customer.
+      //
+      // AND IT IS NOT A ROUTE TEMPLATE. This limiter's whole route knowledge is three-valued:
+      // outside /v1/, the send path, everything else. So "which endpoint is being
+      // rate-limited" asks about a granularity that does not exist, and the operation class
+      // is the finest true answer about one of these refusals.
+      (req as unknown as Record<symbol, unknown>)[REFUSED_AT] = "middleware";
+      (req as unknown as Record<symbol, unknown>)[LIMITED_OPERATION] = refusal.operation;
       const retryAfter = Math.max(1, refusal.resetSeconds - Math.floor(now / 1000));
       res.setHeader("Retry-After", String(retryAfter));
       res.setHeader("X-RateLimit-Remaining", "0");
       // The message names WHICH limit was reached: "too many requests" and "too
       // many messages" are different problems, one saying batch and the other
       // saying slow down. Neither names a credential (NFR-SEC-06).