The gateway's first stream

@@ -325,12 +325,45 @@ export function apiRequestSubject(environmentId: string): string {
 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(".");
 }
 
+/** Connection open and close (FR-ANL-01, chapter 4.5) -- the last arm of that clause
+ *  without a producer.
+ *
+ *  TWO ACTIONS ON ONE DOMAIN, NOT ONE ACTION WITH THE EVENT IN THE PAYLOAD. A consumer
+ *  that wants only closes can filter `analytics.connection.closed.>` on the subject
+ *  rather than shaping every open to find out it did not want it, which is what a
+ *  subject grammar is for.
+ *
+ *  AND THERE IS NO `_none` ARM HERE. 4.4 needed one because a request can be made by
+ *  nobody. A connection event is emitted from `registry.add` and from `meter.closed`,
+ *  and the only function reaching either takes a non-optional `Identity` -- its call
+ *  site passes `result.identity` after the 429, 4001, 1011, 4003 and 4008 refusals have
+ *  each returned. An unauthenticated socket exists; an unauthenticated connection does
+ *  not, so there is no record for a tenantless arm to carry. */
+export const CONNECTION_OPENED_ACTION = { domain: "connection", action: "opened" };
+export const CONNECTION_CLOSED_ACTION = { domain: "connection", action: "closed" };
+
+export function connectionOpenedSubject(environmentId: string): string {
+  return analyticsSubjectFor(
+    CONNECTION_OPENED_ACTION.domain,
+    CONNECTION_OPENED_ACTION.action,
+    environmentId,
+  );
+}
+
+export function connectionClosedSubject(environmentId: string): string {
+  return analyticsSubjectFor(
+    CONNECTION_CLOSED_ACTION.domain,
+    CONNECTION_CLOSED_ACTION.action,
+    environmentId,
+  );
+}
+
 // ---------------------------------------------------------------------------
 // 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

