Quotas and what they cost

select count(*), count(distinct m.user_id)
  from messages m join channels c on c.id = m.channel_id
 where c.environment_id = $1
   and m.created_at >= date_trunc('month', now() at time zone 'utc');

->  Bitmap Heap Scan on messages m
      Recheck Cond: (c.id = channel_id)
      Filter: (created_at >= date_trunc('month'::text, (now() AT TIME ZONE 'utc'::text)))
      Heap Blocks: exact=11
Execution Time: 0.266 ms

  it("reports identical figures across a FLUSHALL", async () => {
    const env = await createEnvironment(db, {
      name: `quota-flush-${randomUUID().slice(0, 8)}`,
    });
    const repo = new Repository(db, env.id);
    const channel = await repo.createChannel(
      `c-${randomUUID().slice(0, 8)}`,
      "public",
    );
    const userId = (await repo.createUser(`u-${randomUUID().slice(0, 8)}`)).id;
    for (let i = 0; i < 3; i++) {
      await repo.sendMessage(channel.id, { text: `m${i}`, userId });
    }
 
    const before = await usageFor(db, env.id, PERIOD);
    expect(before.messagesSent).toBe(3);
 
    // The whole store, not this environment's keys. The rate-limit chapter's counters and
    // everything else go with it.
    const redis = new Redis(
      process.env["RELAY_REDIS_URL"] ?? "redis://localhost:6379",
    );
    try {
      await redis.flushall();
    } finally {
      await redis.quit();
    }
 
    const after = await usageFor(db, env.id, PERIOD);
    expect(after).toEqual(before);

    quotaConfig: jsonb("quota_config").notNull().default({}),

{
  "messages":     { "hard": 10000, "soft": 8000 },
  "active_users": { "hard": null,  "soft": 500  }
}

ALTER TABLE environments
  ADD CONSTRAINT environments_quota_config_shape CHECK (
    jsonb_typeof(quota_config) = 'object'
    AND (quota_config -> 'messages' IS NULL
         OR jsonb_typeof(quota_config -> 'messages') = 'object')
    AND (quota_config -> 'active_users' IS NULL
         OR jsonb_typeof(quota_config -> 'active_users') = 'object')
    AND (quota_config #>> '{messages,hard}' IS NULL
         OR quota_config #>> '{messages,hard}' ~ '^[0-9]+$')
    AND (quota_config #>> '{messages,soft}' IS NULL
         OR quota_config #>> '{messages,soft}' ~ '^[0-9]+$')
    AND (quota_config #>> '{active_users,hard}' IS NULL
         OR quota_config #>> '{active_users,hard}' ~ '^[0-9]+$')
    AND (quota_config #>> '{active_users,soft}' IS NULL
         OR quota_config #>> '{active_users,soft}' ~ '^[0-9]+$')
  );

ERROR:  cannot use subquery in check constraint

CREATE TABLE usage_active_users (
  environment_id uuid        NOT NULL REFERENCES environments(id),
  period         date        NOT NULL,
  user_id        uuid        NOT NULL REFERENCES users(id),
  first_seen_at  timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (environment_id, period, user_id)
);

      // The distinct-user count, and the reason it is a row rather than a
      // counter: incrementing it would need to know whether this user already
      // sent this period, which is a read. The row IS the answer, and
      // `ON CONFLICT DO NOTHING` makes the second send of the month free.
      //
      // ONLY WHEN THE SEND IS ATTRIBUTED. A key-authenticated REST send carries
      // no `userId` — unattributed by design since the outbox chapter — and counts
      // toward the message quota and toward no user.
      if (userId !== undefined) {
        await tx
          .insert(usageActiveUsers)
          .values({ environmentId: this.environmentId, period, userId })
          .onConflictDoNothing();
      }

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"];

      // THE CAP, CHECKED BEFORE THE MESSAGE IS WRITTEN (FR-RTL-08).
      //
      // Here rather than in middleware, because the rate-limit chapter's limiter never sees
      // `/internal/messages` — `operationsFor` returns [] for anything outside
      // `/v1` — and that is the route a WebSocket send arrives on. Both doors
      // reach this method, and it already owns the write transaction, so the
      // check and the increment commit together (research R3).
      //
      // A PLAIN READ, AND THE OVERSHOOT IS STATED RATHER THAN DEFENDED AGAINST.
      //
      // The first version took `FOR UPDATE` on the usage row, which bounds the
      // overshoot to exactly one message. Two things retired it.
      //
      // The caps and the usage are now ONE joined read, and Postgres will not
      // lock that:
      //
      //   ERROR:  FOR UPDATE cannot be applied to the nullable side of an outer join
      //
      // And the specification never asked for the lock. Its edge case reads: "the
      // overshoot is bounded by concurrency, not unbounded, and this is stated
      // rather than defended against." A few dozen sends in flight against a
      // monthly cap of thousands is a bound worth naming rather than engineering
      // around.
      //
      // WHAT THE QUOTA PATH COSTS, measured with the phases instrumented and the
      // config toggled on one environment: 0.56ms per send at 32-way concurrency.
      // The joined read is about 1.2ms of that and US1 needs it whether or not a
      // cap exists. An earlier uncontrolled benchmark reported 273% and sent three
      // separate hypotheses chasing what turned out to be warm-up (T033).
      const quota = await this.assertWithinQuota(tx, period, userId);

   * IN THE SAME TRANSACTION AS THE THING THAT CAUSED IT. The crossing and the
   * message commit together or neither does, which is the same argument the
   * event above them makes and the reason there is no periodic sweep in this
   * chapter at all: usage only ever rises because of a send, and the send knows
   * the value before and after, so it knows what it crossed (research R5).
   *
   * THE PERCENTAGE IS OF `hard ?? soft`. A soft threshold with no hard cap is
   * still a figure an operator asked to be warned about, and 100% of it is worth
   * an email even though nothing will be refused.
   *
   * `ON CONFLICT DO NOTHING` against `quota_notifications_once_per_threshold` is
   * what makes it at-most-once (FR-RTL-07) — the schema, not this code. A concurrent
   * double-crossing resolves to one row rather than two emails. */
  private async recordCrossings(
    tx: Db,
    period: string,
    dimension: Dimension,
    before: number,
    after: number,
    caps: { hard: number | null; soft: number | null },
    organisationId: string,
  ): Promise<void> {
    const reference = caps.hard ?? caps.soft;
    if (reference === null) return;
    const crossed = thresholdsCrossed(before, after, reference);
    if (crossed.length === 0) return;
 
    await tx
      .insert(quotaNotifications)
      .values(
        crossed.map((threshold) => ({
          id: randomUUID(),
          environmentId: this.environmentId,
          organisationId,
          period,
          dimension,
          threshold,
          quota: reference,
          usageAtCrossing: after,
        })),
      )
      .onConflictDoNothing();

      if (error instanceof QuotaExceededError) {
        // ONE THROW, AND IT IS THE ONLY ONE (FR-RTL-08).
        //
        // Both send routes reach this method — `internal.controller.ts` calls
        // `messages.send`, the public controller calls it too — so there is one
        // place to refuse from. An earlier draft of the plan costed "two
        // controller mappings"; this service has no per-controller mappings to
        // add one to, and adding two would be the drift EIR-API-04 and
        // `ProtocolErrorFilter` exist to prevent (research R3).
        //
        // `402`, NOT `429`. The rate-limit chapter owns `429`, and a client that sleeps for
        // `Retry-After` and retries is behaving correctly for a rate limit and
        // wrongly for a quota — which will still be exhausted in an hour and in
        // three weeks. There is a time at which sends resume and it is in the
        // message, not in a header a client will act on.
        //
        // THE CODE IS NAMED HERE, and it has to be. `ProtocolErrorFilter` infers
        // a code from the status for 400, 401, 403 and 404, and everything else
        // becomes `internal_error` — so an unnamed `402` would emit a body
        // calling itself an internal error while carrying a `402`. That is the
        // lie chapter 2.2 fixed for 400 and the credentials chapter for 403, and the credentials chapter's
        // mechanism — a thrower naming its own code — is what this uses. The
        // filter builds the four-field envelope and derives `docs_url` from the
        // code.
        throw new HttpException(
          {
            code: "quota_exceeded",
            message: error.publicMessage(),
          },
          HttpStatus.PAYMENT_REQUIRED,
        );
      }

CREATE TABLE quota_notifications (
  id                uuid        PRIMARY KEY,
  environment_id    uuid        NOT NULL REFERENCES environments(id),
  organisation_id   uuid        NOT NULL REFERENCES organisations(id),
  period            date        NOT NULL,
  dimension         text        NOT NULL,
  threshold         integer     NOT NULL,
  quota             bigint      NOT NULL,
  usage_at_crossing bigint      NOT NULL,
  crossed_at        timestamptz NOT NULL DEFAULT now(),
  delivered_at      timestamptz,
  last_error        text,
  CONSTRAINT quota_notifications_dimension_check
    CHECK (dimension IN ('messages', 'active_users')),
  CONSTRAINT quota_notifications_threshold_check
    CHECK (threshold IN (50, 80, 100)),
  CONSTRAINT quota_notifications_once_per_threshold
    UNIQUE (environment_id, period, dimension, threshold)
);

  CONSTRAINT quota_notifications_once_per_threshold
    UNIQUE (environment_id, period, dimension, threshold)

      // THE MONTH'S USAGE COMMITS WITH THE MESSAGE (FR-RTL-05).
      //
      // Same argument as the event above it, one requirement further on. A quota
      // is about THIS MONTH and must not forget, so the count cannot live in the
      // per-minute counter store the rate-limit chapter built — a flush there costs one
      // window of over-service, a flush here costs the month (a quota must survive the counter store).
      //
      // It is an increment rather than a query because the alternative is a read
      // over `messages`, which carries no `environment_id` and no index on
      // `created_at`: the month predicate becomes a Filter applied after every
      // row the tenant has ever sent is read off the heap. Fast today, and
      // proportional to lifetime traffic forever (research R1).
      //
      // On the INSERTED branch only, like the event. A recognised idempotent
      // retry wrote no message and must consume no quota either, or a client
      // retrying on a flaky link is billed twice for one message.
      await tx
        .insert(usagePeriods)
        .values({ environmentId: this.environmentId, period, messagesSent: 1 })
        .onConflictDoUpdate({
          target: [usagePeriods.environmentId, usagePeriods.period],
          set: { messagesSent: sql`${usagePeriods.messagesSent} + 1` },
        });
 
      // The distinct-user count, and the reason it is a row rather than a
      // counter: incrementing it would need to know whether this user already
      // sent this period, which is a read. The row IS the answer, and
      // `ON CONFLICT DO NOTHING` makes the second send of the month free.
      //
      // ONLY WHEN THE SEND IS ATTRIBUTED. A key-authenticated REST send carries
      // no `userId` — unattributed by design since the outbox chapter — and counts
      // toward the message quota and toward no user.
      if (userId !== undefined) {
        await tx
          .insert(usageActiveUsers)
          .values({ environmentId: this.environmentId, period, userId })
          .onConflictDoNothing();

unconfigured           1.77ms per send    assertWithinQuota = 1.246ms
configured             2.33ms per send    assertWithinQuota = 1.586ms
back to unconfigured   1.77ms per send    assertWithinQuota = 1.121ms

-- Monthly usage quotas (FR-RTL-05 to FR-RTL-08).
--
-- The rate-limit chapter built the per-minute limiter. This is the other half of FR-RTL and
-- the two are different problems wearing the same word: a rate limit is about
-- THIS SECOND and forgets, a quota is about THIS MONTH and must not. Everything
-- below follows from the second half of that sentence.
--
-- WHY A ROLL-UP AND NOT A QUERY OVER `messages`. Deriving usage on read is one
-- statement and needs no tables at all, and it is what this chapter argues
-- against. `messages` carries no `environment_id` — it hangs off `channels` — and
-- no index on `created_at`, so the month predicate is a FILTER applied after the
-- rows are read:
--
--     ->  Bitmap Heap Scan on messages m
--           Recheck Cond: (c.id = channel_id)
--           Filter: (created_at >= date_trunc('month', ...))
--
-- The work is proportional to everything the tenant has ever sent. At 507
-- messages it measures a quarter of a millisecond, which is why the argument is
-- the plan and not the clock.
 
-- ---------------------------------------------------------------------------
-- The policy: the column chapter 2.1 left empty.
-- ---------------------------------------------------------------------------
--
-- THERE IS NO NEW POLICY COLUMN, because `environments.quota_config` has been
-- sitting there since `0000_core_tables.sql` — declared in chapter 2.1, named in
-- SRS §6.1, and read by nothing for eighteen chapters. The rate-limit chapter was offered it
-- for rate-limit policy and refused, in prose, on the grounds that "the column is
-- named for quotas, quotas are a later chapter". This is that chapter.
--
-- Shape:
--
--     { "messages":     { "hard": 10000, "soft": 8000 },
--       "active_users": { "hard": null,  "soft": 500  } }
--
-- ABSENT AND NULL BOTH MEAN NO CAP. ZERO MEANS REFUSE EVERYTHING. That is the
-- same rule 0008 wrote for the limit columns, and jsonb keeps it expressible:
-- `#>> '{messages,hard}'` returns SQL NULL for an absent key and for a JSON null
-- alike, and the string `'0'` for zero. The distinction the rate-limit chapter
-- needed nullable columns for survives the move.
--
-- WHAT THE JSONB BUYS: THE CONNECTION-METERING CHAPTER adds connection-minutes and FR-MED-12 later
-- adds media bytes, and neither needs a table migration — a new dimension is a
-- new key.
--
-- WHAT IT COSTS, said plainly rather than discovered later: the shape below is
-- enforced by a CHECK that ENUMERATES the two dimensions, so a third one does
-- cost a one-line constraint change to keep the guarantee. A constraint that
-- validated any dimension would need `jsonb_each`, and a CHECK may not contain a
-- subquery —
--
--     ERROR:  cannot use subquery in check constraint
--
-- which is the restriction feature 030's R37 met from the other side, in a
-- trigger's WHEN clause. The alternative is a PL/pgSQL validator, and a procedural
-- function in a PRODUCT migration is a constitution VII argument this chapter has
-- not earned; feature 030's guard is exempt precisely because it is never a
-- migration.
--
-- The regex rather than a cast: `(… )::bigint` inside a CHECK throws on bad input
-- instead of rejecting the row, and a constraint that errors is worse than one
-- that refuses.
 
ALTER TABLE environments
  ADD CONSTRAINT environments_quota_config_shape CHECK (
    jsonb_typeof(quota_config) = 'object'
    AND (quota_config -> 'messages' IS NULL
         OR jsonb_typeof(quota_config -> 'messages') = 'object')
    AND (quota_config -> 'active_users' IS NULL
         OR jsonb_typeof(quota_config -> 'active_users') = 'object')
    AND (quota_config #>> '{messages,hard}' IS NULL
         OR quota_config #>> '{messages,hard}' ~ '^[0-9]+$')
    AND (quota_config #>> '{messages,soft}' IS NULL
         OR quota_config #>> '{messages,soft}' ~ '^[0-9]+$')
    AND (quota_config #>> '{active_users,hard}' IS NULL
         OR quota_config #>> '{active_users,hard}' ~ '^[0-9]+$')
    AND (quota_config #>> '{active_users,soft}' IS NULL
         OR quota_config #>> '{active_users,soft}' ~ '^[0-9]+$')
  );
 
-- ---------------------------------------------------------------------------
-- The roll-up.
-- ---------------------------------------------------------------------------
--
-- `period` IS STORED, NOT COMPUTED. It is the first day of the calendar month in
-- UTC, and storing it makes a lookup the whole primary key rather than a
-- predicate over a range — a month boundary becomes a different row instead of a
-- different filter. `services/api/src/quotas/period.ts` is the one definition of
-- which month an instant belongs to; nothing here repeats `date_trunc`.
--
-- THIS IS THE PROJECT'S FIRST `date` COLUMN, against 28 `timestamp` ones, and it
-- is half a primary key. Drizzle's `date` reads and writes `YYYY-MM-DD` strings,
-- so the TypeScript side hands strings across; a `Date` on one side of that
-- comparison and a string on the other is a row that cannot be found rather than
-- an error (research R7a).
--
-- `messages_sent` IS `bigint`, declared `{ mode: "number" }` in the schema like
-- the two bigints this project already has. It is a cumulative count on the hot
-- path, and an overflow here is a wrong bill rather than a wrapped counter.
 
CREATE TABLE usage_periods (
  environment_id uuid        NOT NULL REFERENCES environments(id),
  period         date        NOT NULL,
  messages_sent  bigint      NOT NULL DEFAULT 0,
  created_at     timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (environment_id, period),
  CONSTRAINT usage_periods_messages_sent_non_negative
    CHECK (messages_sent >= 0)
);
 
-- ---------------------------------------------------------------------------
-- The distinct-user membership.
-- ---------------------------------------------------------------------------
--
-- A message count is `+1`. A DISTINCT-USER COUNT IS NOT: incrementing it requires
-- knowing whether this user has already sent this period, which is a read. So the
-- row is the answer — one per user per period, written `ON CONFLICT DO NOTHING`
-- on every attributed send, and the count is an index-only scan over the key
-- prefix.
--
-- Bounded by the tenant's distinct users per month rather than by their traffic,
-- which is what makes it affordable and the reason it is a table and not a
-- counter. HyperLogLog in Redis is the textbook answer and is refused by the rule
-- above: a flush would erase the month.
--
-- EVERY SEND WRITES A ROW, BECAUSE EVERY SEND HAS A SENDER. This paragraph used to
-- say the opposite — that a key-authenticated REST send is unattributed by design,
-- and that an unattributed send counts toward the message quota and toward no
-- user — which was true of a platform whose `sendMessage` took an optional
-- `userId`. The sender chapter made it required, so the state described here is one
-- no write path can reach and the count of distinct users IS the count of senders.
 
CREATE TABLE usage_active_users (
  environment_id uuid        NOT NULL REFERENCES environments(id),
  period         date        NOT NULL,
  user_id        uuid        NOT NULL REFERENCES users(id),
  first_seen_at  timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (environment_id, period, user_id)
);
 
-- ---------------------------------------------------------------------------
-- The outbox, a fourth time.
-- ---------------------------------------------------------------------------
--
-- The outbox chapter published events, the webhook dispatcher chapter dispatched
-- webhook deliveries, the mail-transport chapter sent disablement emails. Each is
-- a table whose claim predicate starts null, drained by a relay, retried by
-- falling due again. This is the fourth, and saying the
-- number out loud is the point: four concrete tables that look alike is a
-- pattern, one abstract table serving four purposes is a framework.
--
-- `webhook_disable_notifications` CANNOT BE REUSED — its `endpoint_id` is NOT
-- NULL and a quota crossing has no endpoint.
--
-- THE UNIQUE CONSTRAINT IS FR-RTL-07. "At most one email per threshold per quota per
-- period" is enforced by the schema rather than promised by the code that writes
-- it, so a concurrent double-crossing resolves to one row instead of two emails.
--
-- `quota` and `usage_at_crossing` are STORED rather than looked up at send time,
-- because the cap can change between the crossing and the delivery, and an email
-- saying "you have used 80% of 10,000" should mean the 10,000 that was true when
-- it happened.
 
CREATE TABLE quota_notifications (
  id                uuid        PRIMARY KEY,
  environment_id    uuid        NOT NULL REFERENCES environments(id),
  organisation_id   uuid        NOT NULL REFERENCES organisations(id),
  period            date        NOT NULL,
  dimension         text        NOT NULL,
  threshold         integer     NOT NULL,
  quota             bigint      NOT NULL,
  usage_at_crossing bigint      NOT NULL,
  crossed_at        timestamptz NOT NULL DEFAULT now(),
  delivered_at      timestamptz,
  last_error        text,
  CONSTRAINT quota_notifications_dimension_check
    CHECK (dimension IN ('messages', 'active_users')),
  CONSTRAINT quota_notifications_threshold_check
    CHECK (threshold IN (50, 80, 100)),
  CONSTRAINT quota_notifications_once_per_threshold
    UNIQUE (environment_id, period, dimension, threshold)
);
 
-- The claim predicate the relay drains on, matching the mail-transport chapter's shape.
CREATE INDEX quota_notifications_undelivered
  ON quota_notifications (crossed_at)
  WHERE delivered_at IS NULL;

/** The usage period an instant belongs to: the first day of its calendar month,
 * in UTC, as a plain `YYYY-MM-DD` string.
 *
 * ONE DEFINITION, IMPORTED BY EVERYTHING. The migration's default, the
 * repository's predicate and the relay's read all name this function rather than
 * repeating `date_trunc`, because a quota that disagrees with itself about which
 * month it is counts a tenant twice in one and not at all in the other.
 *
 * A STRING, NOT A `Date`. The column is a Postgres `date` — this project's first,
 * against 28 `timestamp` columns — and `period` is half the primary key of
 * `usage_periods` and a third of `usage_active_users`'s. Drizzle's `date` in its
 * default mode reads and writes `YYYY-MM-DD` strings, so a string here is the
 * value the key is actually built from; handing a `Date` around instead would put
 * a timezone-bearing object on both sides of a comparison that has no timezone,
 * and the failure would be a row that cannot be found rather than an error
 * (research R7a).
 *
 * UTC, and the tests say why. `date_trunc('month', now())` without a zone answers
 * September on a server running ahead of UTC on the last evening of August, and
 * the row lands in a period nobody reads. */
export function periodOf(at: Date): string {
  const year = at.getUTCFullYear();
  const month = String(at.getUTCMonth() + 1).padStart(2, "0");
  return `${year}-${month}-01`;
}
 
/** The period AFTER this one — the month a refused tenant's sends resume in.
 *
 * ONE COPY, AND THERE WERE TWO. `quota.error.ts` and `quota-email.ts` each carried
 * this arithmetic: the same December wrap, the same four `??` fallbacks, differing
 * only in what they did with the answer — the error returns `2027-01-01` and the
 * email says `January 2027`. Two copies of a rule are two rules, and the rule here is
 * "which month does the refusal end in", which a customer reads on an invoice.
 *
 * The fallbacks live here now, once. They exist because `split("-").map(Number)` is
 * typed `(number | undefined)[]` and cannot be narrowed by the compiler, not because
 * a period can be malformed — every one comes from `periodOf` above or from a
 * Postgres `date` column. That makes them unreachable arms, and having them in one
 * file rather than two is the difference between one uncoverable branch pair and
 * three. */
export function nextPeriod(period: string): string {
  const [y, m] = period.split("-").map(Number);
  const nextMonth = m === 12 ? 1 : (m ?? 1) + 1;
  const nextYear = m === 12 ? (y ?? 0) + 1 : y;
  return `${nextYear}-${String(nextMonth).padStart(2, "0")}-01`;
}

/** The percentages an organisation is emailed at (FR-RTL-07). */
export const THRESHOLDS = [50, 80, 100] as const;
 
/** Which thresholds a usage increase crossed, ascending.
 *
 * Called inside the send transaction, between the increment and the cap check,
 * so it takes the two numbers the transaction already holds and asks nothing
 * else. No database, no clock, no rounding policy hidden in a helper.
 *
 * `quota` is null for an environment with no cap configured, and null crosses
 * nothing at any usage — the absent state stays absent rather than becoming
 * `Infinity` or `-1` somewhere up the call stack.
 *
 * A quota of ZERO is a different thing from an absent one: it means refuse
 * everything, and every threshold is already met. Guarded before the division
 * rather than after it. */
export function thresholdsCrossed(
  before: number,
  after: number,
  quota: number | null,
): number[] {
  if (quota === null) return [];
  if (after <= before) return [];
  if (quota === 0) return [...THRESHOLDS];
 
  const pct = (n: number) => (n / quota) * 100;
  const from = pct(before);
  const to = pct(after);
  // `>` on the left and `>=` on the right: FR-RTL-07 says "reaches", so landing
  // exactly on 50% crosses it, and starting exactly on 50% does not cross it
  // again.
  return THRESHOLDS.filter((t) => from < t && to >= t);
}

import { z } from "zod";
 
/** What `environments.quota_config` holds, and the only thing that reads it.
 *
 * THE COLUMN IS THE ONE CHAPTER 2.1 LEFT EMPTY. Declared in
 * `0000_core_tables.sql`, named in SRS §6.1, read by nothing for eighteen
 * chapters. The rate-limit chapter was offered it for rate-limit policy and refused in
 * prose — "the column is named for quotas, quotas are a later chapter". This is
 * that chapter.
 *
 * WHY A PARSER AT ALL. THE RATE-LIMIT CHAPTER'S limits are typed columns and need no
 * parsing; a jsonb column arrives as `unknown` and something has to turn it into
 * numbers before a cap can be compared. The alternative is a cast at each read
 * site, which is three places to get wrong instead of one.
 *
 * The schema's CHECK constraint already refuses a negative, a non-number and a
 * non-object — measured, not assumed. This is the second gate rather than the
 * only one, and it exists because the constraint cannot express "and nothing
 * else", while a parser can. */
const capsSchema = z
  .object({
    /** Absent or null: no cap. Zero: refuse everything. */
    hard: z.number().int().nonnegative().nullable().optional(),
    /** Absent or null: no alert. Alerts, never refuses. */
    soft: z.number().int().nonnegative().nullable().optional(),
  })
  .strict();
 
export const quotaConfigSchema = z
  .object({
    messages: capsSchema.optional(),
    active_users: capsSchema.optional(),
  })
  // `.strict()` so a dimension nobody implemented is a parse failure rather than
  // a silently ignored cap. The connection-metering chapter adds connection-minutes by adding a key
  // here and a line to the migration's CHECK — the cost the jsonb shape trades
  // for not needing a table migration.
  .strict();
 
export type QuotaConfig = z.infer<typeof quotaConfigSchema>;
 
/** One dimension's caps, resolved. `null` on either means no cap and no alert;
 * the absent state stays absent all the way to the reader rather than becoming
 * `Infinity` or `-1` somewhere up the stack. */
export interface Caps {
  hard: number | null;
  soft: number | null;
}
 
export const NO_CAPS: Caps = { hard: null, soft: null };
 
/** Read one dimension out of whatever the column held.
 *
 * FAILS CLOSED ON A PARSE ERROR — the caller gets `NO_CAPS` and a reason, and a
 * quota that cannot be read refuses nothing rather than refusing everything. A
 * malformed config is an operator's mistake, and suspending a tenant's sends
 * because their configuration is unparseable would turn a typo into an outage.
 * The caller logs; it does not swallow. */
export function capsFor(
  raw: unknown,
  dimension: keyof QuotaConfig,
): { caps: Caps; error: string | null } {
  const parsed = quotaConfigSchema.safeParse(raw ?? {});
  if (!parsed.success) {
    return { caps: NO_CAPS, error: parsed.error.issues[0]?.message ?? "invalid" };
  }
  const d = parsed.data[dimension];
  return {
    caps: { hard: d?.hard ?? null, soft: d?.soft ?? null },
    error: null,
  };
}

import type { QuotaConfig } from "./config";
 
import { nextPeriod } from "./period";
 
/** The dimensions a quota is measured in. `connection_minutes` belongs to the
 * connection-metering chapter. */
export type Dimension = keyof QuotaConfig;
 
/** Raised by the repository when a send would exceed a hard cap.
 *
 * NOT AN HTTP CONCERN. The repository layer does not know what status a caller
 * will map this to, and it holds the four things the message has to name:
 * which dimension, what was used, what was allowed, and which period. Turning
 * that into a `402` is the service boundary's job, and turning it into an
 * envelope is `ProtocolErrorFilter`'s — one place, not three (research R3). */
export class QuotaExceededError extends Error {
  readonly dimension: Dimension;
  readonly usage: number;
  readonly quota: number;
  readonly period: string;
 
  constructor(args: {
    dimension: Dimension;
    usage: number;
    quota: number;
    period: string;
  }) {
    super(
      `${args.dimension} quota exhausted: ${args.usage} of ${args.quota} for ${args.period}`,
    );
    this.name = "QuotaExceededError";
    this.dimension = args.dimension;
    this.usage = args.usage;
    this.quota = args.quota;
    this.period = args.period;
  }
 
  /** The date sends resume: midnight UTC on the first of the next month.
   *
   * In the message rather than in a `Retry-After` header, and that is the whole
   * argument for `402` over `429`. A client that sleeps for the header's value
   * and retries is behaving correctly for a rate limit and wrongly for a quota,
   * which will still be exhausted in an hour and in a week. */
  resumesOn(): string {
    return nextPeriod(this.period);
  }
 
  /** The sentence a developer reads in a log at 3am. Four things in a fixed
   * order: the dimension, the figure used, the figure allowed, and when it
   * changes (contracts/quota.md §1). */
  publicMessage(): string {
    return (
      `monthly ${this.dimension === "messages" ? "message" : "active user"} ` +
      `quota exhausted: ${this.usage} of ${this.quota} for ${this.period}; ` +
      `sends resume on ${this.resumesOn()}`
    );
  }
}

import type { Mail } from "../notifications/mailer";
 
import { nextPeriod } from "./period";
 
export interface CrossingFacts {
  /** "Fleet Ops / production" — how a dashboard would name it, never a uuid. */
  environmentName: string;
  period: string;
  dimension: string;
  threshold: number;
  quota: number;
  usageAtCrossing: number;
  /** Whether a hard cap is in force right now, which decides whether this email
   * reports a stoppage or a warning. */
  hardCapInForce: boolean;
}
 
const NOUN: Record<string, string> = {
  messages: "messages",
  active_users: "active users",
};
 
/** The month, as a month. `2026-08-01` is a row key, not something to show a
 * person who wants to know which bill this is. */
function monthName(period: string): string {
  const [y, m] = period.split("-");
  const months = [
    "January", "February", "March", "April", "May", "June",
    "July", "August", "September", "October", "November", "December",
  ];
  return `${months[Number(m) - 1] ?? m} ${y}`;
}
 
/** The month name a refusal resumes in. The arithmetic is `period.ts`'s and the
 * NAME is this file's: an email says "January 2027" where an error body says
 * `2027-01-01`, and the difference is the audience rather than the rule. */
function resumesOn(period: string): string {
  return monthName(nextPeriod(period));
}
 
/** What an organisation's admins are told when usage crosses a threshold
 * (FR-RTL-07).
 *
 * NO SECRET, NO KEY, NO MESSAGE TEXT. THE MAIL-TRANSPORT CHAPTER established that this is
 * verified by reading what the mail server received rather than by asserting on
 * the call, and the same test shape applies here.
 *
 * AT 100% WITH NO HARD CAP, IT SAYS NOTHING WAS REFUSED. An email that threatens
 * a suspension which will not happen is worse than no email — it teaches the
 * reader that the warnings are noise, which is the one thing a warning cannot
 * afford. */
export function quotaThreshold(facts: CrossingFacts): Mail {
  const noun = NOUN[facts.dimension] ?? facts.dimension;
  const subject =
    `Relay: ${facts.environmentName} has used ${facts.threshold}% of its ` +
    `monthly ${noun} quota`;
 
  const consequence =
    facts.threshold < 100
      ? "Nothing has been refused. This is a warning so the month does not end in a surprise."
      : facts.hardCapInForce
        ? `Sends are now being refused with \`quota_exceeded\`. They resume in ${resumesOn(facts.period)}, or as soon as the quota is raised.`
        : "Nothing has been refused: this environment has no hard cap, only the threshold you asked to be told about.";
 
  const text = [
    `${facts.environmentName} has used ${facts.usageAtCrossing} of ${facts.quota} ${noun} for ${monthName(facts.period)} — ${facts.threshold}%.`,
    "",
    consequence,
    "",
    "Usage resets at the start of the next calendar month.",
  ].join("\n");
 
  return { subject, text };
}

import type { Logger } from "@relay/service-kit";
 
import type { Db } from "../db/client";
import {
  drainQuotaNotifications,
  organisationRecipients,
  type QuotaNotificationRow,
} from "../db/repository";
import type { Mailer } from "../notifications/mailer";
import { quotaThreshold } from "./quota-email";
 
// THE OUTBOX PATTERN, A FOURTH TIME — after the outbox chapter's events, the webhook
// dispatcher chapter's deliveries and the mail-transport chapter's disablement
// emails. Same shape on purpose: a table whose
// claim predicate starts null, a loop that reads it, a side effect, and no state
// shared with the request path.
//
// Four concrete tables that look alike is a pattern; one abstract table serving
// four purposes is a framework. The number is worth saying out loud, because a
// reader who has now seen it three times deserves to be told the repetition is
// deliberate rather than an oversight nobody got round to.
 
const BATCH_SIZE = 100;
const IDLE_INTERVAL_MS = 5_000;
 
export interface QuotaRelay {
  start(): void;
  stop(): Promise<void>;
  /** One pass — the same code path `start` runs, so nothing here is proven only
   * about a loop that tests never enter. */
  drainOnce(): Promise<number>;
}
 
export function createQuotaRelay({
  db,
  mailer,
  logger,
  batchSize = BATCH_SIZE,
  intervalMs = IDLE_INTERVAL_MS,
}: {
  db: Db;
  mailer: Mailer;
  logger: Logger;
  batchSize?: number;
  intervalMs?: number;
}): QuotaRelay {
  let running = false;
  let loop: Promise<void> = Promise.resolve();
 
  async function deliver(row: QuotaNotificationRow): Promise<void> {
    const recipients = await organisationRecipients(db, row.organisationId);
 
    if (recipients.length === 0) {
      // The same real branch the mail-transport chapter met: `humans.email` is nullable, so an
      // organisation whose every member is unaddressable is a state the schema
      // permits. The row is marked delivered because there is no address to
      // retry to, and leaving it claimable would mean reclaiming the same
      // undeliverable row every five seconds for ever. The log line is what
      // replaces the email — the obligation is discharged as far as it can be,
      // and the fact that it could not be met is recorded rather than swallowed.
      logger.log("error", "quotas.unaddressable", {
        organisation_id: row.organisationId,
        notification_id: row.id,
        detail:
          "quota threshold could not be notified: no member has an email address",
      });
      return;
    }
 
    const mail = quotaThreshold({
      environmentName: row.environmentName,
      period: row.period,
      dimension: row.dimension,
      threshold: row.threshold,
      quota: row.quota,
      usageAtCrossing: row.usageAtCrossing,
      hardCapInForce: row.hardCapInForce,
    });
 
    // One message per recipient, never one message with several addresses on
    // it: a customer's colleagues' addresses are that customer's data, and a
    // header every recipient can read is a disclosure nobody asked for.
    for (const to of recipients) {
      await mailer.send(to, mail);
    }
    logger.log("info", "quotas.notified", {
      notification_id: row.id,
      recipients: recipients.length,
    });
  }
 
  async function drainOnce(): Promise<number> {
    return drainQuotaNotifications(db, batchSize, deliver, (row, error) => {
      // One row's failure, one line, and the batch keeps going. A mail server
      // that is down produces one of these per claimed row and then a drain of
      // zero, which the loop treats as idle — correct, because there is nothing
      // this process can do but wait.
      logger.log("error", "quotas.send_failed", {
        notification_id: row.id,
        error: String(error),
      });
    });
  }
 
  async function run(): Promise<void> {
    while (running) {
      try {
        const sent = await drainOnce();
        if (sent > 0) {
          // A count and an id. Never an address (NFR-SEC-06).
          logger.log("info", "quotas.drained", { count: sent });
          continue;
        }
      } catch (error) {
        // A mail server that is down lands here. Rows stay claimable and the
        // next pass tries again, which is the whole reason this is a table
        // rather than a call.
        logger.log("error", "quotas.drain_failed", { error: String(error) });
      }
      await new Promise((resolve) => setTimeout(resolve, intervalMs));
    }
  }
 
  return {
    start() {
      if (running) return;
      running = true;
      loop = run();
    },
    async stop() {
      running = false;
      await loop;
    },
    drainOnce,
  };
}

import { Inject, Injectable, Module, type OnModuleDestroy } from "@nestjs/common";
 
import { createLogger } from "@relay/service-kit";
 
import { createDb, createPool, type Db } from "../db/client";
import { createMailer } from "../notifications/mailer";
import { createQuotaRelay, type QuotaRelay } from "./quota-relay";
 
// The quota relay's home. Same shape as the notification
// module's, which is the same shape as the outbox module's: a loop that reads a
// table, does a side effect, and shares no state with the request path.
 
export const QUOTA_RELAY = "QUOTA_RELAY";
 
/** Off for the suites that want a quiet database, on everywhere else — the same
 * switch and the same reasoning as `RELAY_NOTIFICATION_RELAY`. Feature 030's R39
 * found nine suites booting `AppModule` with every relay defaulting on, so the
 * three lane configs that carry the other flags carry this one too. */
export function quotaRelayEnabled(): boolean {
  return (process.env.RELAY_QUOTA_RELAY ?? "on").toLowerCase() !== "off";
}
 
@Injectable()
export class QuotaRelayService implements OnModuleDestroy {
  constructor(@Inject(QUOTA_RELAY) private readonly relay: QuotaRelay) {}
 
  start(): void {
    if (quotaRelayEnabled()) this.relay.start();
  }
 
  async onModuleDestroy(): Promise<void> {
    await this.relay.stop();
  }
}
 
@Module({
  providers: [
    {
      provide: QUOTA_RELAY,
      useFactory: (): QuotaRelay =>
        createQuotaRelay({
          db: createDb(createPool()) as Db,
          mailer: createMailer(),
          logger: createLogger("quotas"),
        }),
    },
    QuotaRelayService,
  ],
  exports: [QUOTA_RELAY, QuotaRelayService],
})
export class QuotasModule {}

@@ -81,71 +81,98 @@ BEGIN
   -- so a table with no `id` raises `record "old" has no field "id"` — from inside the
   -- refusal path, replacing the diagnosis with a message about the diagnosis.
   --
   -- Two tables carried an `id` when this was written and the third does not:
   -- `read_positions` is keyed `(channel_id, user_id)` because that is what a read
   -- position is. `to_jsonb` turns the row into a document first, so the lookup is a
   -- key that may be absent rather than a field that must exist, and the whole row is
   -- the fallback — which is more useful anyway for a table whose identity is a pair.
   RAISE EXCEPTION
     'global-operation guard: this statement modified sentinel row %.% (key %), which belongs to no test%',
     TG_TABLE_SCHEMA, TG_TABLE_NAME,
     COALESCE(to_jsonb(OLD) ->> 'id', to_jsonb(OLD)::text),
     COALESCE(' — the bait planted by ' || who, '');
 END $$;
 
 -- One trigger per table carrying environment_id, firing only for a sentinel's
 -- rows. Not `outbox`: it has no environment_id because it is platform
 -- bookkeeping, so its bait is protected by the reader mechanism only. A stated
 -- gap rather than an oversight (data-model.md).
 --
 -- THIS ARRAY IS NOT A COUNT, AND THAT IS DELIBERATE. Every table that carries
 -- `environment_id` joins it IN THE CHAPTER THAT CREATES THE TABLE, together with
 -- the sentinel row that makes the trigger's WHEN clause match and the case in
 -- `guard.itest.ts` that drives it. Naming a number here — "five tables", "nine
 -- tables" — would be a fact about the chapter that wrote the number, and every
 -- later chapter would have to remember to change it. Nothing checks a comment.
 --
 -- AND BEING IN THIS ARRAY IS NOT BEING WATCHED. The trigger fires only when
 -- `__is_sentinel(OLD.environment_id)` is true, which needs a sentinel row sitting
 -- in the table. A name added here without bait planted in `sentinel.ts` installs
--- a trigger that can never match, and it reads exactly like protection. That is
--- why the three go together: the name, the bait, and the case that turns red when
--- the name is removed.
+-- a trigger that can never match, and it reads exactly like protection. So the
+-- three go together: the name, the bait, and the case.
+--
+-- AND FOR A LONG TIME ONLY TWO OF THE THREE WERE CHECKED. `guard.itest.ts` compares
+-- this array against its own `SHAPES` and against `pg_trigger`, both directions each
+-- — and asserted nothing about the bait, because every case in it plants its own row.
+-- Deleting an insert from `plant()` left that suite entirely green. The quota chapter
+-- added the third assertion: `plant()` must leave a row in every table named here,
+-- asked of the database rather than of `sentinel.ts`'s source.
 --
 -- `members` IS THE COUNTER-EXAMPLE AND BELONGS NOWHERE NEAR THIS LIST. It has no
 -- `environment_id` — the catalogue classifies it `hop`, reaching the environment
 -- through `channels` — so `OLD.environment_id` would not compile in the WHEN
 -- clause. The rule is the column, not the intuition that a table "feels" tenant.
 DO $$
 DECLARE
   t text;
 BEGIN
   FOREACH t IN ARRAY ARRAY[
     -- The instruments chapter's two. Both carry `environment_id`, both hold bait
     -- planted by `sentinelFor`, and `guard.itest.ts` drives each one.
     'channels',
     'users',
     -- THIS CHAPTER'S, AND IT ARRIVES WITH THE TABLE. `read_positions` carries
     -- `environment_id` although `channel_id` already determines it, precisely so this
     -- trigger can exist — a table without the column is a table the guard cannot
     -- refuse a cross-environment delete on.
     --
     -- AND IT HAS NO `id`, which is the case the refusal message was changed for: it
     -- interpolates `coalesce(to_jsonb(OLD) ->> 'id', to_jsonb(OLD)::text)` rather than
     -- `OLD.id`, so a table keyed on `(channel_id, user_id)` still names the row it
     -- refused.
     --
     -- `members` REMAINS THE COUNTER-EXAMPLE. It is per-member state too and it is
     -- deliberately absent: no `environment_id`, so the catalogue calls it `hop` and
     -- `OLD.environment_id` would not compile in the WHEN clause above. The rule is the
     -- column, not the intuition that a table feels tenant-scoped.
-    'read_positions'
+    'read_positions',
+    -- THE QUOTA CHAPTER'S THREE, AND ALL THREE ARRIVE WITH THE TABLES. Every one
+    -- carries `environment_id` as its first primary-key column, which is the rule this
+    -- array follows — the column, not the intuition.
+    --
+    -- THEY ARE THE FIRST GUARDED TABLES WHOSE ROWS ARE MONEY. A cross-environment
+    -- DELETE on `channels` loses somebody's messages; one on `usage_periods` loses the
+    -- count a customer is billed against, and the platform cannot tell afterwards
+    -- whether the month was quiet or the row was dropped. `usage_active_users` is the
+    -- same fact one dimension over, and `quota_notifications` is the record that a
+    -- customer was warned — deleting it makes the platform willing to warn them twice
+    -- or, if `delivered_at` was set, not at all.
+    --
+    -- TWO OF THEM HAVE NO `id`, which is the case `read_positions` above changed the
+    -- refusal message for: `usage_periods` is keyed `(environment_id, period)` and
+    -- `usage_active_users` on a triple. The message interpolates
+    -- `coalesce(to_jsonb(OLD) ->> 'id', to_jsonb(OLD)::text)`, so both still name the
+    -- row they refused. `quota_notifications` does have one, and it is listed beside
+    -- them rather than apart, because the guard's rule has never been about the key.
+    'usage_periods',
+    'usage_active_users',
+    'quota_notifications'
   ] LOOP
     EXECUTE format('DROP TRIGGER IF EXISTS __sentinel_guard_%1$s ON %1$I', t);
     EXECUTE format(
       'CREATE TRIGGER __sentinel_guard_%1$s
          BEFORE UPDATE OR DELETE ON %1$I FOR EACH ROW
          WHEN (__is_sentinel(OLD.environment_id))
          EXECUTE FUNCTION __sentinel_guard()', t);
   END LOOP;
 END $$;

@@ -51,12 +51,23 @@ export interface Sentinel {
   /** GUARD BAIT, not drain bait. A trigger sits on `users` and `channels`, and its
    * WHEN clause tests `__is_sentinel(OLD.environment_id)` — which needs a row IN the
    * table to have anything to test. Without these two the triggers install, report
    * as installed, and can never match. See `sentinel.sql`. */
   userId: string;
   channelId: string;
+  /** THE QUOTA CHAPTER'S BAIT, and its own three. `usage_periods`,
+   * `usage_active_users` and `quota_notifications` all carry `environment_id` and all
+   * three joined the trigger array, so all three need a row for the WHEN clause to
+   * have something to test.
+   *
+   * `quotaPeriod` is a FIXED month rather than the current one, because two of the
+   * three tables are keyed on it and a bait row whose key moved at midnight on the
+   * first would be a fixture that fails one day in thirty. It is far enough in the
+   * past that no product code will ever write the same key. */
+  quotaPeriod: string;
+  quotaNotificationId: string;
   /** `__sentinel__:<owner>`, on every row, so a failure says whose it is. */
   name: string;
 }
 
 /** A v4-shaped uuid derived from a string. Deterministic, so a file's sentinel is
  * the same on every run and the delete-then-insert in `plant` is exact. */
@@ -77,12 +88,14 @@ export function sentinelFor(owner: string): Sentinel {
     organisationId: id("organisation"),
     humanId: id("human"),
     applicationId: id("application"),
     environmentId: id("environment"),
     userId: id("user"),
     channelId: id("channel"),
+    quotaPeriod: "1999-01-01",
+    quotaNotificationId: id("quota-notification"),
     name: `__sentinel__:${owner}`,
   };
 }
 
 /** The shared sentinel this feature does NOT have, kept as a named export so a
  * reader looking for one finds this comment instead. */
@@ -130,12 +143,17 @@ export async function plant(
   // `read_positions` references BOTH `channels` and `users`, which is why the note
   // the instruments chapter left here said the order was not arbitrary.
   // The subject the plant below writes, not the one it used to: a cleanup keyed on
   // a stale subject leaves every row it was meant to remove.
   await q(`DELETE FROM outbox         WHERE subject = $1`, [`events.${s.name}.bait`]);
   await q(`DELETE FROM read_positions WHERE environment_id = $1`, [s.environmentId]);
+  // The quota chapter's three, and they come before `users` for the reason the note
+  // above gives: `usage_active_users` references it.
+  await q(`DELETE FROM quota_notifications WHERE environment_id = $1`, [s.environmentId]);
+  await q(`DELETE FROM usage_active_users  WHERE environment_id = $1`, [s.environmentId]);
+  await q(`DELETE FROM usage_periods       WHERE environment_id = $1`, [s.environmentId]);
   await q(`DELETE FROM channels       WHERE environment_id = $1`, [s.environmentId]);
   await q(`DELETE FROM users          WHERE environment_id = $1`, [s.environmentId]);
 
   // Register before inserting bait: the trigger's WHEN clause tests membership,
   // so an unregistered sentinel is unguarded bait.
   await q(
@@ -199,12 +217,64 @@ export async function plant(
   await q(
     `INSERT INTO read_positions (environment_id, channel_id, user_id, sequence)
      VALUES ($1, $2, $3, 0) ON CONFLICT (channel_id, user_id) DO NOTHING`,
     [s.environmentId, s.channelId, s.userId],
   );
 
+  // THE QUOTA CHAPTER'S GUARD BAIT, one row per table it added to the array. Same
+  // rule as `read_positions` above: a name in that array with no row behind it
+  // installs a trigger that can never match, and reads as protection.
+  //
+  // `usage_active_users` reuses the sentinel's own user rather than minting one — the
+  // row only has to exist — and `quota_notifications` reuses the organisation, which
+  // it references and the sentinel already owns.
+  await q(
+    `INSERT INTO usage_periods (environment_id, period, messages_sent)
+     VALUES ($1, $2, 0) ON CONFLICT (environment_id, period) DO NOTHING`,
+    [s.environmentId, s.quotaPeriod],
+  );
+  await q(
+    `INSERT INTO usage_active_users (environment_id, period, user_id)
+     VALUES ($1, $2, $3) ON CONFLICT (environment_id, period, user_id) DO NOTHING`,
+    [s.environmentId, s.quotaPeriod, s.userId],
+  );
+  // ALREADY DELIVERED, AND THAT IS THE FOURTH MEASUREMENT OF ONE LAW. The
+  // disablement notifications and the webhook deliveries both had to concede the
+  // same point: **bait may be claimable only where draining it is DATABASE work.**
+  // A sweep or a publish qualifies; anything that does per-row I/O does not.
+  //
+  // `drainQuotaNotifications` claims on `delivered_at IS NULL` and then calls
+  // `deliver(row)` — a mail send. So an undelivered bait row is claimed by every
+  // quota relay in the lane.
+  //
+  // AND HERE THE LAW ARRIVES WITH TEETH RATHER THAN WITH LATENCY. The three earlier
+  // instances cost seconds; this one is a hard failure, because the quota chapter
+  // also put `quota_notifications` under the guard. The drain claims the sentinel's
+  // row on a connection carrying no exemption, the trigger refuses the UPDATE, and
+  // the transaction is poisoned — `25P02 in_failed_sql_transaction` on the next
+  // statement, six tests red, and the message names neither the bait nor the guard.
+  //
+  // Delivered two hours ago, so the row is still a row a global count would see and
+  // is outside every claim window.
+  //
+  // `DO UPDATE`, NOT `DO NOTHING`, AND THAT IS NOT TIDINESS. A sentinel's ids are
+  // derived from its owner, so this row's id is the same on every run for ever — and
+  // `DO NOTHING` would mean the state the row was FIRST planted with is the state it
+  // keeps. A lane that ran this fixture once before the line above said
+  // `delivered_at` holds an undelivered bait row that no later run can repair, and the
+  // failure it causes is the one described above: refused, poisoned, six red. Here the
+  // row's STATE is part of the fixture's contract and not merely its existence.
+  await q(
+    `INSERT INTO quota_notifications
+       (id, environment_id, organisation_id, period, dimension, threshold,
+        quota, usage_at_crossing, delivered_at)
+     VALUES ($1, $2, $3, $4, 'messages', 50, 1, 1, now() - interval '2 hours')
+     ON CONFLICT (id) DO UPDATE SET delivered_at = EXCLUDED.delivered_at`,
+    [s.quotaNotificationId, s.environmentId, s.organisationId, s.quotaPeriod],
+  );
+
   // DRAIN BAIT: unpublished events. `outbox` carries no environment_id — it is
   // platform bookkeeping — so the subject is what identifies these, and it is also
   // why the trigger cannot guard them (data-model.md). The count is `BAIT_ROWS` and
   // not one, because a single row cannot tell a batch that ignored its limit from
   // one that honoured it.
   //

import { readFileSync } from "node:fs";
import { join } from "node:path";
 
import pg from "pg";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
import { databaseUrl } from "./db-url.js";
import { sentinelFor, type Sentinel } from "./sentinel.js";
 
// THE GUARD, DRIVEN ONE TABLE AT A TIME.
//
// `sentinel.sql` names the guarded tables in an array and installs a trigger for
// each. Installing a trigger is cheap and silent, and it is not the same as being
// watched: the WHEN clause tests `__is_sentinel(OLD.environment_id)`, so a table
// with no sentinel row in it installs a trigger that can never match. A passing
// lane looks identical either way.
//
// So every name in that array gets a case here, and the case does the thing the
// guard exists to forbid — an UPDATE and a DELETE with no tenant predicate, over a
// row belonging to another file's sentinel. Remove a name from the array and its
// two cases go red. That is the property; the array on its own asserts nothing.
//
// WHY THIS FILE IS NOT ON THE EXEMPT LIST. It is not performing a global operation
// — it is provoking one and expecting to be refused. An exemption would switch the
// trigger off and every expectation below would invert without the file changing.
 
const url = new URL(databaseUrl());
if (!["localhost", "127.0.0.1"].includes(url.hostname)) {
  throw new Error(
    `integration tests refuse non-local databases (got host "${url.hostname}")`,
  );
}
 
// A VICTIM THAT IS NOT THIS FILE'S OWN SENTINEL. The trigger fires for any
// registered sentinel's rows, and using our own would still pass — but it would
// pass for a weaker reason, since a suite mutating its own bait is a mistake the
// guard is not there to catch. Naming somebody else's makes the refusal the point.
const VICTIM = sentinelFor("packages/test-harness/src/guard.itest.ts#victim");
 
// A NEIGHBOUR THAT IS NOT A SENTINEL, holding a row in every guarded table. The
// permissive half of each pair needs a row the trigger DECLINES to refuse, and a
// predicate matching nothing is not that: `BEFORE UPDATE FOR EACH ROW` never fires
// over zero rows, so an update that matches none passes under a trigger refusing
// everything. Measured, not reasoned about — the first version of this file scoped
// its permitted update to an id nothing held, and it stayed green with the WHEN
// clause deleted from the trigger.
const NEIGHBOUR = sentinelFor("packages/test-harness/src/guard.itest.ts#neighbour");
 
// READ OUT OF `sentinel.sql`, not restated. A second copy of the list would agree
// with the first by somebody remembering, and the cases below are generated from
// it — so a table added to the array arrives here with its cases already written,
// and one removed takes its cases with it. Parsing beats copying wherever the
// parse can fail loudly, which is what the length assertion is for.
const GUARDED: readonly string[] = (() => {
  const sql = readFileSync(join(import.meta.dirname, "sentinel.sql"), "utf8");
  const block = /FOREACH t IN ARRAY ARRAY\[([\s\S]*?)\] LOOP/.exec(sql);
  if (block === null) {
    throw new Error(
      "guard.itest.ts cannot find sentinel.sql's table array — the shape it parses changed",
    );
  }
  // COMMENT LINES FIRST, and this is not tidiness. The array's own comments contain
  // quoted strings — the chapter that added `read_positions` explained the refusal
  // message's `to_jsonb(OLD) ->> 'id'` right there — and a naive scan for quoted
  // lower-case words took `id` for a table. The suite then ran five cases against a
  // table called `id`, all of which failed, and the report named a table nobody had
  // written. A parser over a language it does not parse has to at least know what a
  // comment is.
  const body = block[1]!
    .split("\n")
    .map((l) => l.replace(/--.*$/, ""))
    .join("\n");
  const names = [...body.matchAll(/'([a-z_]+)'/g)].map((m) => m[1]!);
  if (names.length === 0) {
    throw new Error("sentinel.sql's table array parsed empty, which would make every case below vacuous");
  }
  return names;
})();
 
/** ONE ENTRY PER GUARDED TABLE: how to plant a row in it, and a column an UPDATE can
 * touch. Asserted against `GUARDED` in both directions below.
 *
 * WHY THIS IS A TABLE AND NOT A CONVENTION. The generated cases used
 * `SET metadata = '{}'::jsonb` for every guarded table, which worked for exactly as
 * long as every guarded table had a `metadata` column. `read_positions` does not, and
 * the report said `column "metadata" of relation "read_positions" does not exist` —
 * five cases failing on a column, in a suite about a trigger.
 *
 * And the victim fixture had the same shape problem from the other side: it planted a
 * `users` row and a `channels` row because those were the two tables, so adding a
 * third left the trigger with nothing to match and an unscoped DELETE went through.
 * A guard test whose fixture does not cover a guarded table reports that table as
 * unguarded and is right.
 */
interface Shape {
  plant: string;
  values: (s: Sentinel) => unknown[];
  /** A no-op-safe assignment for the refusal cases: they must reach the trigger, and
   * what they set is irrelevant. */
  touch: string;
  /** A detectable assignment for the exemption case, with the read that checks it. */
  mark: string;
  read: string;
  marked: (n: number) => unknown;
}
 
const SHAPES: Readonly<Record<string, Shape>> = {
  users: {
    plant: `INSERT INTO users (id, environment_id, external_id, display_name)
            VALUES ($1, $2, $3, $3) ON CONFLICT (id) DO NOTHING`,
    values: (s) => [s.userId, s.environmentId, s.name],
    touch: `metadata = '{}'::jsonb`,
    mark: `metadata = $1::jsonb`,
    read: `SELECT metadata AS v FROM users WHERE environment_id = $1`,
    marked: (n) => ({ "guard-probe": String(n) }),
  },
  channels: {
    plant: `INSERT INTO channels (id, environment_id, external_id, type, name)
            VALUES ($1, $2, $3, 'private', $3) ON CONFLICT (id) DO NOTHING`,
    values: (s) => [s.channelId, s.environmentId, s.name],
    touch: `metadata = '{}'::jsonb`,
    mark: `metadata = $1::jsonb`,
    read: `SELECT metadata AS v FROM channels WHERE environment_id = $1`,
    marked: (n) => ({ "guard-probe": String(n) }),
  },
  read_positions: {
    plant: `INSERT INTO read_positions (environment_id, channel_id, user_id, sequence)
            VALUES ($1, $2, $3, 0) ON CONFLICT (channel_id, user_id) DO NOTHING`,
    values: (s) => [s.environmentId, s.channelId, s.userId],
    // `sequence` rather than `metadata`, because this table has no metadata — and
    // `updated_at` would be touched by a trigger on some tables, so a column nothing
    // else writes is the honest choice.
    touch: `sequence = sequence`,
    // NOT `metadata`, AND NOT KEYED ON `id`. This table has neither. The mark is a
    // sequence value nothing else writes and the read is scoped by environment, which
    // works for all three tables and does not assume a surrogate key — the absence of
    // one being the reason this table joined the guard with a message change.
    mark: `sequence = $1`,
    read: `SELECT sequence AS v FROM read_positions WHERE environment_id = $1`,
    marked: (n) => n,
  },
};
 
let admin: pg.Client;
let plain: pg.Client;
 
beforeAll(async () => {
  // Two clients on purpose. `admin` carries the exemption and plants the victim;
  // `plain` carries none and is what the expectations run through. One client
  // doing both would mean every refusal below depended on a SET landing on the
  // connection the query happened to get.
  const exempt = new URL(url.toString());
  exempt.searchParams.set("options", "-c relay.allow_global=on");
  admin = new pg.Client({ connectionString: exempt.toString() });
  plain = new pg.Client({ connectionString: url.toString() });
  await admin.connect();
  await plain.connect();
 
  await admin.query(
    `INSERT INTO __sentinel_environments (environment_id, owner) VALUES ($1, $2)
     ON CONFLICT (environment_id) DO UPDATE SET owner = EXCLUDED.owner`,
    [VICTIM.environmentId, VICTIM.owner],
  );
  await admin.query(
    `INSERT INTO organisations (id, name) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING`,
    [VICTIM.organisationId, VICTIM.name],
  );
  await admin.query(
    `INSERT INTO applications (id, organisation_id, name)
     VALUES ($1, $2, $3) ON CONFLICT (id) DO NOTHING`,
    [VICTIM.applicationId, VICTIM.organisationId, VICTIM.name],
  );
  await admin.query(
    `INSERT INTO environments (id, application_id, kind, signing_secret)
     VALUES ($1, $2, 'development', $3) ON CONFLICT (id) DO NOTHING`,
    [VICTIM.environmentId, VICTIM.applicationId, `sentinel-not-a-secret-${VICTIM.environmentId}`],
  );
  // THROUGH `SHAPES`, so a table added to the guard's array is planted here by
  // construction rather than by somebody remembering.
  for (const table of GUARDED) {
    const shape = SHAPES[table]!;
    await admin.query(shape.plant, shape.values(VICTIM));
  }
 
  // The neighbour's tenancy, DELIBERATELY NOT registered in
  // `__sentinel_environments` — that omission is the whole point of these rows.
  await admin.query(
    `INSERT INTO organisations (id, name) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING`,
    [NEIGHBOUR.organisationId, NEIGHBOUR.name],
  );
  await admin.query(
    `INSERT INTO applications (id, organisation_id, name)
     VALUES ($1, $2, $3) ON CONFLICT (id) DO NOTHING`,
    [NEIGHBOUR.applicationId, NEIGHBOUR.organisationId, NEIGHBOUR.name],
  );
  await admin.query(
    `INSERT INTO environments (id, application_id, kind, signing_secret)
     VALUES ($1, $2, 'development', $3) ON CONFLICT (id) DO NOTHING`,
    [NEIGHBOUR.environmentId, NEIGHBOUR.applicationId, `sentinel-not-a-secret-${NEIGHBOUR.environmentId}`],
  );
  for (const table of GUARDED) {
    const shape = SHAPES[table]!;
    await admin.query(shape.plant, shape.values(NEIGHBOUR));
  }
});
 
afterAll(async () => {
  // Children before parents, and through `admin` because deleting a sentinel row
  // is exactly what the guard forbids.
  for (const s of [VICTIM, NEIGHBOUR]) {
    // REVERSED, so children go before parents: `read_positions` references both of
    // the others, and the array is written creation-order.
    for (const t of [...GUARDED].reverse()) {
      await admin.query(`DELETE FROM ${t} WHERE environment_id = $1`, [s.environmentId]);
    }
    await admin.query(`DELETE FROM environments WHERE id = $1`, [s.environmentId]);
    await admin.query(`DELETE FROM applications WHERE id = $1`, [s.applicationId]);
    await admin.query(`DELETE FROM organisations WHERE id = $1`, [s.organisationId]);
    await admin.query(`DELETE FROM __sentinel_environments WHERE environment_id = $1`, [s.environmentId]);
  }
  await admin.end();
  await plain.end();
});
 
describe("the guard refuses an unscoped mutation of a sentinel row", () => {
  it("has a shape for every guarded table and no shape for anything else", () => {
    // THE COMMENT ON `SHAPES` CLAIMED THIS AND THE SUITE DID NOT MAKE IT. Every use
    // site writes `SHAPES[table]!`, so a guarded table with no entry throws
    // `Cannot read properties of undefined` from inside a fixture — a message about
    // JavaScript, in a suite about a trigger, naming no table.
    //
    // BOTH DIRECTIONS. A missing entry is the failure above; an extra one is a table
    // that used to be guarded and is not, which leaves a fixture planting rows the
    // trigger no longer protects and nothing anywhere going red.
    expect(Object.keys(SHAPES).sort()).toEqual([...GUARDED].sort());
  });
 
  it("installs one trigger per name in sentinel.sql's array, and no more", async () => {
    // BOTH DIRECTIONS. A name in the array with no trigger means the DO block
    // failed silently; a trigger with no name means a stale install survived a
    // removal, and `DROP TRIGGER IF EXISTS` only covers names still in the array.
    const { rows } = await plain.query<{ tgname: string; relname: string }>(
      `SELECT t.tgname, c.relname FROM pg_trigger t
         JOIN pg_class c ON c.oid = t.tgrelid
        WHERE t.tgname LIKE '__sentinel_guard_%' AND NOT t.tgisinternal`,
    );
    expect(rows.map((r) => r.relname).sort()).toEqual([...GUARDED].sort());
  });
 
  for (const table of GUARDED) {
    it(`refuses an unscoped UPDATE on ${table}`, async () => {
      // No WHERE at all — the shape a global sweep has. The refusal must name the
      // table and the owner, because a bare failure sends the next reader to the
      // wrong file.
      await expect(
        plain.query(`UPDATE ${table} SET ${SHAPES[table]!.touch}`),
      ).rejects.toThrow(/global-operation guard/);
    });
 
    it(`refuses an unscoped DELETE on ${table}, naming SOME owner`, async () => {
      // Asserted on the MESSAGE, not just on rejection: the diagnosis is the
      // feature. A refusal that does not say whose bait it was leaves the reader
      // grepping for a uuid.
      //
      // "SOME owner", NOT OURS, AND THAT DISTINCTION COST A LANE RUN. An unscoped
      // DELETE trips on whichever sentinel row Postgres reaches first, and which
      // one that is depends on who else is in the database. Pinned to this file's
      // own owner it passed alone and failed in the coverage lane, where the
      // refusal named `services/api/src/isolation/targets.itest.ts` instead —
      // an assertion scoped wider than the thing it tests, failing for somebody
      // else's reason. The case below pins the owner properly.
      await expect(plain.query(`DELETE FROM ${table}`)).rejects.toThrow(
        new RegExp(`global-operation guard.*${table}.*the bait planted by \\S+\\.itest\\.ts`, "s"),
      );
    });
 
    it(`names THIS file's owner when the row can only be ours`, async () => {
      // Scoped to one environment and still refused, because scoping to a TENANT
      // is not the same as being allowed to mutate a sentinel. This is the only
      // shape that can assert the owner, since it is the only shape whose victim
      // is not decided by row order.
      await expect(
        plain.query(`DELETE FROM ${table} WHERE environment_id = $1`, [
          VICTIM.environmentId,
        ]),
      ).rejects.toThrow(
        new RegExp(
          `global-operation guard.*${table}.*${VICTIM.owner.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}`,
          "s",
        ),
      );
    });
 
    it(`lets an EXEMPT connection's UPDATE on ${table} actually land`, async () => {
      // THE EXEMPTION PATH, and the assertion is the VALUE READ BACK — not the
      // row count, and not the absence of a throw.
      //
      // A BEFORE UPDATE trigger returning OLD does not permit the update: it
      // replaces it with a write of the old values. `rowCount` is 1 either way and
      // nothing throws, so a guard that reverts every exempt write looks exactly
      // like one that permits them. The symptom shows up somewhere else entirely —
      // an exempt sweep that disables the same rows on every pass and never runs
      // out — which is a long way from the file holding the fault.
      const shape = SHAPES[table]!;
      const n = 1 + GUARDED.indexOf(table);
      const value = shape.marked(n);
      await admin.query(
        `UPDATE ${table} SET ${shape.mark} WHERE environment_id = $2`,
        [typeof value === "object" ? JSON.stringify(value) : value, VICTIM.environmentId],
      );
      const { rows } = await admin.query<{ v: unknown }>(shape.read, [VICTIM.environmentId]);
      expect(rows[0]?.v, `the exempt write to ${table} was reverted`)
        .toEqual(typeof value === "object" ? value : String(value));
    });
 
    it(`permits a scoped UPDATE on ${table} that hits a non-sentinel row`, async () => {
      // THE OTHER HALF, and without it every expectation above passes under a
      // trigger that refuses everything.
      //
      // `rowCount` IS THE ASSERTION, not the absence of a throw. An update matching
      // zero rows also does not throw, and that is how this case was wrong the
      // first time: it named an environment nothing held, so it stayed green with
      // the trigger's WHEN clause deleted. One row has to change hands for the
      // permission to have been exercised.
      const res = await plain.query(
        `UPDATE ${table} SET ${SHAPES[table]!.touch} WHERE environment_id = $1`,
        [NEIGHBOUR.environmentId],
      );
      expect(res.rowCount, `no ${table} row in the neighbour environment to permit`)
        .toBe(1);
    });
  }
});

@@ -2,13 +2,13 @@ import { readFileSync } from "node:fs";
 import { join } from "node:path";
 
 import pg from "pg";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
 import { databaseUrl } from "./db-url.js";
-import { sentinelFor, type Sentinel } from "./sentinel.js";
+import { plant, sentinelFor, type Sentinel } from "./sentinel.js";
 
 // THE GUARD, DRIVEN ONE TABLE AT A TIME.
 //
 // `sentinel.sql` names the guarded tables in an array and installs a trigger for
 // each. Installing a trigger is cheap and silent, and it is not the same as being
 // watched: the WHEN clause tests `__is_sentinel(OLD.environment_id)`, so a table
@@ -43,12 +43,18 @@ const VICTIM = sentinelFor("packages/test-harness/src/guard.itest.ts#victim");
 // over zero rows, so an update that matches none passes under a trigger refusing
 // everything. Measured, not reasoned about — the first version of this file scoped
 // its permitted update to an id nothing held, and it stayed green with the WHEN
 // clause deleted from the trigger.
 const NEIGHBOUR = sentinelFor("packages/test-harness/src/guard.itest.ts#neighbour");
 
+// A THIRD SENTINEL THAT ONLY `plant()` EVER TOUCHES. Everything else in this file
+// plants the victim's rows through `SHAPES`, which is right for the refusal cases and
+// useless for asking whether `plant()` itself covers the array — a question about the
+// function every OTHER integration suite depends on.
+const CANARY = sentinelFor("packages/test-harness/src/guard.itest.ts#canary");
+
 // READ OUT OF `sentinel.sql`, not restated. A second copy of the list would agree
 // with the first by somebody remembering, and the cases below are generated from
 // it — so a table added to the array arrives here with its cases already written,
 // and one removed takes its cases with it. Parsing beats copying wherever the
 // parse can fail loudly, which is what the length assertion is for.
 const GUARDED: readonly string[] = (() => {
@@ -136,12 +142,61 @@ const SHAPES: Readonly<Record<string, Shape>> = {
     // works for all three tables and does not assume a surrogate key — the absence of
     // one being the reason this table joined the guard with a message change.
     mark: `sequence = $1`,
     read: `SELECT sequence AS v FROM read_positions WHERE environment_id = $1`,
     marked: (n) => n,
   },
+  // THE QUOTA CHAPTER'S THREE. Two of them have no `metadata` and no `id`, which is
+  // the shape `read_positions` above already forced this table to accommodate — so
+  // these three cost three entries and no change to the mechanism, which is what a
+  // table of shapes is for.
+  //
+  // Each marks a column nothing else in the fixture writes, and each reads back scoped
+  // by environment. `usage_periods.messages_sent` is a count, so the mark is a number
+  // and the `touch` is the no-op `messages_sent = messages_sent`: the refusal cases
+  // only have to REACH the trigger, and a case that changed the count would leave the
+  // suite unable to tell a refusal from an arithmetic mistake.
+  usage_periods: {
+    plant: `INSERT INTO usage_periods (environment_id, period, messages_sent)
+            VALUES ($1, $2, 0) ON CONFLICT (environment_id, period) DO NOTHING`,
+    values: (s) => [s.environmentId, s.quotaPeriod],
+    touch: `messages_sent = messages_sent`,
+    mark: `messages_sent = $1`,
+    read: `SELECT messages_sent AS v FROM usage_periods WHERE environment_id = $1`,
+    // `bigint` comes back from `pg` as a STRING, not a number — the driver will not
+    // narrow a 64-bit integer into a float — so the expectation is the string. Found
+    // by the case failing with `expected '7' to be 7`, which is the whole reason this
+    // is a per-table function rather than one shared expectation.
+    marked: (n) => String(n),
+  },
+  usage_active_users: {
+    plant: `INSERT INTO usage_active_users (environment_id, period, user_id)
+            VALUES ($1, $2, $3)
+            ON CONFLICT (environment_id, period, user_id) DO NOTHING`,
+    values: (s) => [s.environmentId, s.quotaPeriod, s.userId],
+    // Every column of this table is either the key or `first_seen_at`, so the mark has
+    // to be the timestamp — there is nothing else to write. It takes an epoch offset so
+    // two marks in one run differ.
+    touch: `first_seen_at = first_seen_at`,
+    mark: `first_seen_at = to_timestamp($1)`,
+    read: `SELECT extract(epoch from first_seen_at)::bigint AS v
+             FROM usage_active_users WHERE environment_id = $1`,
+    marked: (n) => String(n),
+  },
+  quota_notifications: {
+    plant: `INSERT INTO quota_notifications
+              (id, environment_id, organisation_id, period, dimension, threshold,
+               quota, usage_at_crossing)
+            VALUES ($1, $2, $3, $4, 'messages', 50, 1, 1)
+            ON CONFLICT (id) DO NOTHING`,
+    values: (s) => [s.quotaNotificationId, s.environmentId, s.organisationId, s.quotaPeriod],
+    touch: `last_error = last_error`,
+    mark: `last_error = $1::text`,
+    read: `SELECT last_error AS v FROM quota_notifications WHERE environment_id = $1`,
+    marked: (n) => String(n),
+  },
 };
 
 let admin: pg.Client;
 let plain: pg.Client;
 
 beforeAll(async () => {
@@ -179,12 +234,16 @@ beforeAll(async () => {
   // construction rather than by somebody remembering.
   for (const table of GUARDED) {
     const shape = SHAPES[table]!;
     await admin.query(shape.plant, shape.values(VICTIM));
   }
 
+  // AND THE CANARY, THROUGH THE REAL FUNCTION. `plant()` is what `setup.ts` runs once
+  // per test file, so this is the only place in the suite where it is exercised at all.
+  await plant(admin, CANARY);
+
   // The neighbour's tenancy, DELIBERATELY NOT registered in
   // `__sentinel_environments` — that omission is the whole point of these rows.
   await admin.query(
     `INSERT INTO organisations (id, name) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING`,
     [NEIGHBOUR.organisationId, NEIGHBOUR.name],
   );
@@ -244,12 +303,47 @@ describe("the guard refuses an unscoped mutation of a sentinel row", () => {
          JOIN pg_class c ON c.oid = t.tgrelid
         WHERE t.tgname LIKE '__sentinel_guard_%' AND NOT t.tgisinternal`,
     );
     expect(rows.map((r) => r.relname).sort()).toEqual([...GUARDED].sort());
   });
 
+  it("plants a sentinel row in every guarded table, so the canary is never missing", async () => {
+    // THE THIRD DIRECTION, AND IT WAS THE ONE NOBODY ASSERTED. `sentinel.sql` says the
+    // name, the bait and the case "go together" — and two of the three were checked
+    // against each other while the bait was checked by nothing. Measured: deleting the
+    // `usage_periods` insert from `plant()` left this suite at 32 of 32 green, and
+    // deleting `read_positions`' — an older chapter's — did too.
+    //
+    // WHAT THAT COSTS IS NOT THIS SUITE. Every case below plants its OWN row through
+    // `SHAPES`, so this file is unaffected by a missing insert. What `plant()` is for is
+    // every OTHER integration suite: `setup.ts` runs it once per test FILE, and the row
+    // it leaves is what makes an unscoped `DELETE FROM usage_periods` in somebody
+    // else's suite meet a trigger at all. A name added to the array with a shape and no
+    // bait leaves that table with no canary in any lane run, and reads as protection.
+    //
+    // ASKED OF THE DATABASE, not of `sentinel.ts`'s source. A source scan would pass on
+    // an insert that runs and inserts nothing — `ON CONFLICT DO NOTHING` against a row
+    // this fixture does not own, say — which is the shape a fixture fails in.
+    // AGAINST THE CANARY, AND THE FIRST VERSION OF THIS USED THE VICTIM — which is
+    // planted through `SHAPES` in `beforeAll`, so every count was nonzero for a reason
+    // that had nothing to do with `plant()`. It stayed green with the insert deleted,
+    // twice, which is exactly the vacuity this assertion was written to end.
+    const missing: string[] = [];
+    for (const table of GUARDED) {
+      const { rows } = await admin.query<{ n: string }>(
+        `SELECT count(*)::text AS n FROM ${table} WHERE environment_id = $1`,
+        [CANARY.environmentId],
+      );
+      if (rows[0]!.n === "0") missing.push(table);
+    }
+    expect(
+      missing,
+      `guarded, and plant() leaves no row to guard: ${missing.join(", ")}`,
+    ).toEqual([]);
+  });
+
   for (const table of GUARDED) {
     it(`refuses an unscoped UPDATE on ${table}`, async () => {
       // No WHERE at all — the shape a global sweep has. The refusal must name the
       // table and the owner, because a bare failure sends the next reader to the
       // wrong file.
       await expect(

@@ -221,12 +221,28 @@
   // NOT `quota_exceeded`. That is a monthly, billable, resets-on-a-date refusal whose
   // message promises a resume date; this is a structural limit on one channel that no
   // amount of waiting changes. Same status, different fact, and a client that retries on
   // the wrong one waits for ever.
   channel_member_limit_exceeded:
     "this channel already holds the maximum number of members; remove one before adding another",
+  // THE QUOTA CHAPTER'S ONE CODE, AND THIS FILE HAS BEEN CITING IT SINCE BEFORE IT
+  // EXISTED. Two comments above argue against reusing it — the banned-user code says
+  // "the same argument this file already makes for `wrong_credential_type` and
+  // `quota_exceeded`", and `channel_member_limit_exceeded` says "NOT `quota_exceeded`.
+  // That is a monthly, billable, resets-on-a-date refusal whose message promises a
+  // resume date." Both were true and neither was checked: `quota_exceeded` was not in
+  // this object, and a registry that names a code only in prose is a registry that
+  // cannot refuse a typo of it.
+  //
+  // `402`, WHICH IS THE ONLY STATUS IN THIS FILE THAT MEANS MONEY. The rate limiter
+  // owns `429`, and a client that sleeps for `Retry-After` and retries is behaving
+  // correctly for a rate limit and wrongly for this: the month will still be exhausted
+  // in an hour. So the resume date goes in the MESSAGE, where a person reads it,
+  // rather than in a header a client acts on.
+  quota_exceeded:
+    "this environment has used its monthly quota; sends resume when the period rolls over",
     // FR-RTM-09, and the socket's half of the connection cap: an error frame carrying
     // the limit and the count, sent immediately before close 4004.
     //
     // THE FIGURES GO IN THE MESSAGE, not in payload fields. `errorFrameSchema` is a
     // `z.strictObject` of `code`, `message`, `docs_url` and an optional `field`, so two
     // new fields would mean widening a shape every error frame shares for one code's

@@ -80,12 +80,28 @@ export default tseslint.config(
       // "the state under test is one the repository is now unable to reach". Both are
       // listed by path rather than reached through a shared helper, because a helper in
       // another file names none of these specifiers and this rule sees only imports —
       // an invisible exemption is worse than a listed one.
       "services/api/src/internal/backfill.itest.ts",
       "services/api/src/messages/history.itest.ts",
+      // THE QUOTA CHAPTER'S PERIOD SUITE, and its case is the two above's in a third
+      // shape: the state under test is one the repository cannot reach. `periodOf`
+      // returns the month a timestamp falls in, and the property is that a row INSERTED
+      // under that value is FOUND by it — which needs a `usage_periods` row written
+      // directly, because every repository path that writes one derives the period from
+      // the clock and so cannot disagree with the function under test.
+      //
+      // A suite that used the repository here would be asserting that `periodOf` equals
+      // itself.
+      "services/api/src/quotas/period.itest.ts",
+      // AND THE QUOTA SUITE ITSELF, for a different reason from its sibling above.
+      // `period.itest.ts` writes a row the repository cannot; this one READS the two
+      // roll-up tables directly to check what a send left behind. Going through
+      // `usageFor` would mean asserting the roll-up against the function that reads
+      // it — the same circularity, one table over.
+      "services/api/src/quotas/quotas.itest.ts",
       // ── AND EVERY OTHER REDIS CLIENT, BY PATH, WITH THE ARGUMENT IT NEEDS ──
       //
       // The rule arrives here and TWELVE files older than it already import `ioredis`.
       // A missing exemption is not a silent one — this rule goes red on a chapter
       // nobody is editing — so all of them land in the commit that adds the rule.
       // FIVE DIFFERENT ARGUMENTS, and a blanket "the gateway's Redis files" would

@@ -29,12 +29,14 @@ export default defineConfig({
     // the list is exactly the relays that exist, because `setup.ts` refuses a name
     // no module reads.
     env: {
       RELAY_HARNESS_BAIT: "on",
       RELAY_OUTBOX_RELAY: "off",
       RELAY_EVENT_CONSUMER: "off",
+      // The quota relay, the fourth. Same reason as the other three.
+      RELAY_QUOTA_RELAY: "off",
     },
     include: ["src/**/*.itest.ts"],
     // FILES IN PARALLEL AGAIN, AND EIGHT PLACES ARE WHY IT COULD NOT BE.
     //
     // This lane ran one file at a time from the outbox chapter, for a real reason: a
     // PENDING migration lets two suites issue `CREATE TYPE` against one schema, and

import { defineConfig } from "vitest/config";
 
// The dispatcher's integration lane. Same convention 2.1
// established for the api and 2.6 for the gateway: *.itest.ts is invisible to
// the Docker-free unit include, so `pnpm test` stays runnable with no stores.
export default defineConfig({
  test: {
    include: ["src/**/*.itest.ts"],
    // A delivery to a hostile endpoint spends real time on timeouts and the
    // widening schedule. The default 5 s would fail the suite for being honest.
    testTimeout: 120_000,
    hookTimeout: 120_000,
  },
});

@@ -2,12 +2,47 @@ import { defineConfig } from "vitest/config";
 
 // The dispatcher's integration lane. Same convention 2.1
 // established for the api and 2.6 for the gateway: *.itest.ts is invisible to
 // the Docker-free unit include, so `pnpm test` stays runnable with no stores.
 export default defineConfig({
   test: {
+    // Feature 030: the global-operation guard. `globalSetup` migrates and
+    // then installs the trigger once per lane; `setupFiles` sets the
+    // exemption for files on the harness's list and, where the lane carries
+    // bait, plants it per file.
+    globalSetup: ["../../packages/test-harness/src/global-setup.ts"],
+    setupFiles: ["../../packages/test-harness/src/setup.ts"],
+    // FEATURE 030, MEASURED: nine suites in this lane import `AppModule`, and none
+    // of them set a relay flag. Each relay defaults to on when its flag is unset
+    // (`process.env.RELAY_OUTBOX_RELAY ?? "on"`), so those nine booted four
+    // background loops that sweep the whole database while every other suite's
+    // fixtures sit in it. Research R13 recorded the exposure as nil on the strength
+    // of the four suites that spawn an api CHILD and set the flags in the child's
+    // env; it did not look at the suites that boot the app in process.
+    //
+    // A relay catches and logs its own errors, so the guard's refusal inside one is
+    // a log line and a green lane. Setting the flags here makes the quiet database
+    // a property of the lane rather than a convention nobody applied.
+    // NO BAIT IN THIS LANE (feature 030, research R44). Measured: 200 bait
+    // deliveries alone, on a freshly migrated database, fail 10 of this suite's 16
+    // tests — with instance 5's fix in place. The suite waits 8 seconds for the
+    // dispatcher process to deliver its own row, and the dispatcher consumes a
+    // shared FIFO stream, so 200 jobs ahead of it exhaust the poll. Bait that fails
+    // the suite whether or not the fault is present carries no information.
+    //
+    // The exemption handling stays: the trigger is database state and outlives
+    // whichever lane installed it, so every lane pointed at that database meets it.
+    env: {
+      RELAY_HARNESS_BAIT: "on",
+      RELAY_OUTBOX_RELAY: "off",
+      RELAY_DELIVERY_RELAY: "off",
+      RELAY_NOTIFICATION_RELAY: "off",
+      RELAY_EVENT_CONSUMER: "off",
+      // The quota relay, the fourth. Same reason as the other three.
+      RELAY_QUOTA_RELAY: "off",
+    },
     include: ["src/**/*.itest.ts"],
     // A delivery to a hostile endpoint spends real time on timeouts and the
     // widening schedule. The default 5 s would fail the suite for being honest.
     testTimeout: 120_000,
     hookTimeout: 120_000,
   },

@@ -23,12 +23,31 @@ export default defineConfig({
     // then installs the trigger once per lane; `setupFiles` sets the
     // exemption for files on the harness's list and, where the lane carries
     // bait, plants it per file. This lane gets exemption
     // handling and NO bait: it holds no reader-shape fault, and planting
     // would change its workload for no return (FR-022).
     globalSetup: ["./packages/test-harness/src/global-setup.ts"],
+    // FEATURE 030, MEASURED: nine suites in this lane import `AppModule`, and none
+    // of them set a relay flag. Each relay defaults to on when its flag is unset
+    // (`process.env.RELAY_OUTBOX_RELAY ?? "on"`), so those nine booted four
+    // background loops that sweep the whole database while every other suite's
+    // fixtures sit in it. Research R13 recorded the exposure as nil on the strength
+    // of the four suites that spawn an api CHILD and set the flags in the child's
+    // env; it did not look at the suites that boot the app in process.
+    //
+    // A relay catches and logs its own errors, so the guard's refusal inside one is
+    // a log line and a green lane. Setting the flags here makes the quiet database
+    // a property of the lane rather than a convention nobody applied.
+    env: {
+      RELAY_OUTBOX_RELAY: "off",
+      RELAY_DELIVERY_RELAY: "off",
+      RELAY_NOTIFICATION_RELAY: "off",
+      RELAY_EVENT_CONSUMER: "off",
+      // The quota relay, the fourth. Same reason as the other three.
+      RELAY_QUOTA_RELAY: "off",
+    },
     setupFiles: ["./packages/test-harness/src/setup.ts"],
     include: [
       "packages/*/src/**/*.test.ts",
       "services/*/src/**/*.test.ts",
       "packages/*/src/**/*.itest.ts",
       "services/*/src/**/*.itest.ts",
@@ -636,12 +655,108 @@ export default defineConfig({
         "services/gateway/src/limits.ts": {
           branches: 90,
           functions: 100,
           lines: 100,
           statements: 100,
         },
+
+        // HOW THESE SIX WERE VALIDATED, STATED BECAUSE THE USUAL PROBE DOES NOT WORK
+        // HERE. The re-pin ritual is "demand 101% of a real key and watch it name
+        // itself; demand 101% of a key naming no file and watch nothing happen".
+        // Running that against ONE test file is vacuous twice over: without
+        // `--coverage` the thresholds are not evaluated at all, and with it the global
+        // floor fires first — `Coverage for lines (0.28%) does not meet global
+        // threshold (70%)` — and drowns the per-file signal. Both were measured.
+        //
+        // So these are validated by the FULL run, which is the instrument: 87 files,
+        // 1,263 tests, exit 0 with every key below holding. The silent-key half is
+        // carried from the previous chapter's probe rather than re-run.
+        //
+        // THE QUOTA CHAPTER'S SIX, AND PUBLISHED PINNED NONE OF THEM. The chapter
+        // before this one pinned all eight of its files; this one shipped seven and
+        // left the ratchet nothing to hold, which is visible only by comparing two
+        // chapters that are adjacent in THIS order and were not in the published one.
+        //
+        // `policy.ts` and `quota.error.ts` are pure — no clock, no store, no
+        // framework — and reach 100 on every metric. A branch they miss is a case
+        // nobody thought of rather than one nobody could reach.
+        "services/api/src/quotas/policy.ts": {
+          branches: 100,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+        "services/api/src/quotas/quota.error.ts": {
+          branches: 100,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+
+        // AND THREE WHOSE ONLY SHORTFALL IS A `??` THE COMPILER DEMANDS AND THE DATA
+        // NEVER REACHES. Each is a fallback on a value `noUncheckedIndexedAccess`
+        // will not narrow:
+        //
+        //   config.ts:65      `parsed.error.issues[0]?.message ?? "invalid"` — a zod
+        //                     failure carries at least one issue, always
+        //   period.ts:43-44   `(m ?? 1)` and `(y ?? 0)` in `nextPeriod`, on a period
+        //                     that came from `periodOf` or from a `date` column
+        //   quota-email.ts:31 `months[Number(m) - 1] ?? m` — a month outside 1..12
+        //
+        // READ THE FRACTION, NOT THE PERCENTAGE, because 75 reads like a hole and is
+        // not one. These files are small enough that one arm moves the figure a long
+        // way, and the counts say what the percentages hide:
+        //
+        //   config.ts       9/10 branches   the one is the `?? "invalid"`
+        //   period.ts       6/8             the two are `?? 1` and `?? 0`
+        //   quota-email.ts  6/8             the two are `?? m`, both arms of one guard
+        //   quota-relay.ts  29/30 statements  the one is the catch inside `run()`
+        //
+        // A pin of 75 on an eight-branch file leaves room for exactly the two arms
+        // named above and nothing else: lose a third and it goes red.
+        //
+        // NOT DELETED, WHICH IS THIS RATCHET'S USUAL ANSWER, because deleting them
+        // means a non-null assertion: the same assumption moved somewhere a type
+        // change cannot invalidate. Pinned at the reading with the arm named instead.
+        //
+        // AND THE FIGURE IMPROVED BY REMOVING A COPY RATHER THAN ADDING A TEST.
+        // `quota.error.ts` and `quota-email.ts` each carried this arithmetic and each
+        // paid for its own pair of unreachable arms; the extraction into `period.ts`
+        // left one pair, and this directory's branches went 86.53% to 88.63%.
+        "services/api/src/quotas/config.ts": {
+          branches: 90,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+        "services/api/src/quotas/period.ts": {
+          branches: 75,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+        "services/api/src/quotas/quota-email.ts": {
+          branches: 75,
+          functions: 100,
+          lines: 100,
+          statements: 100,
+        },
+
+        // The relay's loop, and the shortfall is the one every relay in this codebase
+        // reports: `start`, `stop` and the `run` loop are entered by no test, because
+        // every suite drives `drainOnce()` directly — which is the right way to assert
+        // on rows and the wrong way to learn whether the loop that calls it in
+        // production works. Line 118 is the catch inside that loop. Three files
+        // already sit in this config unpinned for the same reason; this one is pinned
+        // because the rest of it IS covered and a floor at 70 would hold nothing.
+        "services/api/src/quotas/quota-relay.ts": {
+          branches: 100,
+          functions: 100,
+          lines: 96,
+          statements: 96,
+        },
       },
     },
   },
   plugins: [
     swc.vite({
       module: { type: "es6" },

@@ -47,13 +47,14 @@
         "RELAY_DELIVERY_RELAY",
         "RELAY_INTERNAL_CREDENTIAL",
         "RELAY_AUTH_FAILURES_PER_MINUTE",
         "RELAY_AUTH_KEY_PREFIX",
         "RELAY_WEBHOOK_SECRET_KEY",
         "RELAY_EVENT_CONSUMER",
-        "RELAY_NATS_REPLICAS"
+        "RELAY_NATS_REPLICAS",
+        "RELAY_QUOTA_RELAY"
       ]
     },
     "//#lint:root": {
       "inputs": [
         "**/*.{ts,mts,cts,mjs,js}",
         "eslint.config.mjs",

@@ -312,12 +312,19 @@ export interface System {
     credential: string;
     channel: string;
     dispatcher: Client;
     tuan: Client;
   }>;
   seedForeignTenant: () => Promise<{ channel: string; text: string }>;
+  /** Set an environment's quota policy.
+   *
+   * Here rather than in the test, because `packages/e2e` may not import `pg` —
+   * the driver restriction chapter 2.5 added, and this package is not on its
+   * ignores list. The harness already holds the api's own database handle, so
+   * the one place that may write is the one place that does. */
+  setQuota: (environmentId: string, config: unknown) => Promise<void>;
   client: (name: string, environmentId: string) => Promise<Client>;
   stop: () => Promise<void>;
 }
 
 export async function boot({ gateways = 2 } = {}): Promise<System> {
   const { db, seeder } = loadApiInternals();
@@ -551,12 +558,20 @@ export async function boot({ gateways = 2 } = {}): Promise<System> {
       await repo.addMember(channel.id, user.id);
       const text = "this belongs to another tenant";
       await repo.sendMessage(channel.id, { text, userId: user.id });
       say(`seeded a foreign tenant (${other}) with one message`);
       return { channel: channel.id, text };
     },
+    async setQuota(environmentId, config) {
+      await (
+        db as { execute: (q: string) => Promise<unknown> }
+      ).execute(
+        `UPDATE environments SET quota_config = '${JSON.stringify(config)}'::jsonb
+          WHERE id = '${environmentId}'`,
+      );
+    },
     async client(name, environmentId) {
       return new Client(name, await token(environmentId, name), say);
     },
     async stop() {
       for (const child of children) child.kill("SIGTERM");
       await new Promise((resolve) => setTimeout(resolve, 200));

import { afterAll, beforeAll, describe, expect, it } from "vitest";
 
import type { Frame } from "@relay/protocol";
 
import { boot, type Client, type System } from "./harness.js";
 
// The cap through the OTHER door, and what it leaves alone.
//
// The rate-limit chapter's limiter never sees `/internal/messages`: `operationsFor` returns
// [] for anything outside `/v1`, and that is the route the gateway posts a
// WebSocket send to. This chapter's enforcement point is `sendMessage`, which
// both doors reach — and this is the test that proves it rather than the research
// note that predicted it (research R3).
//
// This lane and not the gateway's, because the refusal has to come from a real
// api child and the gateway's own lane does not spawn one.
 
describe("a quota through the socket", () => {
  let system: System;
  let environmentId: string;
  let channel: string;
  let dispatcher: Client;
 
  beforeAll(async () => {
    system = await boot({ gateways: 1 });
    const seeded = await system.seedConversation();
    ({ environmentId, channel, dispatcher } = seeded);
    await dispatcher.connect(system.gateways[0]!);
  }, 120_000);
 
  afterAll(async () => {
    await system?.stop();
  });
 
  it("refuses the send and leaves the socket open", async () => {
    // One message through, then the cap set at what has already been used.
    dispatcher.send(channel, "before the cap");
    await dispatcher.waitFor(
      (f: Frame) => f.type === "message.created",
      "the first message to land",
    );
 
    await system.setQuota(environmentId, { messages: { hard: 1 } });
 
    const before = dispatcher.frames.length;
    dispatcher.send(channel, "after the cap");
 
    // The socket does not carry the 402 — the gateway holds the connection and
    // the api refuses the POST behind it. What matters here is the negative: no
    // `message.created` for the refused send, and the connection still up.
    await new Promise((r) => setTimeout(r, 2_000));
    const created = dispatcher.frames
      .slice(before)
      .filter((f: Frame) => f.type === "message.created");
    expect(created).toHaveLength(0);
 
    // SC-003 — STILL OPEN, AND STILL RECEIVING. A cap that closed the socket
    // would be an outage dressed as a business control, and FR-RTL-08 exists to
    // say it is not one.
    await system.setQuota(environmentId, {});
    dispatcher.send(channel, "after the cap was lifted");
    await dispatcher.waitFor(
      (f: Frame) =>
        f.type === "message.created" &&
        JSON.stringify(f).includes("after the cap was lifted"),
      "the socket to still be carrying messages",
    );
  }, 120_000);
});

@@ -1,12 +1,13 @@
 import { sql } from "drizzle-orm";
 import {
   bigserial,
   bigint,
   boolean,
   check,
+  date,
   index,
   integer,
   jsonb,
   pgTable,
   primaryKey,
   text,
@@ -930,6 +931,122 @@ export const webhookDisableNotifications = pgTable(
     // Nothing beyond the primary key and the tenant. Volume is one row per
     // endpoint per outage, so an index for any other access pattern would be
     // guessing at a query nobody has written.
     index("webhook_disable_notifications_environment_idx").on(t.environmentId),
   ],
 );
+
+// ---------------------------------------------------------------------------
+// Monthly usage quotas (FR-RTL-05 to FR-RTL-08).
+// ---------------------------------------------------------------------------
+//
+// The POLICY is not here, because it was already here. `environments.quotaConfig`
+// has been declared since chapter 2.1 and read by nothing for eighteen chapters;
+// the rate-limit chapter was offered it for rate-limit policy and refused it in prose, on the
+// grounds that the column is named for quotas and quotas are a later chapter.
+// This is that chapter. `quotas/config.ts` is the only thing that parses it.
+//
+// `date` IS THIS PROJECT'S FIRST, against 28 `timestamp` columns, and it is half
+// a primary key here and a third of one below. Drizzle's `date` in its default
+// mode reads and writes `YYYY-MM-DD` strings, which is what `quotas/period.ts`
+// produces — a `Date` on one side of that comparison and a string on the other is
+// a row that cannot be found rather than an error (research R7a).
+
+export const usagePeriods = pgTable(
+  "usage_periods",
+  {
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    period: date("period").notNull(),
+    // `{ mode: "number" }` like the two bigints this project already has
+    // (`channels.lastSequence`, `messages.sequence`). Drizzle requires a mode,
+    // and a cumulative count that overflowed would be a wrong bill rather than a
+    // wrapped counter.
+    messagesSent: bigint("messages_sent", { mode: "number" })
+      .notNull()
+      .default(0),
+    createdAt: timestamp("created_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+  },
+  (t) => [
+    primaryKey({ columns: [t.environmentId, t.period] }),
+    check(
+      "usage_periods_messages_sent_non_negative",
+      sql`${t.messagesSent} >= 0`,
+    ),
+  ],
+);
+
+// One row per user per period. A message count is `+1`; a distinct-user count is
+// not, because incrementing it needs to know whether this user already sent this
+// period — which is a read. The row IS the answer, written `ON CONFLICT DO
+// NOTHING`, and bounded by the tenant's users rather than by their traffic.
+export const usageActiveUsers = pgTable(
+  "usage_active_users",
+  {
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    period: date("period").notNull(),
+    userId: uuid("user_id")
+      .notNull()
+      .references(() => users.id),
+    firstSeenAt: timestamp("first_seen_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+  },
+  (t) => [primaryKey({ columns: [t.environmentId, t.period, t.userId] })],
+);
+
+// THE OUTBOX, A FOURTH TIME — after the outbox chapter's events, the webhook
+// dispatcher chapter's deliveries and the mail-transport chapter's disablement
+// emails. Four concrete tables that look alike is a pattern; one
+// abstract table serving four purposes is a framework.
+//
+// `webhookDisableNotifications` cannot be reused: its `endpointId` is NOT NULL
+// and a quota crossing has no endpoint.
+export const quotaNotifications = pgTable(
+  "quota_notifications",
+  {
+    id: uuid("id").primaryKey(),
+    environmentId: uuid("environment_id")
+      .notNull()
+      .references(() => environments.id),
+    organisationId: uuid("organisation_id")
+      .notNull()
+      .references(() => organisations.id),
+    period: date("period").notNull(),
+    dimension: text("dimension").notNull(),
+    threshold: integer("threshold").notNull(),
+    // What the figures were WHEN IT HAPPENED. The cap can change between the
+    // crossing and the delivery, and an email saying "80% of 10,000" should mean
+    // the 10,000 that was true at the time.
+    quota: bigint("quota", { mode: "number" }).notNull(),
+    usageAtCrossing: bigint("usage_at_crossing", { mode: "number" }).notNull(),
+    crossedAt: timestamp("crossed_at", { withTimezone: true })
+      .notNull()
+      .defaultNow(),
+    deliveredAt: timestamp("delivered_at", { withTimezone: true }),
+    lastError: text("last_error"),
+  },
+  (t) => [
+    check(
+      "quota_notifications_dimension_check",
+      sql`${t.dimension} IN ('messages', 'active_users')`,
+    ),
+    check(
+      "quota_notifications_threshold_check",
+      sql`${t.threshold} IN (50, 80, 100)`,
+    ),
+    // THIS CONSTRAINT IS FR-RTL-07. At most one email per threshold per quota per
+    // period, enforced by the schema rather than promised by the code that writes
+    // it, so a concurrent double-crossing resolves to one row and not two emails.
+    unique("quota_notifications_once_per_threshold").on(
+      t.environmentId,
+      t.period,
+      t.dimension,
+      t.threshold,
+    ),
+  ],
+);

@@ -20,24 +20,31 @@ import {
   messageEdits,
   readPositions,
   memberships,
   messages,
   organisations,
   outbox,
+  quotaNotifications,
+  usageActiveUsers,
+  usagePeriods,
   users,
   webhookDeadLetters,
   webhookDeliveries,
   webhookDisableNotifications,
   webhookEndpoints,
 } from "./schema";
 import {
   membershipEvent,
   messageCreatedEvent,
   messageDeletedEvent,
   messageUpdatedEvent,
 } from "../outbox/event";
+import { capsFor, type Caps } from "../quotas/config";
+import { thresholdsCrossed } from "../quotas/policy";
+import { QuotaExceededError, type Dimension } from "../quotas/quota.error";
+import { periodOf } from "../quotas/period";
 import { nextAttemptAt } from "../webhooks/schedule";
 import {
   DISABLE_AFTER_MS,
   DISABLE_MIN_ATTEMPTS,
   disableReason,
   runWindowMs,
@@ -275,12 +282,149 @@ export async function environmentLimits(
     rest: row.rest ?? DEFAULT_LIMITS.rest,
     send: row.send ?? DEFAULT_LIMITS.send,
     connect: row.connect ?? DEFAULT_LIMITS.connect,
   };
 }
 
+/** What an environment has consumed in a period, and what it is allowed
+ * (FR-RTL-05).
+ *
+ * ZEROS FOR A PERIOD WITH NO ROWS, not null and not an error. An environment that
+ * has sent nothing has used nothing, and making every caller tell "no usage" apart
+ * from "no row" would push a schema detail into each of them.
+ *
+ * A NULL QUOTA IS CARRIED THROUGH AS NULL rather than resolved to `Infinity` or
+ * `-1`. The absent state stays absent all the way to the reader — the same rule
+ * the rate-limit chapter's nullable limit columns encode, and the reason `capsFor` returns
+ * `null` rather than a sentinel.
+ *
+ * Admin surface: takes an environment id rather than being scoped by construction,
+ * because the relay and the internal route both read it on behalf of the platform.
+ * Bounded by an id, so it crosses environments and cannot run away (the third
+ * category in this file's taxonomy). */
+export async function usageFor(
+  db: Db,
+  environmentId: string,
+  period: string,
+): Promise<{
+  period: string;
+  messagesSent: number;
+  activeUsers: number;
+  messageQuota: number | null;
+  activeUserQuota: number | null;
+}> {
+  const [row] = await db
+    .select({
+      messagesSent: usagePeriods.messagesSent,
+      quotaConfig: environments.quotaConfig,
+    })
+    .from(environments)
+    .leftJoin(
+      usagePeriods,
+      and(
+        eq(usagePeriods.environmentId, environments.id),
+        eq(usagePeriods.period, period),
+      ),
+    )
+    .where(eq(environments.id, environmentId));
+
+  const [users] = await db
+    .select({ n: sql<number>`count(*)::int` })
+    .from(usageActiveUsers)
+    .where(
+      and(
+        eq(usageActiveUsers.environmentId, environmentId),
+        eq(usageActiveUsers.period, period),
+      ),
+    );
+
+  return {
+    period,
+    messagesSent: row?.messagesSent ?? 0,
+    activeUsers: users?.n ?? 0,
+    messageQuota: capsFor(row?.quotaConfig, "messages").caps.hard,
+    activeUserQuota: capsFor(row?.quotaConfig, "active_users").caps.hard,
+  };
+}
+
+/** One quota crossing waiting to be emailed. */
+export interface QuotaNotificationRow {
+  id: string;
+  organisationId: string;
+  environmentName: string;
+  period: string;
+  dimension: string;
+  threshold: number;
+  quota: number;
+  usageAtCrossing: number;
+  /** Whether a hard cap is in force for this dimension right now — which decides
+   * whether the email says sends have stopped or that nothing has changed. Read
+   * at delivery rather than stored, because it is a statement about the present
+   * and the operator may have raised the cap since. */
+  hardCapInForce: boolean;
+}
+
+/** The outbox drain, a FOURTH time — after the outbox chapter's events, the webhook
+ * dispatcher chapter's deliveries and the mail-transport chapter's disablement
+ * emails. Same claim predicate, same per-row error handling, same required batch
+ * size.
+ *
+ * PER-ROW `try`/`catch` WITH A REQUIRED `onError`, and the default that used to
+ * sit on the mail-transport chapter's version is not repeated here: it discarded a
+ * row's failure with no log line, and feature 030's R48 removed it after finding it
+ * as this file's last
+ * uncovered function. One bad recipient must not abort the batch and must not
+ * vanish either. */
+export async function drainQuotaNotifications(
+  db: Db,
+  limit: number,
+  deliver: (row: QuotaNotificationRow) => Promise<void>,
+  onError: (row: QuotaNotificationRow, error: unknown) => void,
+): Promise<number> {
+  return db.transaction(async (tx) => {
+    const claimed = (await tx.execute(
+      sql`SELECT q.id                AS "id",
+                 q.organisation_id   AS "organisationId",
+                 a.name || ' / ' || e.kind AS "environmentName",
+                 to_char(q.period, 'YYYY-MM-DD') AS "period",
+                 q.dimension         AS "dimension",
+                 q.threshold         AS "threshold",
+                 q.quota             AS "quota",
+                 q.usage_at_crossing AS "usageAtCrossing",
+                 (e.quota_config #>> ('{' || q.dimension || ',hard}')::text[])
+                   IS NOT NULL       AS "hardCapInForce"
+            FROM quota_notifications q
+            JOIN environments e ON e.id = q.environment_id
+            JOIN applications a ON a.id = e.application_id
+           WHERE q.delivered_at IS NULL
+           ORDER BY q.crossed_at
+           LIMIT ${limit}
+             FOR UPDATE OF q SKIP LOCKED`,
+    )) as unknown as { rows: QuotaNotificationRow[] };
+
+    let delivered = 0;
+    for (const row of claimed.rows) {
+      try {
+        await deliver(row);
+        await tx.execute(
+          sql`UPDATE quota_notifications SET delivered_at = now(), last_error = NULL
+               WHERE id = ${row.id}::uuid`,
+        );
+        delivered += 1;
+      } catch (error) {
+        onError(row, error);
+        await tx.execute(
+          sql`UPDATE quota_notifications SET last_error = ${String(error)}
+               WHERE id = ${row.id}::uuid`,
+        );
+      }
+    }
+    return delivered;
+  });
+}
+
 // ---------------------------------------------------------------------------
 // The outbox drain (ADR-06). Part of the ADMIN surface for the
 // same reason the credential lookup is: it runs on behalf of the platform
 // rather than of a tenant, and it is deliberately NOT scoped by environment —
 // one relay drains every environment's events, because an outbox row is work
 // the platform owes itself.
@@ -3520,12 +3664,19 @@ export class Repository {
       text: string;
       metadata?: unknown;
       idempotencyKey?: string;
     },
   ): Promise<MessageRow> {
     return this.db.transaction(async (tx) => {
+      // ONE PERIOD FOR THE WHOLE TRANSACTION, taken before anything is checked.
+      // The cap check and the increment must agree about which month this is; a
+      // send that checked August and incremented September would be refused
+      // against one number and counted against another. The app clock rather
+      // than the database's, because both statements need the same value and
+      // only one of them can be `now()`.
+      const period = periodOf(new Date());
 
       // ── THE BAN, FIRST, AND AHEAD OF THE CHANNEL READ (FR-031, FR-021a) ─────
       //
       // T072 left this slot and only Phase 15 can fill it, because until now nothing
       // wrote `banned_at`. The position is the requirement: **before the channel is
       // resolved**, so a banned user gets one answer for every channel id — real,
@@ -3684,12 +3835,43 @@ export class Repository {
       // T155a. Leaving the slot visible is the point; a reader who finds two checks
       // where the requirement names three should be able to see which is missing.
       //
       // History stays readable while archived (FR-020). Only the write refuses.
       if (channel.archivedAt !== null) throw new ChannelArchivedError(channelId);
 
+      // THE CAP, CHECKED BEFORE THE MESSAGE IS WRITTEN (FR-RTL-08).
+      //
+      // Here rather than in middleware, because the rate-limit chapter's limiter never sees
+      // `/internal/messages` — `operationsFor` returns [] for anything outside
+      // `/v1` — and that is the route a WebSocket send arrives on. Both doors
+      // reach this method, and it already owns the write transaction, so the
+      // check and the increment commit together (research R3).
+      //
+      // A PLAIN READ, AND THE OVERSHOOT IS STATED RATHER THAN DEFENDED AGAINST.
+      //
+      // The first version took `FOR UPDATE` on the usage row, which bounds the
+      // overshoot to exactly one message. Two things retired it.
+      //
+      // The caps and the usage are now ONE joined read, and Postgres will not
+      // lock that:
+      //
+      //   ERROR:  FOR UPDATE cannot be applied to the nullable side of an outer join
+      //
+      // And the specification never asked for the lock. Its edge case reads: "the
+      // overshoot is bounded by concurrency, not unbounded, and this is stated
+      // rather than defended against." A few dozen sends in flight against a
+      // monthly cap of thousands is a bound worth naming rather than engineering
+      // around.
+      //
+      // WHAT THE QUOTA PATH COSTS, measured with the phases instrumented and the
+      // config toggled on one environment: 0.56ms per send at 32-way concurrency.
+      // The joined read is about 1.2ms of that and US1 needs it whether or not a
+      // cap exists. An earlier uncontrolled benchmark reported 273% and sent three
+      // separate hypotheses chasing what turned out to be warm-up (T033).
+      const quota = await this.assertWithinQuota(tx, period, userId, senderIsPerson);
+
       const seq = channel.lastSequence + 1;
       const id = randomUUID();
 
       const insert = tx.insert(messages).values({
         id,
         channelId: channel.id,
@@ -3795,12 +3977,125 @@ export class Repository {
       });
       await tx.insert(outbox).values({
         subject: event.subject,
         payload: event.payload,
       });
 
+      // THE MONTH'S USAGE COMMITS WITH THE MESSAGE (FR-RTL-05).
+      //
+      // Same argument as the event above it, one requirement further on. A quota
+      // is about THIS MONTH and must not forget, so the count cannot live in the
+      // per-minute counter store the rate-limit chapter built — a flush there costs one
+      // window of over-service, a flush here costs the month. A quota must survive
+      // the counter store.
+      //
+      // It is an increment rather than a query because the alternative is a read
+      // over `messages`, which carries no `environment_id` and no index on
+      // `created_at`: the month predicate becomes a Filter applied after every
+      // row the tenant has ever sent is read off the heap. Fast today, and
+      // proportional to lifetime traffic forever (research R1).
+      //
+      // On the INSERTED branch only, like the event. A recognised idempotent
+      // retry wrote no message and must consume no quota either, or a client
+      // retrying on a flaky link is billed twice for one message.
+      await tx
+        .insert(usagePeriods)
+        .values({ environmentId: this.environmentId, period, messagesSent: 1 })
+        .onConflictDoUpdate({
+          target: [usagePeriods.environmentId, usagePeriods.period],
+          set: { messagesSent: sql`${usagePeriods.messagesSent} + 1` },
+        });
+
+      // The distinct-user count.
+      // UNCONDITIONAL, AND THE GUARD THAT WAS HERE COULD NOT BE FALSE. It read
+      // `if (userId !== undefined)`, with a comment saying a key-authenticated REST
+      // send carries no `userId` and so counts toward the message quota and toward no
+      // user. That was true of a platform where the parameter was optional; the sender
+      // chapter made it REQUIRED — `userId: string`, and its own comment says required
+      // is the whole mechanism, because SC-003 asks that no write path be able to
+      // produce a senderless message.
+      //
+      // So there is no unattributed send left to guard against, and the arm was one
+      // the type system forbids. Deleted rather than covered, which is what this
+      // repository does with an unreachable arm — and the deletion is also the more
+      // honest statement: every message counted here has a sender, and the distinct-user
+      // count is exactly the senders.
+      //
+      // A row rather than a counter, and that has not changed: incrementing one would
+      // need to know whether this user already sent this period, which is a read. The
+      // row IS the answer, and `ON CONFLICT DO NOTHING` makes the second send of the
+      // month free.
+      await tx
+        .insert(usageActiveUsers)
+        .values({ environmentId: this.environmentId, period, userId })
+        .onConflictDoNothing();
+
+      // What this send crossed, if anything. Almost always nothing, which is why
+      // the caps are read first and the whole block skipped when none is set.
+      // WORK OUT WHETHER ANYTHING WAS CROSSED BEFORE ASKING THE DATABASE ANYTHING.
+      //
+      // `thresholdsCrossed` is pure arithmetic on two numbers the transaction
+      // already holds, and it answers "nothing" for almost every send. The first
+      // version looked up the organisation and counted the period's users FIRST
+      // and consulted the arithmetic afterwards, which put two extra queries on
+      // every send by an environment that merely HAS a quota — measured at 341%
+      // over the unconfigured path and mistaken, at first, for the cost of a lock
+      // (T033).
+      if (quota) {
+        const messageRef =
+          quota.caps.messages.hard ?? quota.caps.messages.soft;
+        const crossedMessages = thresholdsCrossed(
+          quota.sent,
+          quota.sent + 1,
+          messageRef,
+        );
+        // The user count is only worth asking for when a user cap exists AND this
+        // send could have added someone.
+        const userRef =
+          quota.caps.active_users.hard ?? quota.caps.active_users.soft;
+        const mayHaveAddedUser = userId !== undefined && userRef !== null;
+
+        if (crossedMessages.length > 0 || mayHaveAddedUser) {
+          const organisationId = await this.organisationOf(tx);
+          if (organisationId) {
+            if (crossedMessages.length > 0) {
+              await this.recordCrossings(
+                tx,
+                period,
+                "messages",
+                quota.sent,
+                quota.sent + 1,
+                quota.caps.messages,
+                organisationId,
+              );
+            }
+            if (mayHaveAddedUser) {
+              const [n] = await tx
+                .select({ n: sql<number>`count(*)::int` })
+                .from(usageActiveUsers)
+                .where(
+                  and(
+                    eq(usageActiveUsers.environmentId, this.environmentId),
+                    eq(usageActiveUsers.period, period),
+                  ),
+                );
+              const users = n?.n ?? 0;
+              await this.recordCrossings(
+                tx,
+                period,
+                "active_users",
+                users - 1,
+                users,
+                quota.caps.active_users,
+                organisationId,
+              );
+            }
+          }
+        }
+      }
+
       return {
         id,
         channel_id: channel.id,
         seq,
         text,
         /** WHAT WAS SENT, AND THE INSERT IS NOT ENOUGH ON ITS OWN.
@@ -4299,12 +4594,236 @@ export class Repository {
         ),
       )
       .limit(1);
     return rows.length > 0;
   }
 
+  /** Refuse the send if a hard cap is already met (FR-RTL-08).
+   *
+   * Reads the caps and the usage in ONE query, in the transaction that is about to
+   * write. Both dimensions, because FR-RTL-06 configures a cap for each.
+   *
+   * No lock: see the note at the call site. Postgres will not take `FOR UPDATE` on
+   * the nullable side of the outer join this read needs, and the overshoot it
+   * would have bounded is small enough to state instead.
+   *
+   * THE ACTIVE-USER CHECK ONLY BITES ON A NEW SENDER. A tenant at its user cap is
+   * not cut off from the users it already has — the cap is on how many distinct
+   * people may send in a month, not on how much they may say. So a sender already
+   * counted this period passes, and only the one who would be the next new face
+   * is refused. Getting this backwards would suspend a whole tenant the moment
+   * their last allowed user sent their second message. */
+  private async assertWithinQuota(
+    tx: Db,
+    period: string,
+    /** `string`, NOT `string | undefined`, and the narrowing is the sender chapter's.
+     * Its one caller is `sendMessage`, whose `userId` is required — so the optional
+     * type here bought an arm nothing could take, and the arm below it read
+     * `users_.hard === null || userId === undefined`. Second dead `userId` comparison
+     * this chapter's port has met; the first was the distinct-user insert's guard. */
+    userId: string,
+    /** WHETHER THE SENDER IS A PERSON, computed at the ban check from the same row so
+     * this costs nothing beyond passing it. It decides the ENFORCED ceiling and not the
+     * bill — see the two notes below, which are the two halves of one exemption. */
+    senderIsPerson: boolean,
+  ): Promise<{
+    caps: { messages: Caps; active_users: Caps };
+    sent: number;
+  } | null> {
+    // ONE QUERY, NOT TWO. The caps live on `environments` and the usage on
+    // `usage_periods`, and reading them separately costs two round-trips inside
+    // the write transaction — which holds a pooled connection for the duration.
+    // Above the pool size that queues, and T033 measured the two-query version at
+    // 7.95ms per send against 1.45ms unconfigured at 32-way concurrency. Joined,
+    // it is one round-trip on two primary keys.
+    const [env] = await tx
+      .select({
+        quotaConfig: environments.quotaConfig,
+        messagesSent: usagePeriods.messagesSent,
+      })
+      .from(environments)
+      .leftJoin(
+        usagePeriods,
+        and(
+          eq(usagePeriods.environmentId, environments.id),
+          eq(usagePeriods.period, period),
+        ),
+      )
+      .where(eq(environments.id, this.environmentId));
+
+    const messages_ = capsFor(env?.quotaConfig, "messages").caps;
+    const users_ = capsFor(env?.quotaConfig, "active_users").caps;
+    // Nothing configured at all — no cap and no threshold — and the whole block
+    // is skipped. The unconfigured tenant is the common case and pays one
+    // indexed read for it.
+    if (
+      messages_.hard === null &&
+      messages_.soft === null &&
+      users_.hard === null &&
+      users_.soft === null
+    ) {
+      return null;
+    }
+
+    const sent = env?.messagesSent ?? 0;
+
+    if (messages_.hard !== null && sent >= messages_.hard) {
+      // THE CROSSING IS WRITTEN BEFORE THE REFUSAL IS RAISED (the ordering rule).
+      //
+      // Usually the send that reached the cap already recorded 100%. Two cases
+      // where it did not: a cap lowered below current usage, which no send
+      // crossed, and a soft threshold configured at the same value as the hard
+      // cap. The email has to survive the send that did not, so the row goes in
+      // first and the throw comes after. `ON CONFLICT DO NOTHING` makes the
+      // usual case free.
+      const organisationId = await this.organisationOf(tx);
+      if (organisationId) {
+        await this.recordCrossings(
+          tx,
+          period,
+          "messages",
+          sent - 1,
+          sent,
+          messages_,
+          organisationId,
+        );
+      }
+      throw new QuotaExceededError({
+        dimension: "messages",
+        usage: sent,
+        quota: messages_.hard,
+        period,
+      });
+    }
+
+    // A BOT IS EXEMPT FROM THE CEILING, AND BILLED FOR THE SEND (FR-RTL-05 as this
+    // project's own SRS amendment left it). The clause caps "unique active PERSONS";
+    // FR-ANL-05 still meters "unique active users", and the insert in `sendMessage`
+    // counts a bot like anyone — which is what makes a bot billed and exempt at once.
+    //
+    // THE REASON IS WHOSE SEND GETS REFUSED. The ceiling bounds a customer's human
+    // population, and a customer's own software must not be able to lock their people
+    // out of sending. It would: the block below refuses the FIRST send of a period by
+    // anyone once the count is reached, so the person refused is never whoever caused
+    // it.
+    if (!senderIsPerson) {
+      return { caps: { messages: messages_, active_users: users_ }, sent };
+    }
+    // NO CAP ON USERS, NOTHING MORE TO CHECK. The `|| userId === undefined` that stood
+    // beside this is gone with the parameter's type: a send with no sender is a state
+    // no write path can produce since the sender chapter made `userId` required.
+    if (users_.hard === null) {
+      return { caps: { messages: messages_, active_users: users_ }, sent };
+    }
+
+    const [already] = await tx
+      .select({ userId: usageActiveUsers.userId })
+      .from(usageActiveUsers)
+      .where(
+        and(
+          eq(usageActiveUsers.environmentId, this.environmentId),
+          eq(usageActiveUsers.period, period),
+          eq(usageActiveUsers.userId, userId),
+        ),
+      );
+    if (already) {
+      return { caps: { messages: messages_, active_users: users_ }, sent };
+    }
+
+    // AND THIS IS THE EXEMPTION'S SECOND HALF, which is the one that decides whether
+    // it works. Returning early above is visible: a bot's send is not refused. But the
+    // count the ceiling compares against would still hold the bot's row, displacing a
+    // person — so a customer at a ceiling of five with two bots could seat three
+    // people. **A test that only watches a bot's send succeed passes with the first
+    // half alone**, which is why the one below sends as a PERSON after a bot.
+    //
+    // THE JOIN FILTERS `kind` AND NOT `deleted_at`, and the wrong version is the one a
+    // careful reader writes: three `users` joins in this file pair with
+    // `isNull(users.deletedAt)` and it is the house idiom. `deleteUser` is a SOFT
+    // delete and leaves `usage_active_users` alone, so adding that filter would make a
+    // deleted person's row stop counting — and deleting users would become a way to
+    // free ceiling slots, which it is not.
+    const [count] = await tx
+      .select({ n: sql<number>`count(*)::int` })
+      .from(usageActiveUsers)
+      .innerJoin(users, eq(users.id, usageActiveUsers.userId))
+      .where(
+        and(
+          eq(usageActiveUsers.environmentId, this.environmentId),
+          eq(usageActiveUsers.period, period),
+          eq(users.kind, "person"),
+        ),
+      );
+    const active = count?.n ?? 0;
+    if (active >= users_.hard) {
+      throw new QuotaExceededError({
+        dimension: "active_users",
+        usage: active,
+        quota: users_.hard,
+        period,
+      });
+    }
+    return { caps: { messages: messages_, active_users: users_ }, sent };
+  }
+
+  /** Write a row for each threshold a usage increase crossed (FR-RTL-07).
+   *
+   * IN THE SAME TRANSACTION AS THE THING THAT CAUSED IT. The crossing and the
+   * message commit together or neither does, which is the same argument the
+   * event above them makes and the reason there is no periodic sweep in this
+   * chapter at all: usage only ever rises because of a send, and the send knows
+   * the value before and after, so it knows what it crossed (research R5).
+   *
+   * THE PERCENTAGE IS OF `hard ?? soft`. A soft threshold with no hard cap is
+   * still a figure an operator asked to be warned about, and 100% of it is worth
+   * an email even though nothing will be refused.
+   *
+   * `ON CONFLICT DO NOTHING` against `quota_notifications_once_per_threshold` is
+   * what makes it at-most-once (FR-RTL-07) — the schema, not this code. A concurrent
+   * double-crossing resolves to one row rather than two emails. */
+  private async recordCrossings(
+    tx: Db,
+    period: string,
+    dimension: Dimension,
+    before: number,
+    after: number,
+    caps: { hard: number | null; soft: number | null },
+    organisationId: string,
+  ): Promise<void> {
+    const reference = caps.hard ?? caps.soft;
+    if (reference === null) return;
+    const crossed = thresholdsCrossed(before, after, reference);
+    if (crossed.length === 0) return;
+
+    await tx
+      .insert(quotaNotifications)
+      .values(
+        crossed.map((threshold) => ({
+          id: randomUUID(),
+          environmentId: this.environmentId,
+          organisationId,
+          period,
+          dimension,
+          threshold,
+          quota: reference,
+          usageAtCrossing: after,
+        })),
+      )
+      .onConflictDoNothing();
+  }
+
+  /** The organisation an environment belongs to — who gets told. */
+  private async organisationOf(tx: Db): Promise<string | null> {
+    const [row] = await tx
+      .select({ organisationId: applications.organisationId })
+      .from(environments)
+      .innerJoin(applications, eq(applications.id, environments.applicationId))
+      .where(eq(environments.id, this.environmentId));
+    return row?.organisationId ?? null;
+  }
+
   /** Fetch a message by its idempotency key within a channel — the
    * recovery leg of 2.3's duplicate-recognised path. The channel join
    * carries the tenant scope: every query in this layer answers only for
    * its own environment, private helpers included (constitution I). */
   private async getMessageByIdempotencyKey(
     tx: Db,