The consumer that was promised

curl -s 'localhost:8222/jsz?streams=1&consumers=1'

return db.transaction(async (tx) => {
  const claimed = await tx
    .insert(consumedEvents)
    .values({ consumer, eventId })
    .onConflictDoNothing({ target: [consumedEvents.consumer, consumedEvents.eventId] })
    .returning({ eventId: consumedEvents.eventId });
  if (claimed.length === 0) return "duplicate";
  await effect();
  return "handled";
});

-- One row per webhook delivery attempt, shaped by what the publisher already sends rather
-- than by a document: SAD 6.2 publishes `message_events` as *representative* and names
-- `emoji_events`, and no table for delivery attempts is published anywhere. The chapter
-- amends the SAD to publish this one.
CREATE TABLE IF NOT EXISTS relay_analytics.webhook_attempts (
    environment_id  UUID,
    ts              DateTime64(3, 'UTC'),
    delivery_id     UUID,
    endpoint_id     UUID,
    event_id        UUID,
    attempt         UInt8,
    -- NULLABLE BECAUSE THE VALUE IS SOMETIMES NOT KNOWN, AND 0 IS A DIFFERENT CLAIM.
    -- The publisher spreads these in only when present: "an explicit `undefined` is not
    -- the same as an absent key, and the difference is the whole meaning of 'nothing
    -- answered'." A non-nullable `status` would write 0 and assert that an endpoint
    -- answered with status zero. A timeout has no status; inventing one would make every
    -- dashboard built on this lie in the same direction.
    status          Nullable(UInt16),
    error           Nullable(String),
    -- How long the ENDPOINT took to answer. NOT `message_events.delivery_latency_ms`,
    -- which is how long a message took to reach a client and still has no producer.
    latency_ms      UInt32,
    outcome         LowCardinality(String),   -- delivered|rescheduled|dead_lettered
    -- THE ONLY THING BETWEEN A FIELD-NAME TYPO AND AN EMPTY TABLE.
    --
    -- The publisher's field is `attempted_at` and this column is `ts`. A JSONEachRow
    -- insert whose keys do not match column names leaves the column AT ITS DEFAULT with
    -- no error, and a DateTime64 default is the epoch -- which is older than the TTL
    -- below, so the row is deleted at insert. Measured: 0 rows, before and after a merge.
    -- The insert returns OK, the consumer acknowledges, the stream drains to zero, and
    -- every instrument in the chain reports success over an empty table.
    --
    -- `input_format_skip_unknown_fields = 0` on the insert catches a RENAMED field
    -- (Code: 117). It does not catch an ABSENT one -- a row with no `ts` at all takes the
    -- default without complaint. This does.
    CONSTRAINT ts_is_real CHECK ts > toDateTime64('2020-01-01 00:00:00', 3, 'UTC')
)
-- REPLACINGMERGETREE, AND THE SORTING KEY IS ALSO THE DEDUPLICATION KEY.
--
-- `(environment_id, ts)` is the tenant-then-time ordering chapter 4.2's argument is about;
-- `(delivery_id, attempt)` is what makes a record unique -- the same pair the publisher
-- already treats as identity. All four in the sorting key gets the range scans and the
-- deduplication from one declaration, and it is safe because `ts` comes from
-- `attempted_at`, a FIELD OF THE RECORD rather than the time it was consumed: the same
-- record sorts to the same place however it was batched.
--
-- `insert_deduplication_token` would have been cheaper -- the server refuses the duplicate
-- block at insert, with no read cost -- and it cannot be used. It needs a token stable
-- across a redelivery, and JetStream batch boundaries are not: a retry with a different
-- `max_messages` returned 4,5,1,2,3,6,7,8,9,10 where the original batch was 1,2,3,4,5.
-- Worse, the token keys on ITSELF and not the content: the same token with 500 completely
-- different rows dropped all 500 and reported success.
--
-- READS TAKE `FINAL`. The duplicate is physically present until a merge, so a bare
-- SELECT count() over-counts. That is 4.2's rollup lesson one engine over: there the read
-- contract became sum() with GROUP BY, here it is FINAL.
ENGINE = ReplacingMergeTree
PARTITION BY toYYYYMM(ts)                            -- DR-07, as message_events
ORDER BY (environment_id, ts, delivery_id, attempt)
TTL toDateTime(ts) + INTERVAL 90 DAY                 -- DR-09, and toDateTime because the
                                                     -- published form is refused (Code: 450)