@@ -13,12 +13,13 @@
   },
   "dependencies": {
     "@relay/protocol": "workspace:*",
     "@relay/service-kit": "workspace:*",
     "ioredis": "^6.0.0",
     "jose": "^6.2.7",
+    "nats": "^2.29.3",
     "ws": "^8.21.1"
   },
   "devDependencies": {
     "@relay/api": "workspace:*",
     "@types/ws": "^8.18.1",
     "tsx": "^4.23.1"

@@ -166,12 +166,21 @@ services:
     build:
       context: .
       dockerfile: services/gateway/Dockerfile
     environment:
       RELAY_API_URL: http://api:4000
       RELAY_REDIS_URL: redis://redis:6379
+      # CONTAINER NAME, NOT localhost -- and this line is what makes chapter 4.5's
+      # producer reach anything. `createConnectionPublisher` defaults to
+      # `nats://localhost:4222`, which inside this container is the gateway's own
+      # loopback rather than the broker; and a connection-log publish failure is
+      # SWALLOWED by design (constitution III), so without this the records go nowhere
+      # and nothing says so. The api's entry carries the same note about Redis for the
+      # same reason: a missing address is harder to see than a wrong one, because there
+      # is no sentence to disagree with.
+      RELAY_NATS_URL: nats://nats:4222
       PORT: "4001"
       # The gateway's first credential of its own. Every other
       # call it makes forwards the END USER's token; a usage report is nobody's
       # user action, so it speaks for itself. Its own variable rather than the
       # dispatcher's, because `PlatformPrincipal.service` has to stay true and
       # because this service faces the public internet and that one does not.

// Handing over totals rather than reporting them. This handler is already
// documented as the last place that should throw, and a mass disconnect would
// turn one event into a burst of HTTP requests.
meter.closed(connection, new Date());

// FR-ANL-01's last arm: an analytical event when a connection opens and when it closes.
//
// THE FIRE-AND-FORGET ARGUMENT IS 3.20's AND THE SUBJECT GRAMMAR IS 4.3's. Neither is
// re-derived here. What is new is that this producer BUFFERS, and the three paragraphs
// below are the consequences of that one decision.
import {
  connectionClosedSubject,
  connectionOpenedSubject,
} from "@relay/protocol";
import type { Logger } from "@relay/service-kit";
 
import type { Connection } from "../registry.js";
 
/** Five seconds, and the number is a budget rather than a preference.
 *
 * FR-ANL-04 allows 60 seconds from the originating operation to the event being
 * queryable, "under normal conditions" -- and the ingester's own batch bound spends up
 * to 2 of them (`docs/05-sad.md` §4; DR-11 itself names no interval and no row count).
 *
 * COPYING `METER_INTERVAL_MS` WOULD BREACH IT. The meter ticks every 60,000 ms and the
 * whole posture of this module is *be like the meter*, so the obvious number is the
 * wrong one: the meter feeds a monthly quota with no latency clause over it, and this
 * feeds a store that has one.
 *
 * And the two pressures are less opposed than they look. The tick's win is removing the
 * serial round trip from whatever accumulated, not waiting longer -- at 0.0034 ms a
 * record the publish is effectively free at any interval, and the 2.3-second burst a
 * deploy would cost is a property of publishing PER CLOSE, not of a short tick. So a
 * short interval costs almost nothing and buys the whole budget. */
export const FLUSH_INTERVAL_MS = 5_000;
 
/** How many unpublished records to hold before dropping.
 *
 * `meter.ts` faced this and its cap is `MAX_RETAINED_CLOSED = 4_000`, bounded by closes
 * since the last ACCEPTED report. THIS BOUND MEANS THE SAME THING, and only because of
 * the retention rule below: a buffer that emptied on every flush regardless of outcome
 * could never reach a cap at all, because an unreachable broker would drain it every
 * five seconds into failures.
 *
 * AND IT IS TWO RECORDS PER CONNECTION, NOT ONE. The meter retains a closed entry per
 * (connection, period); this retains an open AND a close. At NFR-SCL-01's 10,000
 * sockets a full mass disconnect with the broker away overflows this, which is the
 * case the drop counter exists to make visible rather than the case it prevents. */
export const MAX_BUFFERED = 4_000;
 
/** The wire shape. Snake case because it leaves the platform. */
export interface ConnectionEvent {
  type: "connection.opened" | "connection.closed";
  connection_id: string;
  environment_id: string;
  user_external_id: string;
  ts: string;
  close_code?: number;
  duration_ms?: number;
}
 
/** Which event, and what only a close carries.
 *
 * A DISCRIMINATED UNION RATHER THAN TWO OPTIONAL ARGUMENTS. An open has no close code
 * and no duration, and this makes supplying one a type error rather than a field the
 * shaper has to remember to omit -- a design in which a case cannot arise beats a
 * branch that handles it. */
export type ConnectionEventKind =
  | { kind: "opened" }
  | { kind: "closed"; at: Date; code: number };
 
/** IDENTIFIERS, INSTANTS AND A CODE ONLY.
 *
 * Built by NAMING every field rather than by spreading the connection, and that is the
 * point: `Connection` carries `identity.token` -- the bearer token the client presented
 * at connect -- plus `buffer` (frames), `channelIds` and the socket itself. A spread
 * would put any of them on a stream with seven-day retention. An allow-list fails
 * closed when somebody adds a field; a spread fails open.
 *
 * The authority is constitution III's allow-list: the analytical store holds "only
 * lengths, identifiers, and metadata", and a credential is none of the three. FR-ANL-11
 * governs message text and NFR-SEC-06 governs application LOGS; neither reaches a
 * record on a stream, which is why the rule is stated as an allow-list here.
 *
 * ABSENT, NOT EMPTY. `exactOptionalPropertyTypes` is on, so the close-only fields are
 * spread in rather than assigned -- an explicit `undefined` is not an absent key, and
 * 4.4 measured that a `LowCardinality(String)` column cannot tell the two apart. */
export function toConnectionEvent(
  connection: Connection,
  event: ConnectionEventKind,
): ConnectionEvent {
  const base = {
    connection_id: connection.id,
    environment_id: connection.environmentId,
    user_external_id: connection.identity.userExternalId,
  };
  if (event.kind === "opened") {
    return {
      type: "connection.opened",
      ...base,
      // FR-005a. THE CONNECTION'S OWN INSTANT, NOT THIS MOMENT. `openedAt` is stamped
      // before the resume and before the ack "because the socket is already open and
      // already costing a minute", and it is the field the meter reads. Two instants
      // for one open would make the reconciliation disagree for a reason that is
      // neither of the two the chapter explains.
      ts: connection.openedAt.toISOString(),
    };
  }
  return {
    type: "connection.closed",
    ...base,
    ts: event.at.toISOString(),
    close_code: event.code,
    // Close minus open, in milliseconds. The interval's two endpoints are stated rather
    // than implied: `openedAt` is the handshake, `at` is the socket's close.
    //
    // NOT THE METER'S QUANTITY. The meter charges every calendar minute a connection
    // was open for any part of, so 00:00:59 to 00:01:01 is TWO connection-minutes and
    // 2,000 ms. Different quantities sharing a name; the reconciliation compares
    // buckets against buckets.
    duration_ms: Math.max(0, event.at.getTime() - connection.openedAt.getTime()),
  };
}
 
/** One record, addressed and keyed. The same three fields the api's outbox port uses,
 *  for the same reason: which broker is a configuration detail. */
export interface PublishableRecord {
  subject: string;
  /** `{connection_id}:{event}`, NOT the connection id alone. One connection produces
   *  two records, and 3.20 learned exactly this with `{delivery}:{attempt}`: a delivery
   *  id alone collapsed seven retries into one message. */
  id: string;
  payload: ConnectionEvent;
}
 
export function recordFor(event: ConnectionEvent): PublishableRecord {
  const subject =
    event.type === "connection.opened"
      ? connectionOpenedSubject(event.environment_id)
      : connectionClosedSubject(event.environment_id);
  return {
    subject,
    id: `${event.connection_id}:${event.type === "connection.opened" ? "opened" : "closed"}`,
    payload: event,
  };
}
 
/** What the buffer needs of a publisher: ONE MESSAGE PER RECORD.
 *
 * Not 500 records in one message. One message carries one subject and the subject
 * carries the tenant; one message carries one `Nats-Msg-Id` and the deduplication id is
 * per record; and the ingester's `route()` takes one record per message -- handed an
 * array it finds no `type`, shapes to null, and TERMINATES it. */
export interface ConnectionPublisher {
  publish(record: PublishableRecord): Promise<void>;
  close(): Promise<void>;
}
 
export interface ConnectionLog {
  /** Beside `registry.add`. */
  opened(connection: Connection): void;
  /** Beside `meter.closed`. Must not throw and must not be awaited. */
  closed(connection: Connection, at: Date, code: number): void;
  /** Exposed for the timer, for the shutdown flush, and for tests that drive their own
   *  clock -- the same three reasons `meter.reportOnce` is. */
  flushOnce(): Promise<void>;
  /** How many records are waiting for an accepted publish. */
  buffered(): number;
  /** How many were discarded at the cap. Counted rather than silent. */
  dropped(): number;
  stop(): void;
  /** Stop the tick, send what is left, and close the client -- in that order.
   *
   * ONE FABRIC WITH ONE CLOSE, because `main.test.ts` reads `main.ts`'s source and
   * asserts that every module it builds is both passed into `attachSessions` and
   * awaited in `shutdown()`. A publisher held as a second const would satisfy neither
   * half, and the honest fix is for this module to own its publisher's lifetime rather
   * than for the guard to gain an exemption. The flush is the work `sessions.close()`
   * is ordered first to make possible. */
  close(): Promise<void>;
}
 
export interface ConnectionLogOptions {
  publisher: ConnectionPublisher;
  logger: Logger;
  intervalMs?: number;
  maxBuffered?: number;
}
 
export function createConnectionLog({
  publisher,
  logger,
  intervalMs = FLUSH_INTERVAL_MS,
  maxBuffered = MAX_BUFFERED,
}: ConnectionLogOptions): ConnectionLog {
  let pending: PublishableRecord[] = [];
  let discarded = 0;
 
  /** A RECORD WHOSE PUBLISH FAILED GOES BACK, AND THAT IS THE METER'S RULE FOR ITS ONE
   *  EXCEPTION APPLIED TO ALL OF THIS.
   *
   *  `meter.ts`: "a report that cannot be delivered is DROPPED rather than queued ...
   *  WITH ONE EXCEPTION ... a connection that has CLOSED has no next report to repair a
   *  lost one, so its final total is retained until a report carrying it is accepted."
   *
   *  Every connection event is in that exception. An open is sent once and a close is
   *  sent once; there is no later record carrying the same fact again. So the meter's
   *  special case is this producer's general rule. */
  function enqueue(record: PublishableRecord): void {
    if (pending.length >= maxBuffered) {
      // OLDEST FIRST, and the argument does NOT transfer from the meter unchanged. The
      // meter drops the oldest because under-counting is "the same direction as every
      // other loss in this design" -- it is a bill, and billing for a socket nobody
      // holds is the worse error. A connection log is not a bill: both directions lose
      // a fact and neither is safe.
      //
      // Oldest anyway, for a different reason. Dropping the NEWEST discards the records
      // describing the outage itself -- the opens and closes happening while the broker
      // is away -- which is the window an operator opens this log to see. The oldest
      // records are the ones most likely to have a neighbour that survived.
      pending.shift();
      discarded += 1;
      logger.log("error", "connection_log.buffer_overflow", {
        discarded,
        buffered: pending.length,
      });
    }
    pending.push(record);
  }
 
  function opened(connection: Connection): void {
    enqueue(recordFor(toConnectionEvent(connection, { kind: "opened" })));
  }
 
  function closed(connection: Connection, at: Date, code: number): void {
    enqueue(recordFor(toConnectionEvent(connection, { kind: "closed", at, code })));
  }
 
  async function flushOnce(): Promise<void> {
    if (pending.length === 0) return;
    const batch = pending;
    pending = [];
 
    // PIPELINED, ONE MESSAGE PER RECORD. Every publish goes out without awaiting the
    // one before it and the acks are collected together: 0.229 ms each when awaited
    // serially against 0.0034 ms pipelined, which is 2.3 seconds for a deploy that
    // closes 10,000 sockets against 34 ms.
    const outcomes = await Promise.allSettled(
      batch.map((record) => publisher.publish(record)),
    );
 
    // PER RECORD, because one message per record means this flush has as many answers
    // as it had records. The meter never faces this: it sends one report and gets one
    // answer. A flush that treated a partial failure as total loss would discard
    // records the broker accepted; as total success, records it did not.
    // THE REJECTIONS THEMSELVES, NOT A PARALLEL LIST OF INDICES. An earlier version kept
    // `failed` (the records) and looked the first reason up separately with
    // `outcomes.find(...)`, which needed a `?? "unknown"` fallback for a case that cannot
    // occur: inside `failed.length > 0` there is always a rejection. Coverage reported the
    // file at 93.75% branches with that one arm uncovered, and the arm was not missing a
    // test -- it was unreachable. **A design in which a case cannot arise beats a branch
    // that handles it, because the branch is the thing that rots.** Pairing each record
    // with its own outcome deletes the branch instead of testing it.
    const rejected: Array<{ record: PublishableRecord; reason: unknown }> = [];
    outcomes.forEach((outcome, i) => {
      const record = batch[i];
      if (outcome.status === "rejected" && record !== undefined) {
        rejected.push({ record, reason: outcome.reason });
      }
    });
    for (const { record } of rejected) enqueue(record);
 
    if (rejected[0] !== undefined) {
      // ONE LINE PER FLUSH, not per record, and it carries counts rather than payloads
      // -- the shape `meter.report_failed` uses. No subject, no id, no record.
      logger.log("error", "connection_log.publish_failed", {
        attempted: batch.length,
        failed: rejected.length,
        buffered: pending.length,
        error: String(rejected[0].reason),
      });
    }
  }
 
  // STOPPED, NOT unref'd -- the meter's pattern (`meter.ts:218`), cleared by `stop()`
  // from shutdown. An unref'd timer lets the process exit with records still buffered.
  const timer = setInterval(() => {
    void flushOnce();
  }, intervalMs);
 
  return {
    opened,
    closed,
    flushOnce,
    buffered: () => pending.length,
    dropped: () => discarded,
    stop: () => clearInterval(timer),
    async close(): Promise<void> {
      clearInterval(timer);
      // The last flush before the client goes. A record still buffered here is one a
      // killed process would have lost, which is the difference between a clean stop
      // and a kill that the chapter publishes as a number.
      await flushOnce();
      await publisher.close();
    },
  };
}

// The gateway's first broker client, and the count is the subject rather than a detail:
// ADR-07 refuses core NATS for fan-out on an argument about how many client libraries
// this service holds. It held five dependencies and none of them a broker; it holds six
// now. The fan-out decision does not change -- ADR-10 keeps presence in Redis, so Redis
// is mandatory here regardless -- but the sentence that PRICED it stops being true.
import { connect, type NatsConnection } from "nats";
 
import type { Logger } from "@relay/service-kit";
 
import type { ConnectionPublisher, PublishableRecord } from "./event.js";
 
export const DEFAULT_NATS_URL = "nats://localhost:4222";
 
/** ONE CLIENT, CREATED ONCE, SHARED, CONNECTED LAZILY.
 *
 * LAZY BECAUSE A BROKER THAT IS DOWN AT BOOT MUST NOT COST A SOCKET. The gateway's job
 * is terminating WebSockets; an analytical publisher that refused to start would make a
 * dashboard's dependency into the product's, which is constitution III inverted. So the
 * connection is attempted on the first flush and retried on the next one.
 *
 * AND IT DOES NOT CREATE THE STREAM. `ensureAnalyticsStream` belongs to the api. A
 * gateway that ensured it would also ask for `replicas > 1` under `NODE_ENV=production`
 * and be refused in non-clustered mode (049-2) -- so there is nothing here to retry and
 * nothing to get wrong. A publish to a stream that does not exist comes back 503, which
 * the buffer treats as any other failure: the record stays and goes again. */
export function createConnectionPublisher({
  url = process.env.RELAY_NATS_URL ?? DEFAULT_NATS_URL,
  logger,
}: {
  url?: string;
  logger: Logger;
}): ConnectionPublisher {
  let nc: NatsConnection | null = null;
  let connecting: Promise<NatsConnection> | null = null;
  let closed = false;
 
  /** ONE IN-FLIGHT ATTEMPT, SHARED. A flush publishes up to `MAX_BUFFERED` records at
   *  once and every one of them calls this; without the shared promise a cold start
   *  would open four thousand connections to a broker that is already refusing one. */
  async function client(): Promise<NatsConnection> {
    if (closed) throw new Error("connection log publisher is closed");
    if (nc !== null) return nc;
    connecting ??= connect({ servers: url, name: "gateway-connection-log" })
      .then((c) => {
        nc = c;
        connecting = null;
        return c;
      })
      .catch((error: unknown) => {
        connecting = null;
        throw error;
      });
    return connecting;
  }
 
  return {
    /** One JetStream message per record. The deduplication id goes on the message, so
     *  a redelivered publish of the same record collapses inside the broker's window
     *  and a `ReplacingMergeTree` keyed on the record's own key catches the rest. */
    async publish(record: PublishableRecord): Promise<void> {
      const c = await client();
      await c
        .jetstream()
        .publish(record.subject, JSON.stringify(record.payload), {
          msgID: record.id,
        });
    },
    async close(): Promise<void> {
      closed = true;
      const c = nc;
      nc = null;
      if (c === null) {
        // Nothing was ever opened, which is the ordinary case for a gateway that ran
        // with the broker away. Saying so once beats a silent no-op.
        logger.log("info", "connection_log.publisher_never_connected", {});
        return;
      }
      // DRAIN RATHER THAN CLOSE. `drain()` waits for published messages to be flushed
      // to the server; `close()` does not, and the last flush before a deploy is
      // exactly the one worth keeping.
      await c.drain();
    },
  };
}

@@ -18,12 +18,13 @@ import {
 } from "@relay/protocol";
 import { newRequestId, type Logger } from "@relay/service-kit";
 import { WebSocketServer, type WebSocket } from "ws";
 
 import { ApiError, type ApiClient } from "./api-client.js";
 import { authenticate, type Identity } from "./auth.js";
+import type { ConnectionLog } from "./connection-log/event.js";
 import {
   DEFAULT_HEARTBEAT_MS,
   MAX_CONNECTIONS_PER_USER,
   type Connections,
 } from "./connections.js";
 import type { Fanout } from "./fanout.js";
@@ -267,12 +268,20 @@ export interface SessionServerOptions {
   /** Optional for the reason `limits` and `fanout` are: 2.5's
    * tests and a single-process dev run have no api credential, and a socket
    * server that refused to start without one would be a worse default than an
    * unmetered one. `main.ts` always supplies the interval; the meter itself is
    * built here so its timer has the same owner as the heartbeat's. */
   meterIntervalMs?: number;
+  /** FR-ANL-01's producer. Optional for the reason `limits`, `fanout` and `connections`
+   * are, and the reason matters more here than usual: for an analytical log "optional"
+   * means UNRECORDED, never unsafe. A gateway built without one serves sockets exactly
+   * as before, which is constitution III's independence expressed as a type.
+   *
+   * INJECTED ALREADY BUILT, like every other fabric, so its close has an owner in
+   * `main.ts` and the tests that call this function directly stay broker-free. */
+  connectionLog?: ConnectionLog;
 }
 
 // THE FOUR PRESENCE TIMINGS ARE NOT HERE, and an earlier draft of this chapter put
 // them here. `fanout` and `presence` are INJECTED already built, and an injected thing
 // carries its own configuration: a test that wants a hundred-millisecond grace period
 // constructs `createPresence({ graceMs: 100, … })` and injects that, the way the
@@ -295,12 +304,13 @@ export function attachSessions({
   typing,
   renewalIntervalMs = DEFAULT_RENEWAL_INTERVAL_MS,
   connections,
   heartbeatMs = DEFAULT_HEARTBEAT_MS,
   limits,
   meterIntervalMs = METER_INTERVAL_MS,
+  connectionLog,
 }: SessionServerOptions): {
   registry: Registry;
   meter: Meter;
   /** ASYNC AS OF THE CONNECTION CAP, and `releaseAll` below was the reason. Freeing
    * the places this instance holds is a round trip to Redis that has to COMPLETE
    * before `wss.close()`, or the deploy case the method exists for is a race it can
@@ -954,12 +964,23 @@ export function attachSessions({
       // reconnect storm a free window on every attempt.
       openedAt: new Date(),
       environmentId: identity.environmentId,
     };
 
     registry.add(connection);
+    // THE OPEN RECORD, BESIDE `registry.add` AND NOT INSIDE THE METER -- because there
+    // is no `meter.opened` to sit beside. The `Meter` interface is `closed`,
+    // `reportOnce`, `retained`, `dropped` and `stop`: it learns about OPEN connections
+    // by walking this registry and only needs telling when one leaves. So the two
+    // records take two different anchors, and this is the first.
+    //
+    // NOT AWAITED, AND IT CANNOT THROW. `opened` enqueues and returns; the tick does
+    // the sending. A publish from here would put an analytical fabric on the handshake
+    // path, which constitution III forbids and which R3 priced at 2.3 seconds for a
+    // deploy that closes ten thousand sockets.
+    connectionLog?.opened(connection);
     // Subscriptions follow membership: the first local member of a channel
     // makes this instance a subscriber, and the last one to leave releases
     // it (reference-counted in the fabric).
     const subscribing = Promise.all(
       [...connection.channelIds].flatMap((channelId) => [
         fanout?.subscribe(channelId),
@@ -1140,13 +1161,26 @@ export function attachSessions({
       // That is not a rounding error: it is the one thing the wall-clock-minute unit
       // was chosen to charge (research R19).
       //
       // Handing over totals rather than reporting them. This handler is already
       // documented as the last place that should throw, and a mass disconnect would
       // turn one event into a burst of HTTP requests.
-      meter.closed(connection, new Date());
+      const closedAt = new Date();
+      meter.closed(connection, closedAt);
+      // THE CLOSE RECORD, BESIDE `meter.closed` AND BEFORE `registry.remove` -- the
+      // same ordering constraint, for the same reason: the line below removes this
+      // connection from the registry, and a hand-over that read it afterwards would
+      // read nothing.
+      //
+      // ONE INSTANT FOR BOTH, not two calls to `new Date()`. The meter's minutes and
+      // this record's duration would otherwise disagree by however long the line
+      // between them took, which is a third cause in a comparison built to have two.
+      //
+      // AND THIS HANDLER IS DOCUMENTED AS THE LAST PLACE THAT SHOULD THROW. `closed`
+      // enqueues and returns; nothing it calls awaits a broker.
+      connectionLog?.closed(connection, closedAt, code);
       registry.remove(connection.id);
       // THIS HANDLER NOW CARRIES TWO ORDERING CONSTRAINTS, not none. Presence is told
       // AFTER `registry.remove`, because it asks whether this was the user's last
       // connection on this instance and must not count the one that is leaving. The
       // unsubscribes come last.
       //

@@ -6,12 +6,14 @@ import { createFanout } from "./fanout.js";
 import { createMembership } from "./membership.js";
 import { createPresence } from "./presence.js";
 import { createConnections } from "./connections.js";
 import { createTyping } from "./typing.js";
 import { createGatewayLimits } from "./limits.js";
 import { attachSessions } from "./session.js";
+import { createConnectionLog } from "./connection-log/event.js";
+import { createConnectionPublisher } from "./connection-log/publisher.js";
 
 // The gateway — SAD §4.1: terminates WebSockets and never writes to the
 // database (ADR-05). Chapter 1.4 stood up the HTTP half (health, request
 // ids, structured logs); chapter 2.5 gives it the job it exists for, and
 // 2.6 makes that job survive a second instance. The
 // health payload still advertises the wire vocabulary, computed from
@@ -79,12 +81,25 @@ export function createServer(logger?: Logger) {
   const serviceCredential = process.env.RELAY_INTERNAL_CREDENTIAL_GATEWAY;
   if (serviceCredential === undefined) {
     log.log("info", "metering.disabled", {
       reason: "RELAY_INTERNAL_CREDENTIAL_GATEWAY is not set",
     });
   }
+  // THE SIXTH DEPENDENCY, AND THE FIRST BROKER CLIENT THIS SERVICE HAS EVER HELD.
+  // ADR-07 refuses core NATS for fan-out on an argument about how many client libraries
+  // the gateway holds; this makes it six. The fan-out decision is untouched -- ADR-10
+  // keeps presence in Redis, so Redis is mandatory here regardless -- but the sentence
+  // that PRICED that refusal ("two broker clients where it had one, and remove none")
+  // stops being true, and `docs/06-adr-deep-dives.md` says so now.
+  //
+  // LAZY, so a broker that is down at boot costs no socket, and built here rather than
+  // inside `attachSessions` for the reason the other seven are: its close has an owner.
+  const connectionLog = createConnectionLog({
+    publisher: createConnectionPublisher({ logger: log }),
+    logger: log,
+  });
   const sessions = attachSessions({
     server,
     api: createApiClient(
       process.env.RELAY_API_URL ?? DEFAULT_API_URL,
       serviceCredential,
     ),
@@ -98,12 +113,13 @@ export function createServer(logger?: Logger) {
     // is admitted and no number moves — `**/main.ts` is excluded from the coverage
     // ratchet, so no figure could show it. Registering `close()` below is the other
     // half and neither substitutes for the other: without this line the cap is inert,
     // without that one every gateway leaks a Redis client.
     connections,
     limits,
+    connectionLog,
     // Overridable so `meter.itest.ts` can drive a spawned gateway without
     // waiting a real minute per assertion. The two tests there are the ones an
     // in-process gateway cannot run — a signal has to arrive at a process — and
     // sixty seconds each would put them past the suite's timeout.
     //
     // Spread rather than assigned `undefined`: `exactOptionalPropertyTypes` is
@@ -132,12 +148,23 @@ export function createServer(logger?: Logger) {
    *
    * `sessions` FIRST, because its close is the one with work to finish: a final usage
    * report and the release of the places this instance holds. The fabrics it reports
    * and publishes through have to still be open while it does that. */
   async function shutdown(): Promise<void> {
     await sessions.close();
+    // EIGHT NOW, AND THE ORDER IS THE ARGUMENT. `sessions` is first "because its close
+    // is the one with work to finish ... the fabrics it reports and publishes through
+    // have to still be open while it does that", and flushing this buffer is exactly
+    // that kind of work: the tick is stopped and the last records are sent here, not
+    // dropped.
+    //
+    // NOTHING WILL CATCH A MISSED CLOSE. `main.ts` is excluded from the coverage
+    // ratchet, so no figure could show a shutdown that closed seven of eight -- which
+    // is why the COUNT is recorded in `baseline.txt` before and after rather than
+    // trusted to a test.
+    await connectionLog.close();
     await fanout.close();
     await presence.close();
     await membership.close();
     await typing.close();
     await connections.close();
     await limits.close();

-- FR-ANL-01's LAST ARM WITHOUT A PRODUCER: connection open and close.
--
-- A FOURTH TABLE, for the reason there is a third. `webhook_attempts` keeps 90 days and
-- `api_requests` keeps 30 because FR-ANL-07 says so; one TTL clause cannot express two
-- retentions, and one table cannot hold three column sets that overlap only on
-- `environment_id` and `ts`.
--
-- `environment_id` IS NOT NULLABLE HERE, UNLIKE `api_requests`, and the difference between
-- the two chapters is worth stating rather than inheriting. A request can be made by
-- nobody: every 404, every 401, /healthz, and every call the dispatcher and gateway make on
-- the internal seam, whose `platform` principal carries no environment by design. A
-- CONNECTION cannot. `open()` is the only function that builds one and the only caller of
-- `registry.add`; it takes a non-optional `Identity`, and its single call site is reached
-- only after the 429 upgrade refusal, 4001, 1011, 4003, 4008 and the 4004 connection cap
-- have each returned. An unauthenticated SOCKET exists; an unauthenticated CONNECTION does
-- not. So there is no tenantless record to carry, no `_none` arm, and no
-- `allow_nullable_key` setting either -- every column in the sorting key is non-nullable.
CREATE TABLE IF NOT EXISTS relay_analytics.connection_events (
    environment_id UUID,
    ts             DateTime64(3, 'UTC'),
    connection_id  UUID,
    -- `opened` | `closed`.
    event          LowCardinality(String),
    -- ONLY ON A CLOSE, AND A NUMBER RATHER THAN A STRING. A close code is a small integer,
    -- and a String column answers `WHERE close_code = 1000` with nothing and no error.
    -- Measured: ClickHouse coerces a JSON number into a String column AND a JSON string
    -- into a UInt16, so a type disagreement between producer and table lands silently in
    -- whichever spelling the producer happened to send -- there is no loud direction.
    --
    -- AND IT IS NOT DRAWN FROM `CLOSE_CODES`. That registry holds 4001, 4002, 4003, 4004,
    -- 4008 and 4009 -- the platform's own 4xxx range. A clean close is 1000 and an abnormal
    -- one is 1006, and neither is in it, so `check:errors` does not guard this column's
    -- vocabulary. What survives is the narrower claim: an integer the protocol defines,
    -- not a sentence somebody writes.
    close_code     Nullable(UInt16),
    -- UInt64, NOT UInt32. Nothing caps a socket's lifetime -- the only lifetime control is
    -- `MAX_MISSED_PINGS` on a 30-second ping, which ends a socket that has stopped
    -- answering rather than one that has not -- and UInt32 milliseconds wraps at 49.7 days.
    -- The cost is four bytes on a column that is null on every open record.
    duration_ms    Nullable(UInt64),
    -- NOT nullable. It comes from the same `Identity` as `environment_id`, and both fields
    -- are declared `string` rather than `string | undefined`, so a connection event has
    -- both or neither. (That `Identity` has a THIRD field, `token`, is why the producer
    -- names every field rather than spreading: constitution III's allow-list -- "only
    -- lengths, identifiers, and metadata" -- refuses a credential by construction.)
    user_external_id String,
    -- 048 measured that an unmatched column takes its DEFAULT with no error, that a
    -- DateTime64 default is the epoch, and that the epoch is older than any TTL -- so the
    -- row is deleted at insert while the insert returns OK and the stream drains to zero.
    -- `input_format_skip_unknown_fields=0` catches a RENAMED field and cannot catch an
    -- absent one. This can.
    CONSTRAINT ts_is_real CHECK ts > toDateTime64('2020-01-01 00:00:00', 3, 'UTC')
)
ENGINE = ReplacingMergeTree
PARTITION BY toYYYYMM(ts)                                  -- DR-07
-- `event` LAST, AND IT IS THE ONE DECISION IN THIS FILE THAT CANNOT BE GUESSED. One
-- connection produces TWO rows under one `connection_id`; without `event` in the key a
-- ReplacingMergeTree collapses the open into the close and a reader sees half the story.
-- Verified against the server: `count() FINAL` is 2, not 1. That is 4.3's lesson applied
-- rather than re-learned -- idempotence is the record's own key, and this record's own key
-- includes which event it is.
ORDER BY (environment_id, ts, connection_id, event)
-- `toDateTime(ts)`, not `ts` -- 047 measured `TTL ts + INTERVAL` on a DateTime64 refused
-- with BAD_TTL_EXPRESSION, after the SAD had published the broken form since its first
-- draft.
--
-- AND 90 IS A DEFAULT RATHER THAN A DERIVATION. FR-ANL-07 fixes 30 for the request log and
-- nothing fixes this one; FR-ANL-05's metering window is the calendar month, so 90 matches
-- `webhook_attempts` and DR-09's raw-event retention. Said out loud because a number nobody
-- argued for reads like one somebody did.
TTL toDateTime(ts) + INTERVAL 90 DAY

@@ -184,16 +184,96 @@ export function shapeRequest(raw: unknown): RequestRow | null {
 // ---------------------------------------------------------------------------
 
 /** 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 const CONNECTION_OPENED_TYPE = "connection.opened";
+export const CONNECTION_CLOSED_TYPE = "connection.closed";
+
+/** The gateway's record, as it arrives on `analytics.connection.{opened|closed}.{env}`. */
+export interface ConnectionEvent {
+  type: string;
+  connection_id: string;
+  environment_id: string;
+  user_external_id: string;
+  ts: string;
+  /** Close only. */
+  close_code?: number;
+  /** Close only. */
+  duration_ms?: number;
+}
+
+/** One row of `relay_analytics.connection_events`, keyed by column name. */
+export interface ConnectionRow {
+  environment_id: string;
+  ts: string;
+  connection_id: string;
+  event: string;
+  close_code: number | null;
+  duration_ms: number | null;
+  user_external_id: string;
+}
+
+/** Shape one connection record, or return null if it will never be valid.
+ *
+ * `type` IS READ, RENAMED AND DROPPED -- all three, which is one more than the request
+ * shaper does. The wire carries `type: "connection.opened"` because that is what `route()`
+ * discriminates on; the column is `event` and holds `opened`. So this function both drops a
+ * field the table has no column for and derives the column from it, and getting the derivation
+ * backwards is the failure that looks like success: `event` would hold the whole dotted string,
+ * `LowCardinality` would accept it without complaint, and every query filtering
+ * `event = 'opened'` would return nothing for ever.
+ *
+ * `environment_id` IS REQUIRED HERE, unlike the request shaper's. A connection event only
+ * exists after a handshake, so a record arriving without one is malformed rather than
+ * tenantless -- there is no `_none` arm on this grammar to fall back to.
+ *
+ * ABSENT IS NULL, NOT ZERO. A `close_code` of 0 is a claim that a socket closed with code
+ * zero, and a `duration_ms` of 0 that it lasted no time. An open record carries neither. */
+export function shapeConnection(raw: unknown): ConnectionRow | null {
+  if (typeof raw !== "object" || raw === null) return null;
+  const e = raw as Partial<ConnectionEvent>;
+
+  if (
+    !isString(e.environment_id) ||
+    !isString(e.ts) ||
+    !isString(e.connection_id) ||
+    !isString(e.user_external_id) ||
+    !isString(e.type)
+  ) {
+    return null;
+  }
+
+  // THE RENAME, AND IT IS A CLOSED SET RATHER THAN A SUFFIX. `e.type.split(".")[1]` would
+  // turn any `connection.*` record into a row with whatever word followed the dot, which is
+  // a shaper that cannot be wrong about a record it has never seen -- the wrong kind of
+  // robust. Two types, two events, and anything else is somebody else's record.
+  const event =
+    e.type === CONNECTION_OPENED_TYPE
+      ? "opened"
+      : e.type === CONNECTION_CLOSED_TYPE
+        ? "closed"
+        : null;
+  if (event === null) return null;
+
+  return {
+    environment_id: e.environment_id,
+    ts: e.ts,
+    connection_id: e.connection_id,
+    event,
+    close_code: isNumber(e.close_code) ? e.close_code : null,
+    duration_ms: isNumber(e.duration_ms) ? e.duration_ms : null,
+    user_external_id: e.user_external_id,
+  };
+}
 
 export type Shaped =
   | { kind: "attempt"; row: AttemptRow }
   | { kind: "request"; row: RequestRow }
+  | { kind: "connection"; row: ConnectionRow }
   | { 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.
@@ -214,12 +294,19 @@ export function route(raw: unknown): Shaped {
     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 };
   }
+  // The third arm (chapter 4.5). TWO TYPES, ONE ARM, because they are one table: the
+  // subject separates an open from a close for a consumer that wants only one, and this
+  // consumer wants both.
+  if (type === CONNECTION_OPENED_TYPE || type === CONNECTION_CLOSED_TYPE) {
+    const row = shapeConnection(raw);
+    return row === null ? { kind: "malformed" } : { kind: "connection", 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) };
 }

@@ -1,13 +1,14 @@
 // 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, RequestRow } from "./shape.js";
+import type { AttemptRow, ConnectionRow, RequestRow } from "./shape.js";
 
 const DB = "relay_analytics";
 const ATTEMPTS = "webhook_attempts";
 const REQUESTS = "api_requests";
+const CONNECTIONS = "connection_events";
 
 // 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.
@@ -21,14 +22,19 @@ const SETTINGS = "input_format_skip_unknown_fields=0&date_time_input_format=best
 
 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>;
+  /** The third table (chapter 4.5). A third call for the reason there is a second: three row
+   *  shapes are three types, and a `table` parameter would let the compiler watch a
+   *  `ConnectionRow` go into `api_requests` without a word. */
+  insertConnections(rows: ConnectionRow[]): Promise<void>;
   count(): Promise<number>;
   countRequests(): Promise<number>;
+  countConnections(): 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",
@@ -69,16 +75,30 @@ export function createClickHouse({
       if (rows.length === 0) return;
       await post(
         `INSERT INTO ${DB}.${REQUESTS} FORMAT JSONEachRow`,
         rows.map((r) => JSON.stringify(r)).join("\n"),
       );
     },
+    // THE EMPTY GUARD MATTERS MORE WITH EVERY PRODUCER. One fetch now feeds three tables and
+    // the three rates differ by orders of magnitude -- roughly one connection pair per
+    // session against one request record per request -- so most batches carry requests and
+    // neither of the others. Without this, each of them posts an empty INSERT per batch.
+    async insertConnections(rows: ConnectionRow[]): Promise<void> {
+      if (rows.length === 0) return;
+      await post(
+        `INSERT INTO ${DB}.${CONNECTIONS} 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}.${ATTEMPTS} FINAL`, ""));
     },
     async countRequests(): Promise<number> {
       return Number(await post(`SELECT count() FROM ${DB}.${REQUESTS} FINAL`, ""));
     },
+    async countConnections(): Promise<number> {
+      return Number(await post(`SELECT count() FROM ${DB}.${CONNECTIONS} FINAL`, ""));
+    },
   };
 }

@@ -68,19 +68,28 @@ export async function main(): Promise<void> {
 
   do {
     try {
       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.
+      // happening. The per-table figures are separate because one number cannot say which
+      // table moved.
+      //
+      // THREE NOW, AND THE THIRD WAS MISSING FOR A WHOLE PHASE. `IngestResult` gained
+      // `writtenConnections` with chapter 4.5's third arm and this line still reported two,
+      // so a batch carrying connection records logged `written: 3, attempts: 0, requests: 1`
+      // -- a total that does not add up, with the missing half invisible. Found by draining
+      // the real stream and then querying the table: 41 rows in `connection_events` that no
+      // log line had ever mentioned. **A reporter that omits an arm turns "one number cannot
+      // say which table moved" into three numbers that disagree.**
       if (r.written > 0 || r.malformed > 0 || r.unclaimed > 0) {
         logger.log("info", "ingester.batch", {
           written: r.written,
           attempts: r.writtenAttempts,
           requests: r.writtenRequests,
+          connections: r.writtenConnections,
           malformed: r.malformed,
           unclaimed: r.unclaimed,
         });
       }
     } catch (error) {
       // The store is unreachable, or the insert was refused. Nothing was acknowledged, so