feat(store): own the analytics rollups in application code

.github/workflows/lambda-deploy.yml

@@ -7,6 +7,7 @@ on:
       - 'apps/backend/lambdas/**'
       - 'shared/types/**'
       - 'shared/lambda-http/**'
+      - 'shared/store/**'
       - 'apps/backend/db/migrations/**'
   workflow_dispatch:
     inputs:
@@ -42,7 +43,7 @@ jobs:
           migrate=false
           code=false
           grep -q '^apps/backend/db/migrations/' <<<"$changed_files" && migrate=true
-          grep -qE '^(apps/backend/lambdas/|shared/types/|shared/lambda-http/)' <<<"$changed_files" && code=true
+          grep -qE '^(apps/backend/lambdas/|shared/types/|shared/lambda-http/|shared/store/)' <<<"$changed_files" && code=true
 
           # workflow_dispatch has no meaningful diff: apply everything, or
           # migrations only when explicitly asked (recovery / manual re-run).

apps/backend/db/migrations/20260906215733_move_rollups_to_application.sql

@@ -0,0 +1,46 @@
+-- 20260906215733_move_rollups_to_application
+--
+-- Aurora DSQL supports neither triggers nor PL/pgSQL, so rollup maintenance moves
+-- into @branch/store. The three LANGUAGE sql helpers stay -- DSQL supports those,
+-- and they hold the arithmetic; only the INSERT/UPDATE/DELETE dispatch moves.
+--
+-- The two FKs that flip to RESTRICT are the ones whose cascade used to fire a row
+-- trigger from a lambda that never names the rollup tables (DELETE /donors/{id},
+-- DELETE /users/{id}). RESTRICT turns a forgotten cascade into a loud FK error
+-- instead of a silently wrong total. @branch/store deletes the children first.
+
+DROP TRIGGER IF EXISTS expenditures_rollup_sync ON expenditures;
+DROP TRIGGER IF EXISTS donations_rollup_sync ON project_donations;
+DROP TRIGGER IF EXISTS memberships_rollup_sync ON project_memberships;
+DROP TRIGGER IF EXISTS reports_rollup_sync ON reports;
+DROP TRIGGER IF EXISTS projects_rollup_seed ON projects;
+
+DROP TRIGGER IF EXISTS expenditures_rollup_truncate ON expenditures;
+DROP TRIGGER IF EXISTS donations_rollup_truncate ON project_donations;
+DROP TRIGGER IF EXISTS memberships_rollup_truncate ON project_memberships;
+DROP TRIGGER IF EXISTS reports_rollup_truncate ON reports;
+
+DROP FUNCTION branch.expenditures_rollup_sync();
+DROP FUNCTION branch.donations_rollup_sync();
+DROP FUNCTION branch.memberships_rollup_sync();
+DROP FUNCTION branch.reports_rollup_sync();
+DROP FUNCTION branch.projects_rollup_seed();
+DROP FUNCTION branch.expenditures_rollup_truncate();
+DROP FUNCTION branch.donations_rollup_truncate();
+DROP FUNCTION branch.memberships_rollup_truncate();
+DROP FUNCTION branch.reports_rollup_truncate();
+
+-- NOT VALID because DSQL requires it on ALTER TABLE ADD CONSTRAINT, and it is
+-- accurate either way: the constraint being replaced referenced the same rows,
+-- so existing data cannot violate the new one. Only the delete action changes.
+ALTER TABLE project_donations
+    DROP CONSTRAINT project_donations_donor_id_fkey;
+ALTER TABLE project_donations
+    ADD CONSTRAINT project_donations_donor_id_fkey
+    FOREIGN KEY (donor_id) REFERENCES donors(donor_id) ON DELETE RESTRICT NOT VALID;
+
+ALTER TABLE project_memberships
+    DROP CONSTRAINT project_memberships_user_id_fkey;
+ALTER TABLE project_memberships
+    ADD CONSTRAINT project_memberships_user_id_fkey
+    FOREIGN KEY (user_id) REFERENCES users(user_id) ON DELETE RESTRICT NOT VALID;

apps/backend/db/migrations/20260907213524_project_rollup_bump_reports_hit.sql

@@ -0,0 +1,23 @@
+-- 20260907213524_project_rollup_bump_reports_hit
+--
+-- Returns 1/NULL, not void: only the store seeds project_rollup now, so a bump
+-- matching no row must fail rather than drift.
+
+DROP FUNCTION branch.project_rollup_bump(INT, INT, NUMERIC, INT, INT);
+
+CREATE FUNCTION branch.project_rollup_bump(
+    p_project_id INT,
+    p_members    INT,
+    p_donated    NUMERIC,
+    p_donations  INT,
+    p_reports    INT
+) RETURNS integer LANGUAGE sql AS $$
+    UPDATE branch.project_rollup
+       SET member_count   = member_count + p_members,
+           total_donated  = total_donated + p_donated,
+           donation_count = donation_count + p_donations,
+           report_count   = report_count + p_reports,
+           updated_at     = CURRENT_TIMESTAMP
+     WHERE project_id = p_project_id
+    RETURNING 1;
+$$;