// What the publisher sends, and what the table takes. They are not the same shape, and the
// difference is one rename that fails silently if it stops happening.
//
// `JSONEachRow` matches JSON keys to COLUMN NAMES. The publisher's field is `attempted_at`;
// the column is `ts`. An unmatched column takes its DEFAULT with no error -- and a
// DateTime64 default is the epoch, which is older than the table's ninety-day TTL, so the
// row is deleted at insert. The insert returns OK, the consumer acknowledges, the stream
// drains to zero, and the table stays empty while every instrument reports success.
//
// So the ingester SHAPES AND RENAMES rather than forwarding what it was given. Two server
// settings and one table constraint stand behind this function, each catching a different
// way of getting it wrong -- but the function is the thing that has to be right.
 
/** The publisher's record, as it arrives on `analytics.webhook.attempt.{env}`. */
export interface AttemptEvent {
  delivery_id: string;
  endpoint_id: string;
  environment_id: string;
  event_id: string;
  attempt: number;
  attempted_at: string;
  /** Absent when nothing answered. A timeout has no status. */
  status?: number;
  /** Present when there was no status. */
  error?: string;
  latency_ms: number;
  outcome: string;
}
 
/** One row of `relay_analytics.webhook_attempts`, keyed by column name. */
export interface AttemptRow {
  environment_id: string;
  ts: string;
  delivery_id: string;
  endpoint_id: string;
  event_id: string;
  attempt: number;
  status: number | null;
  error: string | null;
  latency_ms: number;
  outcome: string;
}
 
const isString = (v: unknown): v is string => typeof v === "string" && v.length > 0;
const isNumber = (v: unknown): v is number => typeof v === "number" && Number.isFinite(v);
 
/** Shape one record, or return null if it will never be valid.
 *
 * NULL MEANS TERMINATE, NOT RETRY. The consumer sets no redelivery limit, so a payload that
 * cannot be parsed would otherwise come back until the stream's retention expires. The same
 * bytes fail the same way every time; the api's own consumer runtime draws this line in the
 * same words.
 *
 * An absent `status` or `error` becomes NULL rather than 0 or "". Zero is a measurement and
 * NULL is an absence: a `status` of 0 asserts that an endpoint answered with status zero. */
export function shape(raw: unknown): AttemptRow | null {
  if (typeof raw !== "object" || raw === null) return null;
  const e = raw as Partial<AttemptEvent>;
 
  if (
    !isString(e.environment_id) ||
    !isString(e.attempted_at) ||
    !isString(e.delivery_id) ||
    !isString(e.endpoint_id) ||
    !isString(e.event_id) ||
    !isNumber(e.attempt) ||
    !isNumber(e.latency_ms) ||
    !isString(e.outcome)
  ) {
    return null;
  }
 
  return {
    environment_id: e.environment_id,
    // THE RENAME. Everything else is a copy; this is the line the table cannot survive
    // being wrong, because being wrong about it looks exactly like success.
    ts: e.attempted_at,
    delivery_id: e.delivery_id,
    endpoint_id: e.endpoint_id,
    event_id: e.event_id,
    attempt: e.attempt,
    status: isNumber(e.status) ? e.status : null,
    error: isString(e.error) ? e.error : null,
    latency_ms: e.latency_ms,
    outcome: e.outcome,
  };
}

// 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";
 
const DB = "relay_analytics";
const TABLE = "webhook_attempts";
 
