From 666bb511550307e16267c8b67ab0400a86847e17 Mon Sep 17 00:00:00 2001 From: Jorge Rivera Date: Sat, 24 Oct 2026 14:06:41 +0200 Subject: [PATCH] fix(events): keep three exported accounts behind every named organisation account_daily counts only accounts created before today (UTC), the accounts the export holds, so a named domain always shows at least 3 of them. The account export fails on an empty user table, replaces the table with no rows when every account is from today, and fails on an empty webmail_domain. Each exported event row carries export_rules, a digest of the export rules, ROW_SHAPE_VERSION and an HMAC under the hashing key, and every day holding an older digest reloads. A day that fails no longer blocks later days, the run stops starting days after a 20-minute budget, and its error names the failed, short, and unreached days. --- db/account-export.ts | 84 +- db/connection.ts | 54 +- db/event-export.ts | 348 +++++- db/export-events.ts | 64 +- .../0021_account_daily_before_today.sql | 44 + db/migrations/meta/0021_snapshot.json | 1092 +++++++++++++++++ db/migrations/meta/_journal.json | 7 + db/test/account-export.test.ts | 106 +- db/test/event-export.test.ts | 359 +++++- deploy/README.md | 109 +- 10 files changed, 1991 insertions(+), 276 deletions(-) create mode 100644 db/migrations/0021_account_daily_before_today.sql create mode 100644 db/migrations/meta/0021_snapshot.json diff --git a/db/account-export.ts b/db/account-export.ts index d48ff79..01bf132 100644 --- a/db/account-export.ts +++ b/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 => { - const { rows } = await pool.query( - `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 => { + const client = await pool.connect(); + + try { + await client.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY'); + + const { rows } = await client.query( + `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,10 +107,13 @@ 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, @@ -85,15 +121,21 @@ export const runAccountExport = async ({ scratchFile, log, }: { - read: () => Promise; + read: () => Promise; replace: (file: string) => Promise; scratchFile: string; log: (line: string) => void; }): Promise => { - 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 { diff --git a/db/connection.ts b/db/connection.ts index faf336f..43c46c4 100644 --- a/db/connection.ts +++ b/db/connection.ts @@ -97,9 +97,8 @@ type EnvLike = Record; /** * 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; @@ -107,10 +106,11 @@ export function isCloudSqlConfigured(env: EnvLike = process.env): boolean { /** * 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 diff --git a/db/event-export.ts b/db/event-export.ts index 756e809..7b49a38 100644 --- a/db/event-export.ts +++ b/db/event-export.ts @@ -1,10 +1,12 @@ -import { createHmac } from 'node:crypto'; +import { createHash, createHmac } from 'node:crypto'; import fs from 'node:fs/promises'; import { BigQuery, type TableField } from '@google-cloud/bigquery'; import type { Pool } from 'pg'; import type { GalleryKind } from '../src/components/organisms/Gallery/galleryData.ts'; +import type { CardTarget } from '../src/components/organisms/Studio/hooks/useChat.ts'; +import type { ShareMode } from '../src/lib/chat/shareMode.ts'; import type { ShareKind } from '../src/lib/chat/sharePrune.ts'; import type { Widget } from '../src/lib/chat/widgets.ts'; import type { EventName } from '../src/lib/events/track.ts'; @@ -25,9 +27,10 @@ import { requiredEnv } from './connection.ts'; * BigQuery table `event`, one day partition per UTC day of `created_at`. * Each run replaces every day from the latest day in BigQuery minus one * through today, every day from the first event on when the table is - * empty, and every earlier day whose event count differs from its - * partition's row count. A day is read whole before its partition is - * replaced, so a failed read leaves BigQuery as it was. + * empty, every earlier day whose event count differs from its partition's + * row count, and every day holding a row exported under other rules + * (exportRules). A day is read whole before its partition is replaced, so + * a failed read leaves BigQuery as it was. * * Rows keep no raw ids and no free text. `visitor_id` and `user_id` become * HMAC-SHA256 digests under EXPORT_ID_HMAC_KEY, as does the `shareId` prop, @@ -96,7 +99,10 @@ const ANSWER_KINDS = every()(['resolution', 'catalogue' const DECLINED_REASONS = every>()(['out_of_scope', 'requires_analysis']); const ABANDON_REASONS = every()(['hidden', 'superseded', 'reset', 'left']); const SHARE_KINDS = every()(['conversation', 'canvas', 'card', 'search']); -const SHARE_MODES = every()(['live', 'snapshot']); +// share.create and share.open send Studio's ShareMode, search.action the search's. +const SHARE_MODES = every()(['live', 'snapshot']); +// A prompt with no card target is `new` on the first turn. +const PROMPT_INTENTS = every<'new' | CardTarget['intent']>()(['new', 'follow_up', 'edit']); const CARD_KINDS = every()(['key_number', 'chart', 'table', 'text']); const GALLERY_KINDS = every()(['conversation', 'canvas', 'card', 'dataset']); const DATASET_FORMATS = every()(['csv', 'xlsx', 'parquet']); @@ -144,6 +150,7 @@ export const PROP_RULES = { 'search.action': { searchId: 'token', traceId: 'token', + // DataSearch.tsx's SearchAction, which src/ does not export. action: ['pick_reading', 'download', 'share'], partIndex: 'count', specId: 'token', @@ -154,7 +161,7 @@ export const PROP_RULES = { }, 'studio.prompt': { turn: 'count', - intent: ['new', 'follow_up', 'edit'], + intent: PROMPT_INTENTS, length: 'count', guest: 'boolean', }, @@ -163,6 +170,7 @@ export const PROP_RULES = { 'studio.session_end': { turns: 'count', cards: 'count', pins: 'count', guest: 'boolean' }, 'auth.signin': { provider: 'token' }, 'guest.wall_hit': { prompts: 'count' }, + // Literals in src/app/api/feedback/route.ts, without a named type. 'feedback.submit': { kind: ['chat', 'widget'] }, 'embed.view': { shareId: 'digest', kind: SHARE_KINDS, host: 'token' }, 'share.create': { shareId: 'digest', kind: SHARE_KINDS, mode: SHARE_MODES }, @@ -170,6 +178,8 @@ export const PROP_RULES = { 'gallery.open': { slug: 'token', kind: GALLERY_KINDS }, } satisfies Record>; +// A change to digest, passes, exportProps, exportPage or exportRow bumps +// ROW_SHAPE_VERSION. const digest = (key: string, id: string | null): string | null => id === null ? null : createHmac('sha256', key).update(id).digest('hex'); @@ -187,7 +197,10 @@ const passes = (rule: PropRule, value: unknown): boolean => { } }; -/** The props an event keeps. A name the site does not emit keeps none. */ +/** + * The props an event keeps. A name the site does not emit keeps none. A + * change here bumps ROW_SHAPE_VERSION. + */ export const exportProps = ( name: string, props: Record, @@ -229,9 +242,18 @@ export type ExportRow = { visitor_hash: string | null; user_hash: string | null; props: Record; + export_rules: string; }; -export const EXPORT_SCHEMA: TableField[] = [ +// The columns the table was created with that are REQUIRED. A load's +// ALLOW_FIELD_ADDITION adds NULLABLE columns only, and a load whose schema +// adds a REQUIRED one fails, so a column added later must be NULLABLE and +// tsc refuses a REQUIRED one with another name. +type CreatedRequired = 'id' | 'created_at' | 'name' | 'signed_in' | 'props'; + +export type ExportField = TableField & ({ name: CreatedRequired } | { mode?: 'NULLABLE' }); + +export const EXPORT_SCHEMA: ExportField[] = [ { name: 'id', type: 'INT64', mode: 'REQUIRED' }, { name: 'created_at', type: 'TIMESTAMP', mode: 'REQUIRED' }, { name: 'name', type: 'STRING', mode: 'REQUIRED' }, @@ -240,6 +262,7 @@ export const EXPORT_SCHEMA: TableField[] = [ { name: 'visitor_hash', type: 'STRING' }, { name: 'user_hash', type: 'STRING' }, { name: 'props', type: 'JSON', mode: 'REQUIRED' }, + { name: 'export_rules', type: 'STRING' }, ]; // A path on the site: a slash, then letters, digits and `_ - . : /`, up to @@ -256,15 +279,51 @@ const SHARE_PATH = /^\/(s|embed)\/[^/]+/; /** * The page's path without its query string or fragment and with any share - * id written `:id`, null unless it is a site path. + * id written `:id`, null unless it is a site path. A change here bumps + * ROW_SHAPE_VERSION. */ +// The query-string strip repeats eventLogEntry's in src/lib/events/eventLog.ts, +// which keeps it inline and exports no function for it. const exportPage = (page: string | null): string | null => { const path = page?.split(/[?#]/, 1)[0]; return path !== undefined && PAGE_PATH.test(path) ? path.replace(SHARE_PATH, '/$1/:id') : null; }; -export const exportRow = (row: SourceEvent, hmacKey: string): ExportRow => ({ +// The version of the code that shapes a row: digest, passes, exportProps, +// exportPage and exportRow. Bumped by hand with any change to them, so the +// change reaches every day (exportRules). +const ROW_SHAPE_VERSION = 1; + +/** + * The digest every row carries in `export_rules`: SHA-256 over PROP_RULES, + * TOKEN, PAGE_PATH, SHARE_PATH, EXPORT_SCHEMA, ROW_SHAPE_VERSION and an + * HMAC of a fixed string under `hmacKey`, so a new key changes it too. A + * day holding a row with another digest, or none, reloads (exportDays). + */ +export const exportRules = (hmacKey: string): string => + createHash('sha256') + .update( + JSON.stringify({ + PROP_RULES, + TOKEN: TOKEN.source, + PAGE_PATH: PAGE_PATH.source, + SHARE_PATH: SHARE_PATH.source, + EXPORT_SCHEMA, + ROW_SHAPE_VERSION, + key: createHmac('sha256', hmacKey).update('export_rules').digest('hex'), + }), + ) + .digest('hex'); + +// Every prop key some event keeps, the only keys readDay fetches. +const KEPT_PROPS = [...new Set(Object.values(PROP_RULES).flatMap((rules) => Object.keys(rules)))]; + +/** + * The row BigQuery holds for an event, under `rules` from exportRules. A + * change here bumps ROW_SHAPE_VERSION. + */ +export const exportRow = (row: SourceEvent, hmacKey: string, rules: string): ExportRow => ({ id: row.id, created_at: row.createdAt, name: row.name, @@ -274,10 +333,24 @@ export const exportRow = (row: SourceEvent, hmacKey: string): ExportRow => ({ visitor_hash: digest(hmacKey, row.visitorId), user_hash: digest(hmacKey, row.userId), props: exportProps(row.name, row.props, hmacKey), + export_rules: rules, }); const DAY_MS = 86_400_000; +/** + * How long a run starts new days for, 20 minutes, well inside the Job's + * 1800 s task timeout, so a run that has many days each waiting out a + * statement timeout ends and reports before the platform kills it. + */ +export const EXPORT_BUDGET_MS = 20 * 60_000; + +/** The most one BigQuery query may bill (`maximumBytesBilled`). */ +const MAX_BYTES_BILLED = 1e9; + +/** What the staleDays scan bills per event row, from its two columns. */ +const STALE_BYTES_PER_ROW = 74; + const dayOf = (ms: number): string => new Date(ms).toISOString().slice(0, 10); /** Rows per UTC day (`YYYY-MM-DD`). A day without rows is absent. */ @@ -287,23 +360,26 @@ export type DayCounts = ReadonlyMap; * The UTC days a run replaces, oldest first. They run from the day before * the latest day BigQuery holds through `today`, or from the first event's * day when BigQuery holds nothing, and add every day whose `events` count - * differs from its `partitions` count. A day that loaded short thus stays - * in every run until it matches, however many later days load. None when - * there is no event and BigQuery holds nothing. + * differs from its `partitions` count and every `stale` day. A day that + * loaded short, or failed to load under new rules, thus stays in every run + * until it matches, however many later days load. None when there is no + * event and BigQuery holds nothing. */ export const exportDays = ({ today, events, partitions, + stale = new Set(), }: { today: string; events: DayCounts; partitions: DayCounts; + stale?: ReadonlySet; }): string[] => { const latest = [...partitions.keys()].sort().at(-1); const start = latest !== undefined ? dayOf(Date.parse(latest) - DAY_MS) : [...events.keys()].sort()[0]; - const days = new Set(); + const days = new Set(stale); if (start !== undefined) { for (let ms = Date.parse(start); ms <= Date.parse(today); ms += DAY_MS) { @@ -324,7 +400,7 @@ export const exportDays = ({ export type EventSource = { /** The events per UTC day of `created_at`, over the whole table. */ dayCounts(): Promise; - /** One UTC day's events in id order, a page at a time. */ + /** One UTC day's events in `created_at`, then id, order, a page at a time. */ readDay(day: string): AsyncIterable; }; @@ -334,6 +410,8 @@ export type Warehouse = { ensureTable(): Promise; /** The rows each day's partition holds. */ dayCounts(): Promise; + /** The days holding a row whose `export_rules` is not `rules`, NULL included. */ + staleDays(rules: string): Promise>; /** Replaces the day's partition with the rows in `file`, one JSON object a line. */ replaceDay(day: string, file: string): Promise; }; @@ -344,30 +422,36 @@ export const pgEventSource = (pool: Pool, pageSize = PAGE_SIZE): EventSource => async dayCounts() { // The UTC date of created_at, the same day readDay's range bounds. const { rows } = await pool.query<{ day: string; n: number }>( - `SELECT to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD') AS day, - count(*)::int AS n - FROM event - GROUP BY 1`, + `SELECT to_char(day, 'YYYY-MM-DD') AS day, n + FROM (SELECT (created_at AT TIME ZONE 'UTC')::date AS day, count(*)::int AS n + FROM event + GROUP BY 1) AS days`, ); return new Map(rows.map(({ day, n }) => [day, n])); }, async *readDay(day) { - const from = `${day}T00:00:00Z`; const to = `${dayOf(Date.parse(day) + DAY_MS)}T00:00:00Z`; - let after = 0; + // Keyset paging: each page starts after the last row of the one + // before, so no page reads the rows before it. Ids start at 1, so + // (midnight, 0) precedes the day's first row. `props` keeps only the keys some + // event keeps; exportProps picks each event's own. + let after = { createdAt: `${day}T00:00:00Z`, id: 0 }; for (;;) { const { rows } = await pool.query( `SELECT id::float8 AS id, to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.US"Z"') AS "createdAt", - name, user_id AS "userId", visitor_id AS "visitorId", page, props + name, user_id AS "userId", visitor_id AS "visitorId", page, + coalesce((SELECT jsonb_object_agg(key, value) + FROM jsonb_each(props) + WHERE key = ANY($4::text[])), '{}') AS props FROM event - WHERE created_at >= $1 AND created_at < $2 AND id > $3 - ORDER BY id - LIMIT $4`, - [from, to, after, pageSize], + WHERE created_at >= $1 AND created_at < $2 AND (created_at, id) > ($1, $3) + ORDER BY created_at, id + LIMIT $5`, + [after.createdAt, to, after.id, KEPT_PROPS, pageSize], ); if (rows.length === 0) { @@ -375,7 +459,7 @@ export const pgEventSource = (pool: Pool, pageSize = PAGE_SIZE): EventSource => } yield rows; - after = rows[rows.length - 1]!.id; + after = rows[rows.length - 1]!; } }, }); @@ -392,6 +476,51 @@ export const bigQueryWarehouse = ({ const dataset = bigquery.dataset(datasetId); const table = dataset.table('event'); + // A query job fails past 1 GB billed or past 2 minutes, well inside the + // Job's 1800 s task timeout, and the run fails with it. + const query = async (sql: string, params?: Record): Promise => { + try { + const [rows] = await bigquery.query({ + query: sql, + params, + maximumBytesBilled: String(MAX_BYTES_BILLED), + jobTimeoutMs: 120_000, + }); + + return rows; + } catch (error) { + if (/bytes billed/i.test(messageOf(error))) { + throw new Error( + `a query exceeded the ${MAX_BYTES_BILLED / 1e9} GB maximumBytesBilled cap, ` + + `which staleDays reaches at about ${(MAX_BYTES_BILLED / STALE_BYTES_PER_ROW / 1e6).toFixed(1)} million event rows`, + { cause: error }, + ); + } + + throw error; + } + }; + + const dayCounts = async (): Promise => { + // Each partition's row count from the table's metadata, which a + // finished load updates, so the query scans no table data. + // partition_id is the UTC day as YYYYMMDD, and the pattern leaves + // out the special partitions such as __NULL__. + const rows = await query<{ partition_id: string; total_rows: unknown }>( + `SELECT partition_id, total_rows + FROM \`${projectId}.${datasetId}.INFORMATION_SCHEMA.PARTITIONS\` + WHERE table_name = 'event' AND total_rows > 0 + AND REGEXP_CONTAINS(partition_id, r'^\\d{8}$')`, + ); + + return new Map( + rows.map((row) => [ + row.partition_id.replace(/^(\d{4})(\d{2})(\d{2})$/, '$1-$2-$3'), + Number(row.total_rows), + ]), + ); + }; + return { async ensureTable() { const [exists] = await table.exists(); @@ -403,23 +532,28 @@ export const bigQueryWarehouse = ({ }); } }, - async dayCounts() { - // An exact count that sees every finished load. It scans the - // created_at column alone, 8 bytes a row. DATE() takes the UTC - // date, the day the table partitions on. The job fails past - // 1 GB billed, 125 million rows, or past 2 minutes, well inside - // the Job's 1800 s task timeout, and the run fails with it. - const [rows] = await bigquery.query({ - query: `SELECT FORMAT_DATE('%F', DATE(created_at)) AS day, COUNT(*) AS n - FROM \`${projectId}.${datasetId}.event\` - GROUP BY day`, - maximumBytesBilled: String(1e9), - jobTimeoutMs: 120_000, - }); + dayCounts, + async staleDays(rules) { + const [metadata] = await table.getMetadata(); + const fields: TableField[] = metadata.schema?.fields ?? []; + + // A table created without the column holds no digest, so every + // day with rows is stale. The next load adds the column. + if (!fields.some((field) => field.name === 'export_rules')) { + return new Set((await dayCounts()).keys()); + } - return new Map( - rows.map((row: { day: string; n: unknown }) => [row.day, Number(row.n)]), + // Scans the created_at and export_rules columns, about + // STALE_BYTES_PER_ROW bytes a row. + const rows = await query<{ day: string }>( + `SELECT FORMAT_DATE('%F', DATE(created_at)) AS day + FROM \`${projectId}.${datasetId}.event\` + WHERE export_rules IS DISTINCT FROM @rules + GROUP BY day`, + { rules }, ); + + return new Set(rows.map(({ day }) => day)); }, async replaceDay(day, file) { // Resolves once the load job is done, and throws if it failed. @@ -427,30 +561,43 @@ export const bigQueryWarehouse = ({ sourceFormat: 'NEWLINE_DELIMITED_JSON', writeDisposition: 'WRITE_TRUNCATE', schema: { fields: EXPORT_SCHEMA }, - // A column added to EXPORT_SCHEMA reaches the table on the next load. + // A NULLABLE column added to EXPORT_SCHEMA reaches the table + // on the next load (ExportField). schemaUpdateOptions: ['ALLOW_FIELD_ADDITION'], }); }, }; }; +const messageOf = (error: unknown): string => + error instanceof Error ? error.message : String(error); + /** * One export run over the days exportDays picks. Each day is written to - * `scratchFile` page by page, so memory holds one page. The partition is - * replaced only once the day has been read to its end, and only when the - * read holds at least as many rows as the partition. Nothing deletes - * `event` rows, and they carry no foreign key to `user`, so deleting an - * account leaves them. A shorter read is therefore a fault (a wrong - * database, an emptied table, a lagging replica). Such a day keeps its - * partition, the other days still load, and the run then rejects naming - * each short day with both counts. Its counts still differ, so the day - * stays in every later run and fails it until it matches. A read below - * the day's source count from `source.dayCounts()` is a fault too, - * whatever the partition holds: the day loads by the rule above, and the - * run then rejects naming it with both counts. Today's count is taken - * before its read, so a read above the count is normal. A day that reads - * no rows and holds none loads nothing. Logs each day as it is done, - * names the day in a failure, and returns the rows read per day. + * `scratchFile` page by page before its load, so memory holds one page of + * rows plus the day's file, since Cloud Run's disk is in memory too. A + * day's NDJSON runs to about 360 bytes an event. The partition is replaced + * only once the day has been read to its end, and only when the read holds + * at least as many rows as the partition. Nothing deletes `event` rows, + * and they carry no foreign key to `user`, so deleting an account leaves + * them. A shorter read is therefore a fault (a wrong database, an emptied + * table, a lagging replica). Such a day keeps its partition, the other + * days still load, and the run then rejects naming each short day with + * both counts. Its counts still differ, so the day stays in every later + * run and fails it until it matches. A read below the day's source count + * from `source.dayCounts()` is a fault too, whatever the partition holds: + * the day loads by the rule above, and the run then rejects naming it with + * both counts. Today's count is taken before its read, so a read above the + * count is normal. A day that reads no rows and holds none loads nothing. + * A day whose read or load fails keeps its partition as it was, the later + * days still load, and the run then rejects naming every such day with its + * error. Each row carries the run's exportRules digest, so a day that + * fails under new rules still holds the old digest and is the one day the + * next run reloads for them. A run starts no new day once `budgetMs` has + * passed since it began (`now` is its clock); it ends the day in hand, then + * rejects naming the days it did not reach as well as the failed and short + * ones, and the next run picks those days up. Logs each day as it is done + * and returns the rows read per day. */ export const runExport = async ({ source, @@ -459,6 +606,8 @@ export const runExport = async ({ today, scratchFile, log, + budgetMs = EXPORT_BUDGET_MS, + now = Date.now, }: { source: EventSource; warehouse: Warehouse; @@ -466,17 +615,32 @@ export const runExport = async ({ today: string; scratchFile: string; log: (line: string) => void; + budgetMs?: number; + now?: () => number; }): Promise<{ day: string; rows: number }[]> => { + const startedAt = now(); + const rules = exportRules(hmacKey); + await warehouse.ensureTable(); - const [events, partitions] = await Promise.all([source.dayCounts(), warehouse.dayCounts()]); - const days = exportDays({ today, events, partitions }); + const [events, partitions, stale] = await Promise.all([ + source.dayCounts(), + warehouse.dayCounts(), + warehouse.staleDays(rules), + ]); + const days = exportDays({ today, events, partitions, stale }); const read: { day: string; rows: number }[] = []; + const failed: { day: string; error: unknown }[] = []; const short: string[] = []; - const shortDays = () => `short days: ${short.join(', ')}`; + const unreached: string[] = []; try { for (const day of days) { + if (now() - startedAt >= budgetMs) { + unreached.push(day); + continue; + } + let rows = 0; const held = partitions.get(day) ?? 0; const counted = events.get(day) ?? 0; @@ -485,7 +649,9 @@ export const runExport = async ({ await fs.writeFile(scratchFile, ''); for await (const page of source.readDay(day)) { - const lines = page.map((row) => `${JSON.stringify(exportRow(row, hmacKey))}\n`); + const lines = page.map( + (row) => `${JSON.stringify(exportRow(row, hmacKey, rules))}\n`, + ); await fs.appendFile(scratchFile, lines.join('')); rows += page.length; @@ -495,10 +661,9 @@ export const runExport = async ({ await warehouse.replaceDay(day, scratchFile); } } catch (error) { - const message = error instanceof Error ? error.message : String(error); - const earlier = short.length > 0 ? `; earlier days ${shortDays()}` : ''; - - throw new Error(`event export ${day}: ${message}${earlier}`, { cause: error }); + log(`${day}: failed, partition kept: ${messageOf(error)}`); + failed.push({ day, error }); + continue; } const belowSource = rows < counted ? `, source counts ${counted}` : ''; @@ -523,13 +688,60 @@ export const runExport = async ({ await fs.rm(scratchFile, { force: true }); } - if (short.length > 0) { - throw new Error(`event export ${shortDays()}`); + const faults = [ + ...failed.map(({ day, error }) => `${day}: ${messageOf(error)}`), + ...(short.length > 0 ? [`short days: ${short.join(', ')}`] : []), + ...(unreached.length > 0 + ? [ + `days not reached in the ${budgetMs / 60_000} minute budget: ${unreached.join(', ')}`, + ] + : []), + ]; + + if (faults.length > 0) { + throw new Error(`event export ${faults.join('; ')}`, { + cause: + failed.length > 0 + ? new AggregateError(failed.map(({ error }) => error)) + : undefined, + }); } return read; }; +/** + * The export Job's run (db/export-events.ts): the account export, then the + * event export, which runs whether or not the account export failed. Logs + * an account export failure as it happens and rejects with it once the + * event export is done, so a failing one leaves the other's tables + * current. An event export failure rejects as it is. + */ +export const runExportJob = async ({ + exportAccounts, + exportEvents, + logError, +}: { + exportAccounts: () => Promise; + exportEvents: () => Promise; + logError: (error: unknown) => void; +}): Promise => { + let accountFailure: Error | null = null; + + try { + await exportAccounts(); + } catch (error) { + accountFailure = new Error('account export failed', { cause: error }); + logError(error); + } + + await exportEvents(); + + if (accountFailure !== null) { + throw accountFailure; + } +}; + /** The export's own settings. The database settings are the runtime's (db/connection.ts). */ export const readExportEnv = () => ({ projectId: requiredEnv('EXPORT_BQ_PROJECT'), diff --git a/db/export-events.ts b/db/export-events.ts index b9950e3..aaabd3e 100644 --- a/db/export-events.ts +++ b/db/export-events.ts @@ -8,13 +8,18 @@ import { resolveRuntimeConnection, setCloudSqlRuntimeStream, } from './connection'; -import { bigQueryWarehouse, pgEventSource, readExportEnv, runExport } from './event-export'; +import { + bigQueryWarehouse, + pgEventSource, + readExportEnv, + runExport, + runExportJob, +} from './event-export'; // The event export Job's command (`node dist/export-events.cjs`, see // deploy/Dockerfile). It reads the database as the runtime does, through -// WEBSITE_DB_* over the Cloud SQL connector, or DATABASE_URL locally. It -// runs the account export, then the event export, and fails when either -// fails, so a failing one leaves the other's tables current. +// WEBSITE_DB_* over the Cloud SQL connector, or DATABASE_URL locally, and +// runs the account export, then the event export (runExportJob). async function main(): Promise { const settings = readExportEnv(); const connection = resolveRuntimeConnection(); @@ -27,40 +32,37 @@ async function main(): Promise { // One connection suffices, since the run sends one query at a time. A // query that runs past 5 minutes, the per-day count over the whole - // table or a page read, fails the run, well inside the Job's 1800 s - // task timeout, so a hung read ends before the next hourly run starts. + // table, a page read or an account read, fails that read. The event + // export starts no new day after 20 minutes (EXPORT_BUDGET_MS), so + // days that each wait out the timeout still end the run, with a + // report naming the days it did not reach, inside the Job's 1800 s + // task timeout and before the next hourly run starts. const pool = buildRuntimePool(process.env, { max: 1, statement_timeout: 300_000 }); const bigquery = new BigQuery({ projectId: settings.projectId }); const log = (line: string) => console.log(line); try { - let accountFailure: Error | null = null; - - try { - await runAccountExport({ - read: () => readAccountDaily(pool), - replace: bigQueryAccountTable({ datasetId: settings.datasetId, bigquery }), - scratchFile: path.join(os.tmpdir(), `account-export-${process.pid}.ndjson`), - log, - }); - } catch (error) { - accountFailure = new Error('account export failed', { cause: error }); - console.error('ERROR:', error instanceof Error ? error.stack : error); - } - - await runExport({ - source: pgEventSource(pool), - warehouse: bigQueryWarehouse({ ...settings, bigquery }), - hmacKey: settings.hmacKey, - today: new Date().toISOString().slice(0, 10), - scratchFile: path.join(os.tmpdir(), `event-export-${process.pid}.ndjson`), - log, + await runExportJob({ + exportAccounts: () => + runAccountExport({ + read: () => readAccountDaily(pool), + replace: bigQueryAccountTable({ datasetId: settings.datasetId, bigquery }), + scratchFile: path.join(os.tmpdir(), `account-export-${process.pid}.ndjson`), + log, + }), + exportEvents: () => + runExport({ + source: pgEventSource(pool), + warehouse: bigQueryWarehouse({ ...settings, bigquery }), + hmacKey: settings.hmacKey, + today: new Date().toISOString().slice(0, 10), + scratchFile: path.join(os.tmpdir(), `event-export-${process.pid}.ndjson`), + log, + }), + logError: (error) => + console.error('ERROR:', error instanceof Error ? error.stack : error), }); - - if (accountFailure !== null) { - throw accountFailure; - } } finally { // The pool first, so its connections close over the connector's tunnel. await pool.end(); diff --git a/db/migrations/0021_account_daily_before_today.sql b/db/migrations/0021_account_daily_before_today.sql new file mode 100644 index 0000000..01677f5 --- /dev/null +++ b/db/migrations/0021_account_daily_before_today.sql @@ -0,0 +1,44 @@ +-- `account_daily` holds the UTC days through yesterday's and counts only +-- the accounts created before today, UTC, so a domain it names holds at +-- least 3 of the accounts the export carries. +-- CREATE OR REPLACE keeps the view's grant to `website_export_reader`. +CREATE OR REPLACE VIEW "account_daily" AS +WITH account AS ( + SELECT "created_at"::date AS day, lower(substring("email" FROM '[^@]*$')) AS domain + FROM "user" + WHERE "created_at" < (now() AT TIME ZONE 'UTC')::date +), +labelled AS ( + SELECT day, + CASE + WHEN domain IN (SELECT "domain" FROM "webmail_domain") THEN 'Personal email' + WHEN count(*) OVER (PARTITION BY domain) >= 3 THEN domain + ELSE 'Other organisations' + END AS organisation + FROM account +), +counted AS ( + SELECT day, organisation, count(*)::int AS accounts + FROM labelled + GROUP BY day, organisation +), +days AS ( + SELECT generate_series(min(day), (now() AT TIME ZONE 'UTC')::date - 1, interval '1 day')::date AS day + FROM account + WHERE EXISTS (SELECT 1 FROM "webmail_domain") +) +SELECT days.day, counted.organisation, coalesce(counted.accounts, 0) AS accounts +FROM days +LEFT JOIN counted USING (day); +--> statement-breakpoint +-- Why `account_daily` holds no rows: whether the user table holds any +-- account, and whether `webmail_domain` holds any domain. The account +-- export reads it in the same transaction as the view, and fails on no rows +-- when either is empty. It runs with the migrator's rights, like the view, +-- and leaves no email address or user id with the reader. +CREATE VIEW "account_daily_source" AS +SELECT + EXISTS (SELECT 1 FROM "user") AS accounts, + EXISTS (SELECT 1 FROM "webmail_domain") AS webmail_domains; +--> statement-breakpoint +GRANT SELECT ON "account_daily_source" TO website_export_reader; diff --git a/db/migrations/meta/0021_snapshot.json b/db/migrations/meta/0021_snapshot.json new file mode 100644 index 0000000..ad02bdc --- /dev/null +++ b/db/migrations/meta/0021_snapshot.json @@ -0,0 +1,1092 @@ +{ + "id": "3a669c64-ea49-434a-9f75-2f090520da67", + "prevId": "0327543c-60f8-4b9d-b94f-1bff81ca8bbf", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.conversation": { + "name": "conversation", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "agent_session_id": { + "name": "agent_session_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "turn_count": { + "name": "turn_count", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "conversation_user_updated_idx": { + "name": "conversation_user_updated_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "conversation_user_id_user_id_fk": { + "name": "conversation_user_id_user_id_fk", + "tableFrom": "conversation", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.event": { + "name": "event", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "bigint", + "primaryKey": true, + "notNull": true, + "identity": { + "type": "always", + "name": "event_id_seq", + "schema": "public", + "increment": "1", + "startWith": "1", + "minValue": "1", + "maxValue": "9223372036854775807", + "cache": "1", + "cycle": false + } + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "visitor_id": { + "name": "visitor_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "page": { + "name": "page", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "props": { + "name": "props", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'{}'::jsonb" + } + }, + "indexes": { + "event_created_at_idx": { + "name": "event_created_at_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "event_name_created_at_idx": { + "name": "event_name_created_at_idx", + "columns": [ + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.feedback": { + "name": "feedback", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "bigint", + "primaryKey": true, + "notNull": true, + "identity": { + "type": "always", + "name": "feedback_id_seq", + "schema": "public", + "increment": "1", + "startWith": "1", + "minValue": "1", + "maxValue": "9223372036854775807", + "cache": "1", + "cycle": false + } + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "message": { + "name": "message", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "page": { + "name": "page", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "user_agent": { + "name": "user_agent", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "feedback_created_at_idx": { + "name": "feedback_created_at_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": false, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": { + "feedback_message_length": { + "name": "feedback_message_length", + "value": "char_length(\"feedback\".\"message\") BETWEEN 1 AND 5000" + } + }, + "isRLSEnabled": false + }, + "public.limit_counter": { + "name": "limit_counter", + "schema": "", + "columns": { + "key": { + "name": "key", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "window_start": { + "name": "window_start", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "count": { + "name": "count", + "type": "integer", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "limit_counter_window_start_idx": { + "name": "limit_counter_window_start_idx", + "columns": [ + { + "expression": "window_start", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.roadmap_vote": { + "name": "roadmap_vote", + "schema": "", + "columns": { + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "entry_id": { + "name": "entry_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "roadmap_vote_entry_idx": { + "name": "roadmap_vote_entry_idx", + "columns": [ + { + "expression": "entry_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "roadmap_vote_user_id_user_id_fk": { + "name": "roadmap_vote_user_id_user_id_fk", + "tableFrom": "roadmap_vote", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "roadmap_vote_user_id_entry_id_pk": { + "name": "roadmap_vote_user_id_entry_id_pk", + "columns": [ + "user_id", + "entry_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.share": { + "name": "share", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "mode": { + "name": "mode", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "conversation_id": { + "name": "conversation_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "card_id": { + "name": "card_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "share_user_idx": { + "name": "share_user_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "share_user_id_user_id_fk": { + "name": "share_user_id_user_id_fk", + "tableFrom": "share", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.share_card_result": { + "name": "share_card_result", + "schema": "", + "columns": { + "share_id": { + "name": "share_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "card_id": { + "name": "card_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "card": { + "name": "card", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "frame": { + "name": "frame", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "code": { + "name": "code", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "retryable": { + "name": "retryable", + "type": "boolean", + "primaryKey": false, + "notNull": false + }, + "retry_after_s": { + "name": "retry_after_s", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "checked_at": { + "name": "checked_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "fresh_until": { + "name": "fresh_until", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "lease_until": { + "name": "lease_until", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "share_card_result_share_id_share_id_fk": { + "name": "share_card_result_share_id_share_id_fk", + "tableFrom": "share_card_result", + "tableTo": "share", + "columnsFrom": [ + "share_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "share_card_result_share_id_card_id_pk": { + "name": "share_card_result_share_id_card_id_pk", + "columns": [ + "share_id", + "card_id" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.share_ref": { + "name": "share_ref", + "schema": "", + "columns": { + "share_id": { + "name": "share_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "part_index": { + "name": "part_index", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "spec_id": { + "name": "spec_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "identity": { + "name": "identity", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "entity_groups": { + "name": "entity_groups", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "variable_text": { + "name": "variable_text", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "content_version": { + "name": "content_version", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "artifact_digest": { + "name": "artifact_digest", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "operations": { + "name": "operations", + "type": "jsonb", + "primaryKey": false, + "notNull": true, + "default": "'[]'::jsonb" + }, + "assumptions": { + "name": "assumptions", + "type": "jsonb", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "share_ref_share_id_share_id_fk": { + "name": "share_ref_share_id_share_id_fk", + "tableFrom": "share_ref", + "tableTo": "share", + "columnsFrom": [ + "share_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "share_ref_share_id_part_index_pk": { + "name": "share_ref_share_id_part_index_pk", + "columns": [ + "share_id", + "part_index" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.webmail_domain": { + "name": "webmail_domain", + "schema": "", + "columns": { + "domain": { + "name": "domain", + "type": "text", + "primaryKey": true, + "notNull": true + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.account": { + "name": "account", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider_id": { + "name": "provider_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "access_token": { + "name": "access_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_token": { + "name": "refresh_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "id_token": { + "name": "id_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "access_token_expires_at": { + "name": "access_token_expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "refresh_token_expires_at": { + "name": "refresh_token_expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": false + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "password": { + "name": "password", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "account_userId_idx": { + "name": "account_userId_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "account_user_id_user_id_fk": { + "name": "account_user_id_user_id_fk", + "tableFrom": "account", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.session": { + "name": "session", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "token": { + "name": "token", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "ip_address": { + "name": "ip_address", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "user_agent": { + "name": "user_agent", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "user_id": { + "name": "user_id", + "type": "text", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "session_userId_idx": { + "name": "session_userId_idx", + "columns": [ + { + "expression": "user_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "session_user_id_user_id_fk": { + "name": "session_user_id_user_id_fk", + "tableFrom": "session", + "tableTo": "user", + "columnsFrom": [ + "user_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "session_token_unique": { + "name": "session_token_unique", + "nullsNotDistinct": false, + "columns": [ + "token" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.user": { + "name": "user", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email": { + "name": "email", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "email_verified": { + "name": "email_verified", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "image": { + "name": "image", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "user_email_unique": { + "name": "user_email_unique", + "nullsNotDistinct": false, + "columns": [ + "email" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.verification": { + "name": "verification", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "identifier": { + "name": "identifier", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "verification_identifier_idx": { + "name": "verification_identifier_idx", + "columns": [ + { + "expression": "identifier", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/db/migrations/meta/_journal.json b/db/migrations/meta/_journal.json index 4121e67..31bc3a0 100644 --- a/db/migrations/meta/_journal.json +++ b/db/migrations/meta/_journal.json @@ -141,6 +141,13 @@ "when": 1792794247981, "tag": "0020_account_daily", "breakpoints": true + }, + { + "idx": 21, + "version": "7", + "when": 1792794347981, + "tag": "0021_account_daily_before_today", + "breakpoints": true } ] } \ No newline at end of file diff --git a/db/test/account-export.test.ts b/db/test/account-export.test.ts index bfeed53..7c010f7 100644 --- a/db/test/account-export.test.ts +++ b/db/test/account-export.test.ts @@ -9,6 +9,7 @@ import freeEmailDomains from 'free-email-domains'; import { Pool } from 'pg'; import { ACCOUNT_SCHEMA, + type AccountRead, type AccountRow, bigQueryAccountTable, readAccountDaily, @@ -120,8 +121,11 @@ test('the export reader reads accounts per UTC day and organisation, and not "us options: `-c TimeZone=${hour >= 12 ? 'Etc/GMT-14' : 'Etc/GMT+12'}`, }); + const { rows, accounts, webmailDomains } = await readAccountDaily(readerPool); + + assert.deepEqual({ accounts, webmailDomains }, { accounts: true, webmailDomains: true }); assert.deepEqual( - sorted(await readAccountDaily(readerPool)), + sorted(rows), sorted([ { day: day(4), organisation: 'Personal email', accounts: 1 }, { day: day(3), organisation: 'Other organisations', accounts: 2 }, @@ -138,6 +142,29 @@ test('the export reader reads accounts per UTC day and organisation, and not "us await assert.rejects(readerPool.query('select 1 from webmail_domain'), /permission denied/); }); +test('account_daily names a domain only once it holds 3 accounts before today', async (t) => { + const database = await createTestDatabase(); + t.after(() => database.teardown()); + + await migrate(database); + await database.runtimePool().query( + `insert into "user" (id, name, email, created_at) + select email, 'Name', email, (now() at time zone 'UTC')::date - back + from unnest(array['a@three.org', 'b@three.org', 'c@three.org'], array[2, 1, 0]) + as seed (email, back)`, + ); + + const { rows } = await readAccountDaily(database.adminPool()); + + assert.deepEqual( + rows.map(({ organisation, accounts }) => [organisation, accounts]), + [ + ['Other organisations', 1], + ['Other organisations', 1], + ], + ); +}); + test('account_daily holds no rows while webmail_domain is empty, and the export fails', async (t) => { const database = await createTestDatabase(); t.after(() => database.teardown()); @@ -165,11 +192,39 @@ test('account_daily holds no rows while webmail_domain is empty, and the export scratchFile: path.join(os.tmpdir(), `account-export-test-${process.pid}.ndjson`), log: () => {}, }), - /read no rows/, + /webmail_domain is empty/, ); assert.equal(replaced, false); }); +test('account_daily_source tells accounts only from today from an empty user table', async (t) => { + const cases: [string, string[], boolean][] = [ + ['an account today', ['a@one.org'], true], + ['no account', [], false], + ]; + + for (const [label, emails, accounts] of cases) { + await t.test(label, async (t) => { + const database = await createTestDatabase(); + t.after(() => database.teardown()); + + await migrate(database); + await database.runtimePool().query( + `insert into "user" (id, name, email, created_at) + select email, 'Name', email, now() at time zone 'UTC' + from unnest($1::text[]) as email`, + [emails], + ); + + assert.deepEqual(await readAccountDaily(database.adminPool()), { + rows: [], + accounts, + webmailDomains: true, + }); + }); + } +}); + test('syncWebmailDomains rewrites webmail_domain to the package list', async (t) => { const database = await createTestDatabase(); const admin = database.adminPool(); @@ -233,7 +288,7 @@ test('runAccountExport loads every row in one file and removes it', async () => const loaded: string[] = []; await runAccountExport({ - read: async () => ROWS, + read: async () => ({ rows: ROWS, accounts: true, webmailDomains: true }), replace: async (file) => { loaded.push(await fs.readFile(file, 'utf8')); }, @@ -253,19 +308,36 @@ test('runAccountExport loads every row in one file and removes it', async () => await assert.rejects(fs.access(scratchFile)); }); -test('runAccountExport rejects a read with no rows and keeps the table', async () => { - let replaced = false; +test('runAccountExport empties the table when every account is from today, and rejects other empty reads', async (t) => { + const cases: [string, Omit, RegExp | null][] = [ + ['every account from today', { accounts: true, webmailDomains: true }, null], + ['no account', { accounts: false, webmailDomains: true }, /read no accounts/], + [ + 'an empty webmail_domain', + { accounts: true, webmailDomains: false }, + /webmail_domain is empty/, + ], + ]; - await assert.rejects( - runAccountExport({ - read: async () => [], - replace: async () => { - replaced = true; - }, - scratchFile: path.join(os.tmpdir(), `account-export-test-${process.pid}.ndjson`), - log: () => {}, - }), - /read no rows/, - ); - assert.equal(replaced, false); + for (const [label, source, rejects] of cases) { + await t.test(label, async () => { + const loaded: string[] = []; + const run = runAccountExport({ + read: async () => ({ rows: [], ...source }), + replace: async (file) => { + loaded.push(await fs.readFile(file, 'utf8')); + }, + scratchFile: path.join(os.tmpdir(), `account-export-test-${process.pid}.ndjson`), + log: () => {}, + }); + + if (rejects === null) { + await run; + assert.deepEqual(loaded, ['']); + } else { + await assert.rejects(run, rejects); + assert.deepEqual(loaded, []); + } + }); + } }); diff --git a/db/test/event-export.test.ts b/db/test/event-export.test.ts index cc3616a..cd3abf2 100644 --- a/db/test/event-export.test.ts +++ b/db/test/event-export.test.ts @@ -9,11 +9,15 @@ import { Pool } from 'pg'; import { applyMigrations } from '../apply-migrations.ts'; import { bigQueryWarehouse, + EXPORT_SCHEMA, type EventSource, exportDays, + type ExportField, exportRow, + exportRules, pgEventSource, runExport, + runExportJob, type SourceEvent, type Warehouse, } from '../event-export.ts'; @@ -22,6 +26,7 @@ import { createTestDatabase } from './harness.ts'; const MIGRATIONS_FOLDER = path.resolve(import.meta.dirname, '../migrations'); const KEY = 'test-key'; +const RULES = exportRules(KEY); const sourceEvent = (overrides: Partial): SourceEvent => ({ id: 1, @@ -61,6 +66,7 @@ test('exportRow keeps allow-listed props that pass their rule and drops the rest }, }), KEY, + RULES, ); assert.deepEqual(row, { @@ -83,9 +89,15 @@ test('exportRow keeps allow-listed props that pass their rule and drops the rest timedOut: false, kind: 'resolution', }, + export_rules: RULES, }); }); +test('exportRules changes with the HMAC key', () => { + assert.equal(exportRules(KEY), RULES); + assert.notEqual(exportRules('rotated-key'), RULES); +}); + test('exportRow drops a value its rule refuses', async (t) => { const cases: [string, Record][] = [ [ @@ -102,13 +114,13 @@ test('exportRow drops a value its rule refuses', async (t) => { for (const [label, overrides] of cases) { await t.test(label, () => { - assert.deepEqual(exportRow(sourceEvent(overrides), KEY).props, {}); + assert.deepEqual(exportRow(sourceEvent(overrides), KEY, RULES).props, {}); }); } }); test('exportRow leaves the hashes null for a guest without consent', () => { - const row = exportRow(sourceEvent({ page: '/topics/health' }), KEY); + const row = exportRow(sourceEvent({ page: '/topics/health' }), KEY, RULES); assert.equal(row.signed_in, false); assert.equal(row.visitor_hash, null); @@ -140,7 +152,7 @@ test('exportRow keeps a site path as the page and refuses other text', async (t) for (const [label, page, kept] of cases) { await t.test(label, () => { - assert.equal(exportRow(sourceEvent({ page }), KEY).page, kept); + assert.equal(exportRow(sourceEvent({ page }), KEY, RULES).page, kept); }); } }); @@ -156,12 +168,17 @@ test('exportRow exports a share id as its digest, so share events still join', ( ]; for (const [name, props] of events) { - assert.equal(exportRow(sourceEvent({ name, props }), KEY).props.shareId, hashed, name); + assert.equal( + exportRow(sourceEvent({ name, props }), KEY, RULES).props.shareId, + hashed, + name, + ); } const refused = exportRow( sourceEvent({ name: 'share.open', props: { shareId: 'not a share id' } }), KEY, + RULES, ); assert.deepEqual(refused.props, {}); @@ -203,6 +220,16 @@ test('exportDays replaces from the day before the latest loaded day through toda }, ['2026-10-22', '2026-10-23'], ], + [ + 'days holding rows under other export rules', + { + today: '2026-10-23', + events: counts({ '2026-10-17': 1, '2026-10-20': 2, '2026-10-23': 1 }), + partitions: counts({ '2026-10-17': 1, '2026-10-20': 2, '2026-10-23': 1 }), + stale: new Set(['2026-10-17', '2026-10-20']), + }, + ['2026-10-17', '2026-10-20', '2026-10-22', '2026-10-23'], + ], [ 'a month boundary', { @@ -265,8 +292,15 @@ test('exportDays adds every earlier day whose counts differ', async (t) => { } }); -/** An in-memory warehouse. Each partition is the parsed lines of its last load. */ -const fakeWarehouse = (partitions: Map) => { +/** + * An in-memory warehouse. Each partition is the parsed lines of its last + * load, and a day is stale while any of its rows carries an `export_rules` + * other than the run's. A load of a day in `failLoad` throws. + */ +const fakeWarehouse = ( + partitions: Map, + { failLoad = [] }: { failLoad?: string[] } = {}, +) => { const calls: string[] = []; const warehouse: Warehouse = { async ensureTable() { @@ -275,7 +309,22 @@ const fakeWarehouse = (partitions: Map) => { async dayCounts() { return new Map([...partitions].map(([day, rows]) => [day, rows.length])); }, + async staleDays(rules) { + return new Set( + [...partitions] + .filter(([, rows]) => + rows.some( + (row) => (row as { export_rules?: string }).export_rules !== rules, + ), + ) + .map(([day]) => day), + ); + }, async replaceDay(day, file) { + if (failLoad.includes(day)) { + throw new Error('load job failed'); + } + calls.push(`replace ${day}`); const text = await fs.readFile(file, 'utf8'); partitions.set( @@ -293,7 +342,7 @@ const fakeWarehouse = (partitions: Map) => { const fakeSource = ( pages: Record, - failOn: string | null = null, + failOn: readonly string[] = [], // Source counts that differ from the rows a day reads. counted: Record = {}, ): EventSource => ({ @@ -310,7 +359,7 @@ const fakeSource = ( yield page; } - if (day === failOn) { + if (failOn.includes(day)) { throw new Error('connection lost'); } }, @@ -346,8 +395,11 @@ test('runExport backfills an empty table day by day and logs each day', async () logged.map((line) => line.slice(0, 10)), ['2026-10-20', '2026-10-21', '2026-10-22'], ); - assert.deepEqual(partitions.get('2026-10-20'), [exportRow(first, KEY), exportRow(second, KEY)]); - assert.deepEqual(partitions.get('2026-10-22'), [exportRow(third, KEY)]); + assert.deepEqual(partitions.get('2026-10-20'), [ + exportRow(first, KEY, RULES), + exportRow(second, KEY, RULES), + ]); + assert.deepEqual(partitions.get('2026-10-22'), [exportRow(third, KEY, RULES)]); await assert.rejects(fs.access(scratchFile)); }); @@ -411,7 +463,7 @@ test('runExport keeps a partition that holds more rows than the day reads', asyn assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-23']); assert.equal(partitions.get('2026-10-22'), held); - assert.deepEqual(partitions.get('2026-10-23'), [exportRow(fresh, KEY)]); + assert.deepEqual(partitions.get('2026-10-23'), [exportRow(fresh, KEY, RULES)]); assert.deepEqual( logged.map((line) => line.slice(0, 10)), ['2026-10-21', '2026-10-22', '2026-10-23'], @@ -492,7 +544,7 @@ test('runExport rechecks an earlier day whose counts differ until they match', a const partitions = new Map( Object.entries(later).map(([day, dayPages]) => [ day, - dayPages.flat().map((row) => exportRow(row, KEY)), + dayPages.flat().map((row) => exportRow(row, KEY, RULES)), ]), ); @@ -502,7 +554,7 @@ test('runExport rechecks an earlier day whose counts differ until they match', a const { warehouse, calls } = fakeWarehouse(partitions); const run = runExport({ - source: fakeSource({ '2026-10-19': [read], ...later }, null, counted), + source: fakeSource({ '2026-10-19': [read], ...later }, [], counted), warehouse, hmacKey: KEY, today: '2026-10-23', @@ -519,61 +571,131 @@ test('runExport rechecks an earlier day whose counts differ until they match', a assert.deepEqual(calls, expectedCalls); assert.deepEqual( partitions.get('2026-10-23'), - later['2026-10-23']!.flat().map((row) => exportRow(row, KEY)), + later['2026-10-23']!.flat().map((row) => exportRow(row, KEY, RULES)), ); assert.deepEqual( partitions.get('2026-10-19'), calls.includes('replace 2026-10-19') - ? read.map((row) => exportRow(row, KEY)) + ? read.map((row) => exportRow(row, KEY, RULES)) : (held ?? undefined), ); }); } }); -test('runExport names the day it failed on and leaves that day as it was', async () => { +test('runExport loads the days after a failed one, then names every failed day', async () => { const kept = [{ id: 7 }]; const partitions = new Map([ ['2026-10-21', [{ id: 6 }]], ['2026-10-22', kept], ]); - const { warehouse, calls } = fakeWarehouse(partitions); - const fresh = sourceEvent({ id: 8, createdAt: '2026-10-21T10:00:00.000000Z' }); + const before = structuredClone(partitions); + const { warehouse, calls } = fakeWarehouse(partitions, { failLoad: ['2026-10-22'] }); + const fresh = sourceEvent({ id: 9, createdAt: '2026-10-23T10:00:00.000000Z' }); const scratchFile = scratch(); const logged: string[] = []; await assert.rejects( runExport({ source: fakeSource( - { '2026-10-21': [[fresh]], '2026-10-22': [[sourceEvent({ id: 9 })]] }, - '2026-10-22', + { + '2026-10-21': [[sourceEvent({ id: 6 })]], + '2026-10-22': [[sourceEvent({ id: 7 })]], + '2026-10-23': [[fresh]], + }, + ['2026-10-21'], ), warehouse, hmacKey: KEY, - today: '2026-10-22', + today: '2026-10-23', scratchFile, log: (line) => logged.push(line), }), (error: unknown) => { assert.ok(error instanceof Error); - assert.match(error.message, /^event export 2026-10-22: /); - assert.ok(error.cause instanceof Error); - assert.equal(error.cause.message, 'connection lost'); + assert.equal( + error.message, + 'event export 2026-10-21: connection lost; 2026-10-22: load job failed', + ); + assert.ok(error.cause instanceof AggregateError); + assert.deepEqual( + error.cause.errors.map((cause: Error) => cause.message), + ['connection lost', 'load job failed'], + ); return true; }, ); - assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-21']); + assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-23']); assert.deepEqual( logged.map((line) => line.slice(0, 10)), - ['2026-10-21'], + ['2026-10-21', '2026-10-22', '2026-10-23'], ); - assert.deepEqual(partitions.get('2026-10-21'), [exportRow(fresh, KEY)]); + assert.deepEqual(partitions.get('2026-10-21'), before.get('2026-10-21')); assert.equal(partitions.get('2026-10-22'), kept); + assert.deepEqual(partitions.get('2026-10-23'), [exportRow(fresh, KEY, RULES)]); await assert.rejects(fs.access(scratchFile)); }); -test('runExport names the short days read before the day it failed on', async () => { +test('runExport reloads the days holding rows under other rules, and retries a failed one alone', async () => { + const at = (id: number, day: string) => + sourceEvent({ id, createdAt: `${day}T10:00:00.000000Z` }); + const pages = Object.fromEntries( + ['2026-10-18', '2026-10-19', '2026-10-20', '2026-10-21', '2026-10-22'].map((day, i) => [ + day, + [[at(i + 1, day)]], + ]), + ); + // 18 and 20 October hold rows under old rules, 19 October under these. + const partitions = new Map( + Object.entries(pages).map(([day, dayPages]) => [ + day, + dayPages.flat().map((row) => ({ + ...exportRow(row, KEY, RULES), + ...(day === '2026-10-19' ? {} : { export_rules: 'old-rules' }), + })), + ]), + ); + const first = fakeWarehouse(partitions); + + await assert.rejects( + runExport({ + source: fakeSource(pages, ['2026-10-18']), + warehouse: first.warehouse, + hmacKey: KEY, + today: '2026-10-22', + scratchFile: scratch(), + log: () => undefined, + }), + /event export 2026-10-18: connection lost$/, + ); + assert.deepEqual(first.calls, [ + 'ensureTable', + 'replace 2026-10-20', + 'replace 2026-10-21', + 'replace 2026-10-22', + ]); + assert.deepEqual(partitions.get('2026-10-20'), [exportRow(at(3, '2026-10-20'), KEY, RULES)]); + + const second = fakeWarehouse(partitions); + + await runExport({ + source: fakeSource(pages), + warehouse: second.warehouse, + hmacKey: KEY, + today: '2026-10-22', + scratchFile: scratch(), + log: () => undefined, + }); + assert.deepEqual(second.calls, [ + 'ensureTable', + 'replace 2026-10-18', + 'replace 2026-10-21', + 'replace 2026-10-22', + ]); +}); + +test('runExport names the short days beside a failed day', async () => { const partitions = new Map([['2026-10-21', [{ id: 5 }, { id: 6 }]]]); const { warehouse, calls } = fakeWarehouse(partitions); @@ -584,7 +706,7 @@ test('runExport names the short days read before the day it failed on', async () '2026-10-21': [[sourceEvent({ id: 6 })]], '2026-10-22': [[sourceEvent({ id: 9 })]], }, - '2026-10-22', + ['2026-10-22'], ), warehouse, hmacKey: KEY, @@ -604,6 +726,46 @@ test('runExport names the short days read before the day it failed on', async () assert.equal(partitions.get('2026-10-21')?.length, 2); }); +test('runExport starts no new day past its budget and fails naming the days it did not reach', async () => { + const partitions = new Map(); + const { warehouse, calls } = fakeWarehouse(partitions, { failLoad: ['2026-10-21'] }); + const clock = { ms: 0 }; + const source = fakeSource({ + '2026-10-20': [[sourceEvent({ id: 1, createdAt: '2026-10-20T08:00:00.000000Z' })]], + '2026-10-21': [[sourceEvent({ id: 2, createdAt: '2026-10-21T08:00:00.000000Z' })]], + '2026-10-22': [[sourceEvent({ id: 3, createdAt: '2026-10-22T08:00:00.000000Z' })]], + '2026-10-23': [[sourceEvent({ id: 4, createdAt: '2026-10-23T08:00:00.000000Z' })]], + }); + + await assert.rejects( + runExport({ + source: { + ...source, + // Each day read costs 4 minutes of the injected clock. + readDay: (day) => { + clock.ms += 240_000; + return source.readDay(day); + }, + }, + warehouse, + hmacKey: KEY, + today: '2026-10-23', + scratchFile: scratch(), + log: () => undefined, + budgetMs: 600_000, + now: () => clock.ms, + }), + (error: unknown) => { + assert.ok(error instanceof Error); + assert.match(error.message, /2026-10-21: load job failed/); + assert.match(error.message, /not reached\D+10 minute budget: 2026-10-23$/); + return true; + }, + ); + + assert.deepEqual(calls, ['ensureTable', 'replace 2026-10-20', 'replace 2026-10-22']); +}); + test('bigQueryWarehouse loads a day into its partition and lets the schema gain columns', async () => { const loads: { table: string; file: string; metadata: Record }[] = []; const fake = { @@ -631,11 +793,136 @@ test('bigQueryWarehouse loads a day into its partition and lets the schema gain assert.deepEqual(loads[0]?.metadata.schemaUpdateOptions, ['ALLOW_FIELD_ADDITION']); }); +test('EXPORT_SCHEMA admits a REQUIRED column only among those the table was created with', () => { + // @ts-expect-error ALLOW_FIELD_ADDITION adds NULLABLE columns only. + const required: ExportField = { name: 'added', type: 'STRING', mode: 'REQUIRED' }; + const nullable: ExportField = { name: 'added', type: 'STRING' }; + + assert.ok(required && nullable); +}); + +test('bigQueryWarehouse creates the table, finds stale days and counts from partition metadata', async (t) => { + const fakeTable = ({ exists, fields = [] }: { exists: boolean; fields?: string[] }) => { + const calls: unknown[] = []; + const table = { + exists: async () => [exists], + getMetadata: async () => [{ schema: { fields: fields.map((name) => ({ name })) } }], + }; + const bigquery = { + dataset: () => ({ + table: () => table, + createTable: async (name: string, options: { schema: unknown }) => { + calls.push(['createTable', name, options.schema]); + return [table]; + }, + }), + query: async ({ query, params }: { query: string; params?: unknown }) => { + calls.push(['query', params]); + return query.includes('PARTITIONS') + ? [ + [ + { partition_id: '20261021', total_rows: 3 }, + { partition_id: '20261022', total_rows: '1434' }, + ], + ] + : [[{ day: '2026-10-21' }]]; + }, + }; + const warehouse = bigQueryWarehouse({ + projectId: 'p', + datasetId: 'd', + bigquery: bigquery as unknown as BigQuery, + }); + + return { warehouse, calls }; + }; + + await t.test('a missing table is created with the schema', async () => { + const { warehouse, calls } = fakeTable({ exists: false }); + + await warehouse.ensureTable(); + assert.deepEqual(calls, [['createTable', 'event', { fields: EXPORT_SCHEMA }]]); + }); + + await t.test('staleDays queries the days whose rows carry other rules', async () => { + const { warehouse, calls } = fakeTable({ exists: true, fields: ['id', 'export_rules'] }); + + assert.deepEqual(await warehouse.staleDays('r1'), new Set(['2026-10-21'])); + assert.deepEqual(calls, [['query', { rules: 'r1' }]]); + }); + + await t.test('a table without export_rules is stale on every day', async () => { + assert.deepEqual( + await fakeTable({ exists: true, fields: ['id'] }).warehouse.staleDays('r1'), + new Set(['2026-10-21', '2026-10-22']), + ); + }); + + await t.test('a query over the bytes-billed cap fails naming the cap', async () => { + const bigquery = { + dataset: () => ({ + table: () => ({ + getMetadata: async () => [{ schema: { fields: [{ name: 'export_rules' }] } }], + }), + }), + query: async () => { + throw new Error('Query exceeded limit for bytes billed: 1000000000.'); + }, + }; + const warehouse = bigQueryWarehouse({ + projectId: 'p', + datasetId: 'd', + bigquery: bigquery as unknown as BigQuery, + }); + + await assert.rejects(warehouse.staleDays('r1'), /maximumBytesBilled cap/); + }); + + await t.test('dayCounts reads each partition_id as its day', async () => { + assert.deepEqual( + await fakeTable({ exists: true }).warehouse.dayCounts(), + new Map([ + ['2026-10-21', 3], + ['2026-10-22', 1434], + ]), + ); + }); +}); + +test('runExportJob runs the event export after a failed account export, then fails', async () => { + const ran: string[] = []; + const logged: unknown[] = []; + const accountError = new Error('bigquery down'); + + await assert.rejects( + runExportJob({ + exportAccounts: async () => { + ran.push('accounts'); + throw accountError; + }, + exportEvents: async () => { + ran.push('events'); + }, + logError: (error) => logged.push(error), + }), + (error: unknown) => { + assert.ok(error instanceof Error); + assert.equal(error.cause, accountError); + return true; + }, + ); + + assert.deepEqual(ran, ['accounts', 'events']); + assert.deepEqual(logged, [accountError]); +}); + /** * Reads a migrated database as a login member of `website_export_reader`, * the way the Job's IAM user reads it, and checks each UTC day's rows * against Postgres's own per-day count. The page size is 2, so every day - * spans pages, and rows sit either side of a midnight. + * spans pages, rows sit either side of a midnight, three rows share one + * time across a page boundary, and each row's props hold a key no event + * keeps. */ test('the export reader reads every row of each UTC day and no other table', async (t) => { const database = await createTestDatabase(); @@ -661,7 +948,8 @@ test('the export reader reads every row of each UTC day and no other table', asy ); await database.runtimePool().query( `insert into event (name, created_at, user_id, visitor_id, page, props) - select 'search.query', ts::timestamptz, null, 'v', '/data', '{"post": 0}' + select 'search.query', ts::timestamptz, null, 'v', '/data', + '{"post": 0, "query": "aid to kenya"}' from unnest($1::text[]) ts`, [ [ @@ -671,8 +959,8 @@ test('the export reader reads every row of each UTC day and no other table', asy '2026-10-20T23:59:59.999999Z', '2026-10-21T00:00:00.000001Z', '2026-10-22T05:00:00Z', - '2026-10-22T06:00:00Z', - '2026-10-22T07:00:00Z', + '2026-10-22T05:00:00Z', + '2026-10-22T05:00:00Z', ], ], ); @@ -693,19 +981,24 @@ test('the export reader reads every row of each UTC day and no other table', asy from event group by 1 order by 1`, ); const read: { day: string; count: number }[] = []; + const ids: number[] = []; for (const { day } of expected) { let count = 0; for await (const page of source.readDay(day)) { assert.ok(page.every((row) => row.createdAt.startsWith(day))); + assert.ok(page.every((row) => Object.keys(row.props).join() === 'post')); count += page.length; + ids.push(...page.map((row) => row.id)); } read.push({ day, count }); } assert.deepEqual(read, expected); + // The three rows that share 05:00 on 22 October span two pages. + assert.deepEqual(ids, [1, 2, 3, 4, 5, 6, 7, 8]); assert.deepEqual( expected.map(({ count }) => count), [1, 3, 1, 3], diff --git a/deploy/README.md b/deploy/README.md index 937cfaa..36ebeb3 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -71,54 +71,27 @@ migration. ## Event export -In production a Cloud Run Job, `dfo-prod-website-export`, copies the -`event` table into the BigQuery table `event` in `EXPORT_BQ_DATASET` -(`website_export`) every hour. It runs the website image with the command -`node dist/export-events.cjs` (`db/export-events.ts`, bundled by -`pnpm build:export-events`). `build-migrate-deploy.yml` points it at each -new image after the service update, only where `EXPORT_JOB_WEBSITE` is -set. infra defines the Job, its identity, the dataset and the schedule. - -The Job reads the database as its own IAM user, a member of the NOLOGIN -role `website_export_reader`, which migration `0019_event_export_grant` -grants `SELECT` on `event` and migration `0020_account_daily` on the view -`account_daily`, and on nothing else. infra creates the role in every tier, -so its apply must land before the deploy that first carries the migration. - -| Env var | What it is | -| --------------------------------------------------------------------------- | ------------------------------------------------------------ | -| `WEBSITE_DB_INSTANCE_CONNECTION_NAME`, `WEBSITE_DB_NAME`, `WEBSITE_DB_USER` | The database as the service reads it, `DATABASE_URL` locally | -| `EXPORT_BQ_PROJECT` | The BigQuery project | -| `EXPORT_BQ_DATASET` | The BigQuery dataset | -| `EXPORT_ID_HMAC_KEY` | Key for the visitor and user id digests | - -Each run replaces every UTC day from the latest day in the table minus one -through today, so a missed run is made up by the next one. When the table -is empty or missing, the Job creates it and loads every day from the first -event. The run also replaces every earlier day whose count of `event` rows -differs from its partition's row count. It takes both counts per day up -front, one `GROUP BY` over the whole `event` table and one over the -BigQuery table's `created_at` column. Each day is read in pages into a -scratch file, then loaded into its partition (`event$YYYYMMDD`, -`WRITE_TRUNCATE`, `ALLOW_FIELD_ADDITION`, so a column added to the schema -reaches the table on the next load). A replace never shrinks a partition. -Nothing deletes `event` rows, and they carry no foreign key to `user`, so -deleting an account leaves them in place. A day that reads fewer rows than -its partition holds is therefore a fault, such as a wrong database, an -emptied table or a lagging replica. That day keeps its partition, the -other days still load, and the run then fails naming each short day with -both counts, as it does for a day that reads fewer rows than its source -count from the start of the run, whatever its partition holds. Its counts -still differ, so the day stays in every later run, however many later days -load, and fails each one until it matches. The Job's failure alert -therefore fires on every run until the export is complete. A day that -reads no rows and holds none loads nothing. The Job logs each day as it -finishes. A failure names the day it failed on and any short day before -it, and that day keeps its partition as it was. A database query that runs -past 5 minutes fails the run (`statement_timeout` on the Job's pool), and -so does a BigQuery count job that bills more than 1 GB, about 125 million -rows, or runs past 2 minutes (`maximumBytesBilled` and `jobTimeoutMs`), -all well inside the Job's 1800 s task timeout. +In production a Cloud Run Job, `dfo-prod-website-export`, copies the `event` table into the BigQuery table `event` in `EXPORT_BQ_DATASET` (`website_export`) every hour. It runs the website image with the command `node dist/export-events.cjs` (`db/export-events.ts`, bundled by `pnpm build:export-events`). `build-migrate-deploy.yml` points it at each new image after the service update, only where `EXPORT_JOB_WEBSITE` is set. infra defines the Job, its identity, the dataset and the schedule. + +The Job reads the database as its own IAM user, a member of the NOLOGIN role `website_export_reader`, which migration `0019_event_export_grant` grants `SELECT` on `event`, migration `0020_account_daily` on the view `account_daily`, and migration `0021_account_daily_before_today` on the view `account_daily_source`, and on nothing else. infra creates the role in every tier, so its apply must land before the deploy that first carries the migration. + +| Env var | What it is | +| -------------------------------------------------------- | ------------------------------------------------------------------------- | +| `WEBSITE_DB_INSTANCE_CONNECTION_NAME`, `WEBSITE_DB_NAME` | The database the Job reads, `DATABASE_URL` locally | +| `WEBSITE_DB_USER` | The exporter's own IAM database user, a member of `website_export_reader` | +| `EXPORT_BQ_PROJECT` | The BigQuery project | +| `EXPORT_BQ_DATASET` | The BigQuery dataset | +| `EXPORT_ID_HMAC_KEY` | Key for the visitor and user id digests | + +Each run replaces every UTC day from the latest day in the table minus one through today, so a missed run is made up by the next one. When the table is empty or missing, the Job creates it and loads every day from the first event. The run also replaces every earlier day whose count of `event` rows differs from its partition's row count. It takes both counts per day up front, one `GROUP BY` over the whole `event` table and the partitions' row counts from the dataset's `INFORMATION_SCHEMA.PARTITIONS`, which reads table metadata and scans no rows. Each day is read in pages, in `created_at` and id order with each page starting after the last row of the one before, and only with the prop keys `PROP_RULES` keeps. The pages go to a scratch file, then load into the day's partition (`event$YYYYMMDD`, `WRITE_TRUNCATE`, `ALLOW_FIELD_ADDITION`). A load adds a new NULLABLE column to the table, and fails on a new REQUIRED one, so `ExportField` in `db/event-export.ts` lets a column be REQUIRED only if the table was created with it. + +The scratch file sits in `os.tmpdir()`, which on Cloud Run is in memory, so a run holds one page of rows plus one day's file, about 360 bytes an event. The busiest day so far holds 1,434 events, about 0.5 MB, against the Job's 512 MiB. The file is written in full before its load starts, so a day loads only once it has been read to its end. + +A replace never shrinks a partition. Nothing deletes `event` rows, and they carry no foreign key to `user`, so deleting an account leaves them in place. A day that reads fewer rows than its partition holds is therefore a fault, such as a wrong database, an emptied table or a lagging replica. That day keeps its partition, the other days still load, and the run then fails naming each short day with both counts, as it does for a day that reads fewer rows than its source count from the start of the run, whatever its partition holds. Its counts still differ, so the day stays in every later run, however many later days load, and fails each one until it matches. The Job's failure alert therefore fires on every run until the export is complete. A day that reads no rows and holds none loads nothing. + +A day whose read or load fails keeps its partition as it was, and the later days still load. The run then fails naming every such day with its error, and any short day. The Job logs each day as it finishes. A database query that runs past 5 minutes fails its day (`statement_timeout` on the Job's pool); a query that fails outside a day's read, such as the per-day count, fails the run. A BigQuery query that bills more than 1 GB or runs past 2 minutes fails the run (`maximumBytesBilled` and `jobTimeoutMs`). The stale-day query scans about 74 bytes an event row, so it reaches the 1 GB cap at about 13.5 million rows, and the run then fails with an error naming the cap. Days that each wait out a 5 minute timeout could outlast the Job's 1800 s task timeout, so a run starts no new day once 20 minutes have passed since it began (`EXPORT_BUDGET_MS` in `db/event-export.ts`). It finishes the day in hand, then fails naming every day it did not reach, with any failed or short day, and the next run picks those days up. + +Every row carries in `export_rules` a digest of what shapes it (`exportRules` in `db/event-export.ts`): `PROP_RULES`, `PAGE_PATH`, `SHARE_PATH`, the token rule, the schema, `ROW_SHAPE_VERSION` and an HMAC of a fixed string under `EXPORT_ID_HMAC_KEY`. `ROW_SHAPE_VERSION` is bumped by hand with any change to the functions that build a row. Each run finds the days holding a row with another digest, or none, in one query over the `created_at` and `export_rules` columns, and replaces those days too. A change to the rules or the key thus reaches every partition. A day that fails keeps its old rows, so the next run retries it with the window days and the count-mismatch days, and reloads no day that is already current. The table is partitioned by day on `created_at`: @@ -132,35 +105,15 @@ The table is partitioned by day on `created_at`: | `visitor_hash` | STRING | HMAC-SHA256 hex of `visitor_id`, null without one | | `user_hash` | STRING | HMAC-SHA256 hex of `user_id`, null without one | | `props` | JSON | the event's allow-listed props, `shareId` as its HMAC-SHA256 hex | +| `export_rules` | STRING | the digest of the rules the row was exported under | -`page` keeps a path only when it is a slash followed by letters, digits -and `_ - . : /`, up to 300 characters, the characters the site's routes -use (`PAGE_PATH` in `db/event-export.ts`). Any other text the browser -posts as its page exports as null. The id in `/s/`, `/embed/` and -the pages under them exports as `:id`, since a share id opens its share. - -`PROP_RULES` in `db/event-export.ts` lists the props each event keeps: -enums, booleans, counts and ids, each checked against its rule, since the -browser posts its own props. Search query text, refused terms, card units -and every list or nested object stay in Postgres. `shareId` exports as -its HMAC-SHA256 hex under `EXPORT_ID_HMAC_KEY`, so share, embed and -search events still join on it. A new event name fails -the typecheck until it has an entry there. Changing `EXPORT_ID_HMAC_KEY` -changes every digest, so drop the BigQuery table afterwards and let the -next run reload it. +`page` keeps a path only when it is a slash followed by letters, digits and `_ - . : /`, up to 300 characters, the characters the site's routes use (`PAGE_PATH` in `db/event-export.ts`). Any other text the browser posts as its page exports as null. The id in `/s/`, `/embed/` and the pages under them exports as `:id`, since a share id opens its share. + +`PROP_RULES` in `db/event-export.ts` lists the props each event keeps: enums, booleans, counts and ids, each checked against its rule, since the browser posts its own props. Search query text, refused terms, card units and every list or nested object stay in Postgres. `shareId` exports as its HMAC-SHA256 hex under `EXPORT_ID_HMAC_KEY`, so share, embed and search events still join on it. A new event name fails the typecheck until it has an entry there. Changing `EXPORT_ID_HMAC_KEY` changes every digest and the rules digest with them, so the next run reloads every day. ### Accounts -Each run first replaces the BigQuery table `account_daily` with every row -of the Postgres view `account_daily` (`db/account-export.ts`), in one -`WRITE_TRUNCATE` load job, so a failed run leaves the previous table. The -event export still runs when it fails, and the run then fails too. A read -with no rows fails the run and keeps the table. The view runs with the -migrator's rights, so the Job reads accounts per day and organisation and -no email address or user id. It holds every UTC day from the first -account's through yesterday's, so the latest day is the last complete -one, and a day without new accounts is one row with a null `organisation` -and 0 `accounts`. +Each run first replaces the BigQuery table `account_daily` with every row of the Postgres view `account_daily` (`db/account-export.ts`), in one `WRITE_TRUNCATE` load job, so a failed run leaves the previous table. The event export still runs when it fails, and the run then fails too. The view runs with the migrator's rights, so the Job reads accounts per day and organisation and no email address or user id. It holds every UTC day from the first account's through yesterday's, so the latest day is the last complete one, and a day without new accounts is one row with a null `organisation` and 0 `accounts`. It counts only the accounts created before today, UTC, so a domain it names holds at least 3 of the accounts the export carries. | Column | Type | Value | | -------------- | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | @@ -168,13 +121,9 @@ and 0 `accounts`. | `organisation` | STRING | `Personal email` for a webmail domain, else the lower-cased domain after the last `@` once it holds at least 3 accounts, else `Other organisations`; null on a day without new accounts, whose one row has 0 accounts | | `accounts` | INT64 | the accounts created that day, 0 on a day without new accounts | -A domain in `webmail_domain` is `Personal email` whatever its count. The -view holds no rows while that table is empty, so the run fails and keeps -the BigQuery table. The table holds the `free-email-domains` package's -list, which `db/migrate.ts` -writes after the migrations on every deploy, so bumping the package -updates it. pnpm skips the package's postinstall script, which would -refetch the list. +A domain in `webmail_domain` is `Personal email` whatever its count. The table holds the `free-email-domains` package's list, which `db/migrate.ts` writes after the migrations on every deploy, so bumping the package updates it. pnpm skips the package's postinstall script, which would refetch the list. + +A read with no rows fails the run and leaves the table as it was while `webmail_domain` is empty, since the view then holds no rows and webmail domains would show by name, and while the user table holds no account, which is a wrong database or an emptied table. When accounts exist and every one is from today, the run replaces the table with no rows, so a deleted account and the organisation it named leave the export. The view `account_daily_source` tells the cases apart, and the Job reads it in the same transaction as `account_daily`. ## GitHub vars and secrets