fix(events): keep three exported accounts behind every named organisation

db/account-export.ts

@@ -6,11 +6,11 @@ import type { Pool } from 'pg';
 /**
  * The account export, which the event export Job (db/export-events.ts) runs
  * before the event export: copies the Postgres view `account_daily`
- * (migration 0020) into the BigQuery table `account_daily`, replacing the
- * whole table in one load job, so a failed run leaves the previous table.
- * The view counts new accounts per UTC day and organisation, for every day
- * from the first account's through yesterday's, and the rows hold no email
- * address or user id.
+ * (migrations 0020 and 0021) into the BigQuery table `account_daily`,
+ * replacing the whole table in one load job, so a failed run leaves the
+ * previous table. The view counts new accounts per UTC day and
+ * organisation, for every day from the first account's through
+ * yesterday's, and the rows hold no email address or user id.
  */
 
 /** One `account_daily` row. */
@@ -45,15 +45,48 @@ export const ACCOUNT_SCHEMA: TableField[] = [
     },
 ];
 
-/** Every row of the view, oldest day first. */
-export const readAccountDaily = async (pool: Pool): Promise<AccountRow[]> => {
-    const { rows } = await pool.query<AccountRow>(
-        `SELECT to_char(day, 'YYYY-MM-DD') AS day, organisation, accounts
-         FROM account_daily
-         ORDER BY day, organisation`,
-    );
+/**
+ * What one read of the view returns. `accounts` and `webmailDomains` come
+ * from the view `account_daily_source`: whether the user table holds any
+ * account, and whether `webmail_domain` holds any domain.
+ */
+export type AccountRead = {
+    rows: AccountRow[];
+    accounts: boolean;
+    webmailDomains: boolean;
+};
 
-    return rows;
+/**
+ * Every row of the view, oldest day first, with the facts that say why it
+ * holds none. One read-only transaction reads both, so they share one
+ * `now()` and one snapshot.
+ */
+export const readAccountDaily = async (pool: Pool): Promise<AccountRead> => {
+    const client = await pool.connect();
+
+    try {
+        await client.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');
+
+        const { rows } = await client.query<AccountRow>(
+            `SELECT to_char(day, 'YYYY-MM-DD') AS day, organisation, accounts
+             FROM account_daily
+             ORDER BY day, organisation`,
+        );
+        const { rows: source } = await client.query<{
+            accounts: boolean;
+            webmailDomains: boolean;
+        }>(`SELECT accounts, webmail_domains AS "webmailDomains" FROM account_daily_source`);
+
+        await client.query('COMMIT');
+        client.release();
+
+        return { rows, ...source[0]! };
+    } catch (error) {
+        // A failed query leaves the transaction open, and a release with
+        // the error discards the connection and the transaction with it.
+        client.release(error instanceof Error ? error : true);
+        throw error;
+    }
 };
 
 /** Replaces the BigQuery table `account_daily` with the rows in `file`, one JSON object a line. */
@@ -74,26 +107,35 @@ export const bigQueryAccountTable =
 
 /**
  * One account export run. Writes every row to `scratchFile`, then replaces
- * the table with it. The view holds a row for every day since the first
- * account's, so a read with no rows is a fault (a wrong database, an
- * emptied table) unless every account is from today, and rejects with the
- * table kept.
+ * the table with it. The view holds a row for every day from the first
+ * account's through yesterday's while `webmail_domain` holds any domain,
+ * so a read with no rows means one of three things. While `webmail_domain`
+ * is empty, or while the user table holds no account (a wrong database, an
+ * emptied table), the run rejects with the table kept. When every account
+ * is from today, the table is replaced with no rows, so an account deleted
+ * since the last run, and the organisation it named, leave the export.
  */
 export const runAccountExport = async ({
     read,
     replace,
     scratchFile,
     log,
 }: {
-    read: () => Promise<AccountRow[]>;
+    read: () => Promise<AccountRead>;
     replace: (file: string) => Promise<void>;
     scratchFile: string;
     log: (line: string) => void;
 }): Promise<void> => {
-    const rows = await read();
+    const { rows, accounts, webmailDomains } = await read();
+
+    if (rows.length === 0 && !webmailDomains) {
+        throw new Error(
+            'account export: account_daily read no rows, webmail_domain is empty, table kept',
+        );
+    }
 
-    if (rows.length === 0) {
-        throw new Error('account export: account_daily read no rows, table kept');
+    if (rows.length === 0 && !accounts) {
+        throw new Error('account export: account_daily read no accounts, table kept');
     }
 
     try {

db/connection.ts

@@ -97,20 +97,20 @@ type EnvLike = Record<string, string | undefined>;
 /**
  * Non-throwing check for which pool kind the runtime should build. Shared
  * by `resolveRuntimeConnection` below (full validation, safe to throw --
- * called from instrumentation.ts's `register()` at server start) and
- * src/lib/db/index.ts (must never throw at import, including during
- * `next build` with no env at all).
+ * called at process start) and src/lib/db/index.ts (must never throw at
+ * import, including during `next build` with no env at all).
  */
 export function isCloudSqlConfigured(env: EnvLike = process.env): boolean {
     return !!env.WEBSITE_DB_INSTANCE_CONNECTION_NAME;
 }
 
 /**
  * Resolves which database the runtime pool should connect to. Cloud SQL
- * (deployed services) wins whenever its instance var is set; a local
- * Postgres via DATABASE_URL is the fallback. Throws a clear error naming
- * the missing variable(s) otherwise -- safe here because this is only
- * called from instrumentation.ts's `register()` at server start, never at
+ * (deployed services and the event export Job) wins whenever its instance
+ * var is set; a local Postgres via DATABASE_URL is the fallback. Throws a
+ * clear error naming the missing variable(s) otherwise -- safe here
+ * because its callers, instrumentation.ts's `register()` at server start
+ * and db/export-events.ts's `main()`, call it at process start, never at
  * module import time.
  */
 export function resolveRuntimeConnection(env: EnvLike = process.env): RuntimeConnection {
@@ -145,10 +145,11 @@ export function resolveRuntimeConnection(env: EnvLike = process.env): RuntimeCon
  * Opens one `Connector` for the runtime pool's lifetime and resolves the
  * stream factory pg's `Pool` uses to dial Cloud SQL over PSC with IAM auth
  * -- the same path `buildMigratorPool` uses for the migrator role, just
- * authenticating as the runtime IAM database user instead. Called once
- * from instrumentation.ts's `register()`; the `Connector` is intentionally
- * never closed here (SIGTERM exits the process, same as the migrator's
- * shutdown note above but there is no request to await first).
+ * authenticating as the connection's IAM database user instead. Called
+ * once per process, from instrumentation.ts's `register()` and from
+ * db/export-events.ts's `main()`. The caller owns the `Connector`: the
+ * server leaves it open until SIGTERM exits the process, and the export
+ * Job closes it after ending its pool.
  */
 export async function connectCloudSqlRuntime(
     connection: CloudSqlRuntimeConnection,
@@ -170,18 +171,19 @@ interface CloudSqlRuntimeGlobal {
 const globalForCloudSqlRuntime = globalThis as unknown as CloudSqlRuntimeGlobal;
 
 /**
- * Stores the stream factory `instrumentation.ts`'s `register()` resolved,
- * so src/lib/db/index.ts's pool can read it lazily at connect time without
- * importing the connector itself (index.ts must stay import-safe during
- * `next build`, and the connector is only ever constructed once, in
- * `register()`).
+ * Stores the stream factory `connectCloudSqlRuntime` resolved, so a pool
+ * from `buildRuntimePool` (src/lib/db/index.ts's, the export Job's) can
+ * read it lazily at connect time without importing the connector itself
+ * (index.ts must stay import-safe during `next build`, and the connector
+ * is constructed once per process, in `register()` or the export Job's
+ * `main()`).
  */
 export function setCloudSqlRuntimeStream(stream: DriverOptions['stream']): void {
     globalForCloudSqlRuntime.cloudSqlRuntimeStream = stream;
 }
 
 /**
- * Reads the stream factory `register()` stored. Throws only when actually
+ * Reads the stored stream factory. Throws only when actually
  * called -- i.e. when pg attempts a connection -- never at import, so a
  * misconfigured deploy still builds and starts; it just fails the first
  * request that touches the database.
@@ -200,23 +202,23 @@ export function getCloudSqlRuntimeStream(): DriverOptions['stream'] {
 }
 
 /**
- * Builds the one runtime `pg.Pool`, shared by src/lib/db/index.ts and
- * db/test/runtime-pool.test.ts so both exercise the same construction
- * path. Picks the pool kind from `isCloudSqlConfigured` alone -- a
+ * Builds a runtime `pg.Pool`, shared by src/lib/db/index.ts, the export
+ * Job (db/export-events.ts) and db/test/runtime-pool.test.ts so all
+ * exercise the same construction path. Picks the pool kind from `isCloudSqlConfigured` alone -- a
  * non-throwing check -- rather than the full, throwing
  * `resolveRuntimeConnection`, because this must never throw when called at
  * module import time (`next build` runs with no DB env at all):
  *
  * - Cloud SQL: `stream` is a function pg only calls when it actually opens
- *   a connection. It reads the factory `instrumentation.ts`'s `register()`
- *   stored on globalThis at that point, not now -- a clear error surfaces
- *   on the first connection attempt if `register()` never ran, not at
- *   import or build.
+ *   a connection. It reads the factory `setCloudSqlRuntimeStream` stored
+ *   on globalThis at that point, not now -- a clear error surfaces on the
+ *   first connection attempt if none was stored, not at import or build.
  * - URL: `connectionString` may be `undefined` here (no env at build); pg
  *   only fails to connect with it later.
  *
- * `sizing` overrides the pool size and adds timeouts, for a pool that must
- * give up fast (src/lib/db/index.ts's counter pool).
+ * `sizing` overrides the pool size and adds timeouts: src/lib/db/index.ts's
+ * counter pool gives up fast, and the export Job's pool holds one
+ * connection with a 5-minute `statement_timeout`.
  *
  * Every pool gets an `'error'` listener. pg-pool emits `'error'` when an
  * idle client fails (Cloud SQL or a proxy dropping the connection), and an