// 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.
//
// `date_time_input_format=best_effort` is what parses the ISO-8601 string at all. The
// default `basic` refuses it outright -- `Code: 27. Cannot parse input: expected '"' before
// 'Z"...'` -- which is the loud half of the pair, and the least dangerous.
//
// 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>;
  count(): 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",
  password = process.env["RELAY_CLICKHOUSE_PASSWORD"] ?? "relay",
}: {
  host?: string;
  port?: string;
  user?: string;
  password?: string;
} = {}): ClickHouse {
  const auth = "Basic " + Buffer.from(`${user}:${password}`).toString("base64");
 
  const post = async (query: string, body: string): Promise<string> => {
    const url = `http://${host}:${port}/?${SETTINGS}&query=${encodeURIComponent(query)}`;
    const res = await fetch(url, { method: "POST", headers: { Authorization: auth }, body });
    const text = await res.text();
    if (!res.ok) throw new Error(text.trim().split("\n")[0] ?? "clickhouse insert failed");
    return text.trim();
  };
 
  return {
    // One statement, one block. No deduplication token: the table is a ReplacingMergeTree
    // 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`,
        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`, ""));
    },
  };
}

// The ingester — the consumer chapter 3.20 promised and did not build.
//
// `ANALYTICS` has carried one record per webhook delivery attempt since that chapter, with
// seven-day retention and `discard: old`, and nothing has ever read it. The stream's own
// comment says so: "no acknowledgement anywhere: nothing consumes this stream in this
// chapter." This is the chapter it was waiting for.
//
// IT DOES NOT REUSE `createConsumerRuntime`, AND THAT IS THE ARGUMENT RATHER THAN AN
// 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 { ALL_ANALYTICS_SUBJECT, ANALYTICS_STREAM } from "@relay/protocol";
import { createLogger, type Logger } from "@relay/service-kit";
 
import { createClickHouse, type ClickHouse } from "./clickhouse.js";
import { shape, type AttemptRow } from "./shape.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
 * dispatcher, each at a 30-second ack_wait. For a webhook endpoint that is sound: one that
 * has failed ten times is probably gone, and retrying it forever helps nobody.
 *
 * A store that is restarting is not an endpoint that is gone. Measured at max_deliver 3: the
 * message is delivered on rounds 1, 2 and 3 and NEVER AGAIN, while the stream still holds it
 * and the consumer reports `num_pending 0`. At five attempts and thirty seconds, two and a
 * half minutes of ClickHouse being down strands everything in flight -- and the loss hides
 * from the instrument you would reach for, because stream depth stays high and consumer lag
 * goes to zero.
 *
 * 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 });
  const jsm = await nc.jetstreamManager();
  await jsm.consumers.add(ANALYTICS_STREAM, {
    durable_name: DURABLE,
    ack_policy: AckPolicy.Explicit,
    ack_wait: ACK_WAIT_NS,
    max_deliver: MAX_DELIVER,
    filter_subject: ALL_ANALYTICS_SUBJECT,
  }).catch(() => undefined); // already there; leave it alone
 
  const store = createClickHouse();
  let running = true;
  const stop = (): void => {
    running = false;
  };
  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 });
      }
    } 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));
    }
  } while (running && !once);
 
  await nc.close();
}
 
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((error: unknown) => {
    process.stderr.write(`ingester failed: ${String(error)}\n`);
    process.exit(1);
  });
}

{
  "name": "@relay/ingester",
  "private": true,
  "version": "0.0.0",
  "type": "module",
  "scripts": {
    "dev": "tsx watch src/main.ts",
    "build": "tsc -p tsconfig.build.json",
    "start": "node dist/main.js",
    "typecheck": "tsc --noEmit",
    "test": "vitest run",
    "test:integration": "vitest run --config vitest.integration.config.mts"
  },
  "dependencies": {
    "@relay/protocol": "workspace:*",
    "@relay/service-kit": "workspace:*",
    "nats": "^2.29.3"
  },
  "devDependencies": {
    "tsx": "^4.23.1"
  }
}