apps/backend/db/testkit.ts

@@ -196,6 +196,98 @@ async function truncateAll(client: Queryable): Promise<string> {
 export async function resetData(client: Queryable): Promise<void> {
   await client.query(await truncateAll(client));
   await client.query(seedSql());
+  await reconcileRollups(client);
+}
+
+const EXPECTED_EXPENDITURE_ROLLUP = `
+  SELECT project_id,
+         date_trunc('month', spent_on)::date AS month,
+         category,
+         status,
+         SUM(amount) AS total_amount,
+         COUNT(*)::int AS expenditure_count
+    FROM ${SCHEMA}.expenditures
+   GROUP BY project_id, date_trunc('month', spent_on)::date, category, status`;
+
+const EXPECTED_PROJECT_ROLLUP = `
+  SELECT p.project_id,
+         COALESCE(m.c, 0)     AS member_count,
+         COALESCE(d.total, 0) AS total_donated,
+         COALESCE(d.c, 0)     AS donation_count,
+         COALESCE(r.c, 0)     AS report_count
+    FROM ${SCHEMA}.projects p
+    LEFT JOIN (SELECT project_id, COUNT(*) AS c FROM ${SCHEMA}.project_memberships GROUP BY project_id) m
+           ON m.project_id = p.project_id
+    LEFT JOIN (SELECT project_id, COUNT(*) AS c, SUM(amount) AS total FROM ${SCHEMA}.project_donations GROUP BY project_id) d
+           ON d.project_id = p.project_id
+    LEFT JOIN (SELECT project_id, COUNT(*) AS c FROM ${SCHEMA}.reports GROUP BY project_id) r
+           ON r.project_id = p.project_id`;
+
+// A zero-count expenditure_rollup row equals a missing one: _remove decrements without deleting.
+export async function findRollupDrift(client: Queryable): Promise<string[]> {
+  const expenditures = await client.query(`
+    WITH expected AS (${EXPECTED_EXPENDITURE_ROLLUP})
+    SELECT 'expenditure_rollup project=' || COALESCE(e.project_id, a.project_id)
+           || ' month=' || COALESCE(e.month, a.month)
+           || ' status=' || COALESCE(e.status, a.status)
+           || ' category=' || COALESCE(e.category, a.category, '<null>')
+           || ' expected=' || COALESCE(e.total_amount, 0) || '/' || COALESCE(e.expenditure_count, 0)
+           || ' actual='   || COALESCE(a.total_amount, 0) || '/' || COALESCE(a.expenditure_count, 0) AS drift
+      FROM expected e
+      FULL OUTER JOIN ${SCHEMA}.expenditure_rollup a
+        ON a.project_id = e.project_id
+       AND a.month = e.month
+       AND a.status = e.status
+       AND (a.category IS NULL) = (e.category IS NULL)
+       AND COALESCE(a.category, '') = COALESCE(e.category, '')
+     WHERE COALESCE(a.expenditure_count, 0) <> COALESCE(e.expenditure_count, 0)
+        OR COALESCE(a.total_amount, 0) <> COALESCE(e.total_amount, 0)`);
+
+  const projects = await client.query(`
+    WITH expected AS (${EXPECTED_PROJECT_ROLLUP})
+    SELECT 'project_rollup project=' || e.project_id
+           || ' expected=' || e.member_count || '/' || e.total_donated || '/' || e.donation_count || '/' || e.report_count
+           || ' actual='   || COALESCE(a.member_count::text, 'ROW MISSING')
+           || '/' || COALESCE(a.total_donated::text, '-')
+           || '/' || COALESCE(a.donation_count::text, '-')
+           || '/' || COALESCE(a.report_count::text, '-') AS drift
+      FROM expected e
+      LEFT JOIN ${SCHEMA}.project_rollup a ON a.project_id = e.project_id
+     WHERE a.project_id IS NULL
+        OR a.member_count <> e.member_count
+        OR a.total_donated <> e.total_donated
+        OR a.donation_count <> e.donation_count
+        OR a.report_count <> e.report_count`);
+
+  return [...(expenditures.rows ?? []), ...(projects.rows ?? [])].map(
+    (row) => row.drift as string,
+  );
+}
+
+export async function assertRollupsConsistent(client: Queryable): Promise<void> {
+  const drift = await findRollupDrift(client);
+  if (drift.length > 0) {
+    throw new Error(
+      `rollups disagree with the base tables:\n  ${drift.join('\n  ')}`,
+    );
+  }
+}
+
+export async function reconcileRollups(client: Queryable): Promise<void> {
+  await client.query(`DELETE FROM ${SCHEMA}.expenditure_rollup`);
+  await client.query(
+    `INSERT INTO ${SCHEMA}.expenditure_rollup
+       (project_id, month, category, status, total_amount, expenditure_count)
+     SELECT project_id, month, category, status, total_amount, expenditure_count
+       FROM (${EXPECTED_EXPENDITURE_ROLLUP}) AS expected`,
+  );
+  await client.query(`DELETE FROM ${SCHEMA}.project_rollup`);
+  await client.query(
+    `INSERT INTO ${SCHEMA}.project_rollup
+       (project_id, member_count, total_donated, donation_count, report_count)
+     SELECT project_id, member_count, total_donated, donation_count, report_count
+       FROM (${EXPECTED_PROJECT_ROLLUP}) AS expected`,
+  );
 }
 
 /**

apps/backend/lambdas/auth/auth.ts

@@ -1,6 +1,6 @@
 import { authenticateRequest as _authenticateRequest, loadRbacSubject } from '@branch/lambda-auth';
 import { createAuthResolver } from '@branch/lambda-http';
-import db from './db';
+import { db } from '@branch/store';
 
 export * from '@branch/lambda-auth';
 

apps/backend/lambdas/auth/controllers/mfa.ts

@@ -7,7 +7,7 @@ import {
 } from '@aws-sdk/client-cognito-identity-provider';
 import { json, parseBody, reportError, serverError } from '@branch/lambda-http';
 import type { RouteHandler } from '@branch/lambda-http';
-import db from '../db';
+import { db } from '@branch/store';
 import { cognitoClient } from '../services/cognito';
 
 /**

apps/backend/lambdas/auth/controllers/register.ts

@@ -9,7 +9,7 @@ import {
   ResendConfirmationCodeCommand,
 } from '@aws-sdk/client-cognito-identity-provider';
 import { json, reportError, serverError } from '@branch/lambda-http';
-import db from '../db';
+import { db, claimUser } from '@branch/store';
 import { cognitoClient, USER_POOL_CLIENT_ID, USER_POOL_ID, validatePassword } from '../services/cognito';
 
 export async function handleRegister(event: any): Promise<APIGatewayProxyResult> {
@@ -125,15 +125,10 @@ export async function handleRegister(event: any): Promise<APIGatewayProxyResult>
             );
             const sub = cognitoUser.UserAttributes?.find((a) => a.Name === 'sub')?.Value;
             if (sub && cognitoUser.UserStatus === 'CONFIRMED') {
-              const linkResult = await db
-                .updateTable('branch.users')
-                .set({ cognito_sub: sub })
-                .where('user_id', '=', claimingUserId)
-                .where('cognito_sub', 'is', null)
-                .executeTakeFirst();
+              const linked = await claimUser(claimingUserId, { cognito_sub: sub });
               // A concurrent claim already took this row; do not delete the
               // pre-existing Cognito user, it may back a working account.
-              if (linkResult.numUpdatedRows > 0n) {
+              if (linked > 0n) {
                 return json(200, {
                   message: 'Existing account linked',
                   claimed: true,
@@ -187,16 +182,14 @@ export async function handleRegister(event: any): Promise<APIGatewayProxyResult>
       // claim one an admin already approved. The cognito_sub IS NULL predicate
       // makes a concurrent claim a no-op rather than an overwrite;
       // UNIQUE(cognito_sub) is the backstop.
-      const claimResult = await db
-        .updateTable('branch.users')
-        .set({ cognito_sub: cognitoUserSub, name: name.trim() })
-        .where('user_id', '=', claimingUserId)
-        .where('cognito_sub', 'is', null)
-        .executeTakeFirst();
+      const claimed = await claimUser(claimingUserId, {
+        cognito_sub: cognitoUserSub,
+        name: name.trim(),
+      });
 
       // No-op claim: the Cognito sub we just created would reference no row, so
       // every later login would fail. Undo the Cognito user instead.
-      if (claimResult.numUpdatedRows === 0n) {
+      if (claimed === 0n) {
         console.error('Invitation already claimed for user_id:', claimingUserId);
         await rollbackCognitoUser();
         return json(409, {

apps/backend/lambdas/auth/package.json

@@ -31,6 +31,7 @@
     "@branch/lambda-auth": "file:../../../../shared/lambda-auth",
     "@branch/lambda-http": "file:../../../../shared/lambda-http",
     "@branch/rbac": "file:../../../../shared/rbac",
+    "@branch/store": "file:../../../../shared/store",
     "aws-jwt-verify": "^5.1.1",
     "dotenv": "^17.2.3",
     "kysely": "^0.28.10",