chore(db): staging ops scripts + newsletter_emails.approved_at migration

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.

backend/db/_staging_ops.py

@@ -0,0 +1,67 @@
+"""Shared helpers for the one-off staging ops scripts in this directory:
+copy_courses_to_staging, grant_staging_admin, approve_staging_users.
+
+Those scripts span two Supabase projects and so deliberately bypass
+db/connection.py::table() (which is single-environment) — see each script's module
+docstring. The common bits live here to keep them from diverging: dotenv reading, a
+safe-by-default write guard, the staging-key setup, and the decrypt-email→user index.
+
+Credentials are read from gitignored dotenv files (never argv/process env), so the
+DSN/keys don't leak into shell history or logs.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+
+
+def dotenv_value(path: str | Path, key: str) -> str:
+ p = Path(path)
+ for line in p.read_text().splitlines():
+ line = line.strip()
+ if line.startswith(key + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ raise SystemExit(f"ERROR: {key} not found in {p}")
+
+
+def set_encryption_key(staging_env: str | Path = ".env.staging") -> None:
+ """Put the staging ENCRYPTION_KEY in the process env. MUST be called before
+ importing services.encryption, which loads the key at import time."""
+ os.environ["ENCRYPTION_KEY"] = dotenv_value(staging_env, "ENCRYPTION_KEY")
+
+
+def target_host(db_url: str) -> str:
+ return urlparse(db_url).hostname or "<unknown>"
+
+
+def confirm_write(db_url: str, apply: bool, action: str) -> bool:
+ """Safe-by-default guard against pointing a write at the wrong project.
+
+ Prints the destination host. Returns True only when --yes (apply=True) was
+ passed; otherwise prints a preview notice and returns False so the caller can
+ report what it *would* do without mutating anything.
+ """
+ print(f"Write target: {target_host(db_url)}")
+ if not apply:
+ print(f" PREVIEW ONLY — re-run with --yes to {action}. No changes made.")
+ return apply
+
+
+def user_email_index(cur) -> dict[str, tuple[str, bool]]:
+ """Map decrypted plaintext email -> (user_id, is_approved) for all users.
+
+ users.email is AES-GCM encrypted with a random nonce (non-deterministic), so
+ matching requires decrypting each row. Imports decrypt lazily so callers can
+ call set_encryption_key() first.
+ """
+ from services.encryption import decrypt_if_present
+
+ cur.execute("SELECT id, email, is_approved FROM users")
+ index: dict[str, tuple[str, bool]] = {}
+ for uid, email_ct, approved in cur.fetchall():
+ plain = (decrypt_if_present(email_ct) or "").strip().lower()
+ if plain:
+ index[plain] = (uid, approved)
+ return index

backend/db/approve_staging_users.py

@@ -0,0 +1,69 @@
+"""One-off ops script: approve staging user accounts by email (is_approved=true).
+
+This is the access gate the sign-in flow checks: a brand-new sign-in always lands
+`is_approved=false` and there's no auto-approve path, so accounts must be approved
+out of band. Mirrors the admin portal's `user.approve`, but works directly against
+the DB (handy for bootstrapping before anyone is an admin).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). Emails with no
+user row yet (haven't signed into staging) are reported PENDING and skipped — they
+can't be pre-approved because sign-in matches on google_id, not email. Idempotent.
+Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu # preview
+ venv/bin/python -m db.approve_staging_users a@bu.edu b@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("emails", nargs="+")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply approvals (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ targets = {e.strip().lower() for e in args.emails}
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ index = user_email_index(cur)
+ to_approve = [] # (email, uid, was_approved)
+ pending = [] # email with no row yet
+ for email in sorted(targets):
+ if email in index:
+ uid, was = index[email]
+ to_approve.append((email, uid, was))
+ else:
+ pending.append(email)
+
+ applying = confirm_write(db_url, args.yes, f"approve {len(to_approve)} account(s)")
+ if applying:
+ for email, uid, was in to_approve:
+ if not was:
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ conn.commit()
+
+ verb = "approved" if args.yes else "would approve"
+ for email, uid, was in to_approve:
+ state = "already approved" if was else verb
+ print(f" ✓ {email} ({uid}) -> {state}")
+ for email in pending:
+ print(f" … {email} -> PENDING: no staging account yet (must sign in once)")
+ print(f"\n{len(to_approve)} matched, {len(pending)} pending.")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/copy_courses_to_staging.py

@@ -0,0 +1,136 @@
+"""One-off ops script: copy the canonical `courses` catalog from prod to staging.
+
+`courses` is the shared course catalog (no `user_id`, no encrypted columns), so it
+can be copied row-for-row across projects safely — unlike `users`, whose encrypted
+columns are keyed to a per-environment ENCRYPTION_KEY.
+
+This spans two Supabase projects, so it deliberately bypasses
+db/connection.py::table() (single-environment). It reads PROD via PostgREST (prod has
+no direct DB URL) and writes STAGING via psycopg, mirroring db/migrate.py's direct
+connection. Writes target STAGING only; prod is read-only.
+
+Sync semantics: additive upsert keyed on `id` (INSERT … ON CONFLICT DO UPDATE).
+Courses deleted in prod are NOT removed from staging — this seeds/refreshes, it does
+not mirror. Safe to rerun.
+
+Usage (from backend/):
+ venv/bin/python -m db.copy_courses_to_staging # preview (read-only)
+ venv/bin/python -m db.copy_courses_to_staging --yes # apply the upsert
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+import httpx
+import psycopg
+
+from db._staging_ops import confirm_write, dotenv_value
+
+# Columns present on prod's `courses` table (a subset of staging's schema). Staging
+# defaults fill the rest (credits, meeting_times, location, syllabus_url).
+COLUMNS = [
+ "id",
+ "course_code",
+ "course_name",
+ "department",
+ "description",
+ "instructor_name",
+ "school",
+ "semester",
+ "created_at",
+]
+
+PAGE = 1000 # rows requested per PostgREST Range page
+
+
+def fetch_prod_courses(base_url: str, service_key: str) -> list[dict]:
+ # Page via the PostgREST Range header with Prefer: count=exact, and loop until
+ # we've fetched the total reported in Content-Range. Inferring completion from
+ # batch length vs PAGE is unreliable: if prod's db-max-rows is below PAGE the
+ # first response is short and the loop would stop early, under-fetching silently.
+ headers = {
+ "apikey": service_key,
+ "Authorization": f"Bearer {service_key}",
+ "Prefer": "count=exact",
+ }
+ select = ",".join(COLUMNS)
+ rows: list[dict] = []
+ offset = 0
+ total: int | None = None
+ with httpx.Client(timeout=30) as client:
+ while True:
+ range_headers = {**headers, "Range-Unit": "items", "Range": f"{offset}-{offset + PAGE - 1}"}
+ r = client.get(
+ f"{base_url}/rest/v1/courses",
+ params={"select": select, "order": "id"},
+ headers=range_headers,
+ )
+ r.raise_for_status()
+ batch = r.json()
+ rows.extend(batch)
+
+ # Content-Range looks like "0-999/1234" (or "*/1234" when empty).
+ content_range = r.headers.get("Content-Range", "")
+ if total is None and "/" in content_range:
+ tail = content_range.split("/", 1)[1].strip()
+ if tail.isdigit():
+ total = int(tail)
+
+ if not batch:
+ break
+ offset += len(batch)
+ if total is not None and offset >= total:
+ break
+ return rows
+
+
+def upsert_staging(db_url: str, rows: list[dict]) -> int:
+ cols = ", ".join(COLUMNS)
+ placeholders = ", ".join(["%s"] * len(COLUMNS))
+ # Keep `created_at` in the INSERT (set on first seed) but exclude it from the
+ # UPDATE: it is effectively immutable, so re-runs must not overwrite staging's
+ # existing value with prod's.
+ immutable = {"id", "created_at"}
+ updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in COLUMNS if c not in immutable)
+ sql = (
+ f"INSERT INTO courses ({cols}) VALUES ({placeholders}) "
+ f"ON CONFLICT (id) DO UPDATE SET {updates}"
+ )
+ values = [tuple(row.get(c) for c in COLUMNS) for row in rows]
+ with psycopg.connect(db_url, connect_timeout=15) as conn:
+ with conn.cursor() as cur:
+ cur.executemany(sql, values)
+ cur.execute("SELECT count(*) FROM courses")
+ total = cur.fetchone()[0]
+ conn.commit()
+ return total
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("--prod-env", default=".env")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the upsert (default: preview)")
+ args = ap.parse_args()
+
+ prod_url = dotenv_value(args.prod_env, "SUPABASE_URL")
+ prod_key = dotenv_value(args.prod_env, "SUPABASE_SERVICE_KEY")
+ staging_db = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ print(f"Reading courses from prod ({prod_url}) ...")
+ rows = fetch_prod_courses(prod_url, prod_key)
+ print(f" fetched {len(rows)} course rows")
+
+ if not confirm_write(staging_db, args.yes, f"upsert {len(rows)} courses"):
+ return 0
+
+ print("Upserting into staging ...")
+ total = upsert_staging(staging_db, rows)
+ print(f" done. staging courses table now has {total} rows")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/grant_staging_admin.py

@@ -0,0 +1,82 @@
+"""One-off ops script: approve a user and grant them the `admin` role in staging.
+
+Bootstraps the first staging admin, which the in-app `assign_role` endpoint can't do
+(it requires an *existing* admin). The user must have signed into staging at least
+once (so their row exists).
+
+Matches by decrypted email (see db/_staging_ops.user_email_index). The `admin`
+role_id is a per-project UUID, so it's looked up by slug='admin', not hardcoded.
+Idempotent. Targets STAGING only.
+
+Usage (from backend/):
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu # preview
+ venv/bin/python -m db.grant_staging_admin aflopez@bu.edu --yes # apply
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+from db._staging_ops import confirm_write, dotenv_value, set_encryption_key, user_email_index
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__)
+ ap.add_argument("email")
+ ap.add_argument("--staging-env", default=".env.staging")
+ ap.add_argument("--yes", action="store_true", help="apply the grant (default: preview)")
+ args = ap.parse_args()
+
+ set_encryption_key(args.staging_env) # before importing services.encryption
+ db_url = dotenv_value(args.staging_env, "SUPABASE_DB_URL")
+
+ import psycopg
+
+ target = args.email.strip().lower()
+
+ with psycopg.connect(db_url, connect_timeout=15) as conn, conn.cursor() as cur:
+ match = user_email_index(cur).get(target)
+ if match is None:
+ print(
+ f"ERROR: no staging user with email {target!r}. "
+ f"Have them sign into staging once, then re-run."
+ )
+ return 1
+ uid, approved = match
+
+ cur.execute("SELECT id FROM roles WHERE slug = 'admin'")
+ role_row = cur.fetchone()
+ if not role_row:
+ print("ERROR: no role with slug='admin' in staging.")
+ return 1
+ admin_role_id = role_row[0]
+
+ print(f"{target} -> id={uid} (currently approved={approved})")
+ if not confirm_write(db_url, args.yes, "approve + grant admin"):
+ return 0
+
+ cur.execute("UPDATE users SET is_approved = true WHERE id = %s", (uid,))
+ cur.execute(
+ "INSERT INTO user_roles (user_id, role_id, granted_by) "
+ "VALUES (%s, %s, %s) ON CONFLICT (user_id, role_id) DO NOTHING",
+ (uid, admin_role_id, "bootstrap:grant_staging_admin"),
+ )
+ conn.commit()
+
+ cur.execute(
+ "SELECT u.is_approved, "
+ "coalesce(string_agg(r.slug, ',' ORDER BY r.slug), '') "
+ "FROM users u "
+ "LEFT JOIN user_roles ur ON ur.user_id = u.id "
+ "LEFT JOIN roles r ON r.id = ur.role_id "
+ "WHERE u.id = %s GROUP BY u.is_approved",
+ (uid,),
+ )
+ is_approved, roles = cur.fetchone()
+ print(f"OK: {target} (id={uid}) -> approved={is_approved}, roles=[{roles}]")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())

backend/db/migrations/0019_newsletter_approved_at.sql

@@ -0,0 +1,16 @@
+-- newsletter_emails.approved_at: timestamp an admin approved an email for the
+-- newsletter allowlist (NULL = pending). The admin allowlist endpoints
+-- (POST /api/admin/allowlist/approve and /revoke in routes/admin.py) read and
+-- write this column.
+--
+-- Drift fix: prod had this column added out-of-band (it exists there), but it was
+-- never captured as a migration, so environments built purely from migrations
+-- (staging, CI, fresh DBs) were missing it — which 500s the allowlist endpoints.
+-- IF NOT EXISTS makes this a safe no-op where the column already exists (prod).
+ALTER TABLE newsletter_emails
+ ADD COLUMN IF NOT EXISTS approved_at timestamptz;
+
+-- Manual run notes:
+-- 1. Apply on staging first (python -m db.migrate with the staging SUPABASE_DB_URL).
+-- 2. On prod the column already exists, so this only records the migration.
+-- 3. NULL approved_at = not yet approved; /allowlist/approve stamps it, /revoke nulls it.