The rollup nobody read

grep -rln "daily_usage" --include=*.ts --include=*.mjs . | grep -v node_modules

for t in message_events api_requests connection_events webhook_attempts; do
  echo -n "$t: "
  curl -s -X POST http://localhost:8123/ -u relay:relay \
    --data-binary "SELECT count() FROM relay_analytics.$t FORMAT TSV"
done

-- FR-ANL-05's four quantities, per tenant per day, in ONE ROW.
--
-- A TABLE, NOT A VIEW WITH AN INLINE ENGINE, AND THAT IS THE WHOLE REASON THIS FILE EXISTS.
-- Chapter 4.2's `daily_usage` was written as `CREATE MATERIALIZED VIEW ... ENGINE = ...`,
-- which gives it an IMPLICIT inner table (`.inner_id.<uuid>`) and no name a second view can
-- write into. A materialised view fires on inserts to exactly ONE source table, and this
-- chapter's four quantities come from two -- messages and active users from
-- `message_events`, connection-minutes from `connection_events` -- so a single view cannot
-- carry them.
--
-- Measured before anything was written (research R1): two views over two different sources,
-- both `TO` this table, each naming only its own columns. The columns a view omits arrive at
-- their type's zero and `SummingMergeTree` adds them, so one tenant-day ends up as one row.
-- The probe that first "proved" this had both views writing the SAME column and therefore
-- proved nothing about the shape that ships; the re-run is in `baseline.txt`.
CREATE TABLE IF NOT EXISTS relay_analytics.daily_usage_v2
(
    environment_id  UUID,
    -- FR-ANL-09 names four attribution dimensions -- application, environment, channel, day
    -- -- and `message_events` carries `channel_id` on the row, so channel costs a key column
    -- rather than a join. APPLICATION IS ON NO ROW IN THIS STORE and its mapping lives in
    -- Postgres, so it stays an open item rather than becoming a cross-path read.
    --
    -- A CONNECTION BELONGS TO A TENANT AND NOT TO A CHANNEL, so rows from the connection
    -- view carry the zero UUID here. That is a value a reader will meet in the first query
    -- they write, which is why it is documented rather than discovered.
    channel_id      UUID,
    day             Date,
    messages        UInt64,
    -- AN AGGREGATE STATE, NOT A NUMBER. Written with `uniqState`, read with `uniqMerge`.
    -- `Nullable(UUID)` because 046 measured a NULL `user_id` inserted into a non-nullable
    -- column becoming the ZERO UUID silently -- one phantom active user per environment,
    -- holding a deleted author's messages. 047 then measured `uniqState`/`uniqMerge`
    -- ignoring NULL exactly as `uniqExact` does, so the fix survives into the rollup.
    --
    -- `uniq` IS APPROXIMATE: exact to roughly 60,000-65,000 distinct and off by 0.51% at
    -- 70,000. Any figure published from this column states the corpus cardinality beside it,
    -- and 047-1 is the record of why that matters against FR-ANL-06's 0.1% bound.
    active_users_state AggregateFunction(uniq, Nullable(UUID)),
    -- Int64, NOT UInt64, and it is a DELTA rather than a balance.
    --
    -- The stored message count is the only quantity here that is a stock: every other one is
    -- a daily flow. So the column holds the day's change -- +1 created, -1 deleted -- and the
    -- stored count is the cumulative sum up to and including a day. DR-17 states the
    -- technique for the media analogue in as many words: "summing `media_events` deltas
    -- (uploaded/deleted)".
    --
    -- UNSIGNED WOULD WRAP on any day whose deletions exceed its creations, which is any day
    -- a tenant clears a backlog. Measured signed: 2 created, 1 deleted, 1 edited -> 1.
    stored_delta    Int64,
    connection_minutes UInt64
)
ENGINE = SummingMergeTree
PARTITION BY toYYYYMM(day)
ORDER BY (environment_id, channel_id, day)
-- NO TTL, AND THAT IS DELIBERATE -- the same reasoning FR-003a records for `daily_usage`.
-- `message_events` expires at 90 days (DR-09) and metering must not lose history when raw
-- events do.
--
-- IT ALSO WIDENS AN OPEN ITEM. 048-4 carries "the rollup is still unbounded ... no clause
-- says how long metering history is kept", against a key of (environment_id, day). This key
-- adds `channel_id`, so the product becomes environments x channels x days. The item is
-- re-measured in this feature's gaps.md rather than carried forward as a sentence.

CORPUS_ENVIRONMENTS=3 CORPUS_MESSAGES=200000 CORPUS_DAYS=91 \
  node scripts/scale/corpus.mjs > corpus.json
node scripts/scale/load-analytics.mjs --corpus corpus.json

import type { ClickHouse } from "./clickhouse.js";
 
// FR-ANL-05's four quantities, read from rollup rows and from nothing else.
//
// DR-10: "Materialised views shall maintain daily per-tenant rollups for metering, so billing
// never scans raw events." This is the read that clause is about, and before this chapter
// nothing in the platform performed it -- the only file that had ever asked FR-ANL-05's
// question of the analytical store was `analytics/query.mjs`, referenced by no script, no
// service and no config.
//
// IT GOES THROUGH `ClickHouse` RATHER THAN BESIDE IT. That interface was insert-and-count
// until this chapter; a read placed in this directory with its own `fetch` would open a
// second client against the same four environment variables and make "one client per
// service" a sentence rather than a property.
 
/** One tenant-day's metered totals. */
export interface DailyUsage {
  day: string;
  messages: number;
  activeUsers: number;
  connectionMinutes: number;
}
 
const DB = "relay_analytics";
 
/** FR-ANL-05 per tenant per day, over a closed period.
 *
 * `sum()` WITH `GROUP BY` IS THE CONTRACT, NOT A STYLE. `SummingMergeTree` holds one physical
 * row per key per insert until a background merge, so a bare `SELECT messages` returns one
 * row per insert -- 047 measured it answering `1000 1000 1000` where the truth was 3000. A
 * read that is correct only after somebody has run `OPTIMIZE` is right in a demo and wrong in
 * production.
 *
 * `uniqMerge`, NOT `sum`, for active users: the column is an aggregate state, and `uniq` is
 * approximate above roughly 60,000-65,000 distinct. Any figure published from it states the
 * corpus cardinality beside it.
 *
 * A DAY WITH NO ACTIVITY IS A MISSING ROW, NEVER A ROW OF ZEROS. A materialised view emits
 * nothing for a group that had no input, so the absence of a row is the absence of data --
 * not a measurement of zero. The caller fills gaps if it needs a dense series, and this
 * function does not pretend to. */
export async function dailyUsage(
  store: ClickHouse,
  environmentId: string,
  from: string,
  to: string,
): Promise<DailyUsage[]> {
  const rows = await store.query(
    `SELECT day,
            sum(messages)                 AS messages,
            uniqMerge(active_users_state) AS active_users,
            sum(connection_minutes)       AS connection_minutes
       FROM ${DB}.daily_usage_v2
      WHERE environment_id = toUUID('${environmentId}')
        AND day BETWEEN '${from}' AND '${to}'
      GROUP BY day
      ORDER BY day
      FORMAT TSV`,
  );
  // NO FALLBACKS, AND THAT IS A DESIGN RATHER THAN AN OVERSIGHT. `?? ""` and `?? 0` on each
  // column looked defensive and measured 50% branches: the SELECT above names four columns,
  // so the absent arm cannot arise through the running query and no test can reach it. A
  // branch that cannot go both ways is a branch that is never checked -- 4.5 reached
  // 100/100/100/100 by deleting one, and this is the same move.
  return rows.map((r) => ({
    day: String(r[0]),
    messages: Number(r[1]),
    activeUsers: Number(r[2]),
    connectionMinutes: Number(r[3]),
  }));
}
 
/** The stored message count: a BALANCE, where every other quantity is a flow.
 *
 * NO LOWER BOUND, AND THAT IS THE WHOLE DIFFERENCE. The column holds a day's delta -- +1 for
 * a creation, -1 for a deletion -- so the count of messages stored as of a day is the sum of
 * every delta up to and including it. A `BETWEEN` here would report the period's CHANGE in
 * stored messages, which is a different question that reads as a plausible wrong answer.
 *
 * DR-17 states the technique for the media analogue: "summing `media_events` deltas
 * (uploaded/deleted)". This is the same shape one table over. */
export async function storedMessages(
  store: ClickHouse,
  environmentId: string,
  asOf: string,
): Promise<number> {
  const rows = await store.query(
    `SELECT sum(stored_delta) FROM ${DB}.daily_usage_v2
      WHERE environment_id = toUUID('${environmentId}') AND day <= '${asOf}'
      FORMAT TSV`,
  );
  // NO EMPTY CHECK, BECAUSE THE EMPTY CASE DOES NOT EXIST. A first version guarded
  // `rows.length === 0` and carried a comment claiming a test drove both arms; it did not.
  // **A bare aggregate with no GROUP BY always returns exactly one row** -- asked of the
  // server directly, `sum()` over a tenant with nothing stored answers `0`, not an empty
  // result. The guard was unreachable and measured as half this file's branches.
  //
  // `flat()[0]` rather than `rows[0]?.[0]`: the optional chain is a branch too, and the
  // same one. A tenant with no rows would give NaN here if the server could produce one,
  // and it cannot.
  return Number(rows.flat()[0]);
}

@@ -29,12 +29,24 @@ export interface ClickHouse {
    *  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>;
+  /** A READ, AND THE FIRST ONE THIS INTERFACE HAS HAD (chapter 4.6).
+   *
+   * Everything above writes or counts. Chapter 4.6 needs to ASK the store a question --
+   * FR-ANL-05's four quantities for a tenant and a period -- and a read placed beside this
+   * file without going through it would open a second client against the same four
+   * environment variables. One client per service is the argument 4.5 made for NATS, and it
+   * is a property of the code only if the read comes through here.
+   *
+   * Rows as TSV lines, split by tab. The caller shapes them: this interface has refused a
+   * `table` parameter three times on the grounds that the compiler should watch the types,
+   * and a generic row decoder would be the same mistake one level up. */
+  query(sql: string): Promise<string[][]>;
 }
 
 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",
@@ -53,12 +65,16 @@ export function createClickHouse({
     const text = await res.text();
     if (!res.ok) throw new Error(text.trim().split("\n")[0] ?? "clickhouse insert failed");
     return text.trim();
   };
 
   return {
+    query: async (sql: string): Promise<string[][]> => {
+      const text = await post(sql, "");
+      return text === "" ? [] : text.split("\n").map((line) => line.split("\t"));
+    },
     // 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;

export async function storedMessages(
  store: ClickHouse,
  environmentId: string,
  asOf: string,
): Promise<number> {
  const rows = await store.query(
    `SELECT sum(stored_delta) FROM relay_analytics.daily_usage_v2
      WHERE environment_id = toUUID('${environmentId}') AND day <= '${asOf}'
      FORMAT TSV`,
  );
  return Number(rows.flat()[0]);
}