Admin Teams notifications and role change requests

api/ADMIN_NOTIFICATIONS.md

@@ -0,0 +1,110 @@
+# Admin notifications and role requests
+
+## What administrators are told about
+
+| Event | When | Where |
+|---|---|---|
+| New user | A user signs in for the first time (including a user an admin invited earlier) | Teams + email to admins |
+| Equipment awaiting approval | A non-admin registers equipment, or submits changes that need review (one notification each time it enters review; resubmitting while it is still pending does not notify again) | Teams + email to admins |
+| Role request | A user asks for a different role from their Settings page | Teams + email to admins |
+| Role decision | An admin approves or rejects a role request | Email to the requester (only if email is configured; they always see it on their Settings page) |
+
+Each notification links to the page where it is handled (`/admin/users`,
+`/admin/reviews`, `/admin/role-requests`, `/settings`) when
+`PLATFORM_BASE_URL` is set.
+
+## How it works
+
+Every event writes a row to the `platform_notifications` outbox in the same
+database transaction as the event, so a notification exists exactly when the
+event does, and a unique key stops duplicates (the sign-in check runs on every
+page load but notifies once per user). A background pass delivers new rows
+straight away, and each API worker runs a sweeper every
+`NOTIFICATION_SWEEP_SECONDS` (default 60) for anything missed. Each
+row is claimed on its own just before sending, under a lease and a claim
+token, so the two API workers never send the same row at once. Failed
+deliveries are retried with backoff (1, 2, 4 … minutes, at most an hour
+apart) up to `NOTIFICATION_MAX_ATTEMPTS` (default 10), then marked failed; if
+an SMTP server refuses only some recipients, only those are retried. Delivery
+is at-least-once: a crash between sending and recording can send twice. A
+channel that is not configured is skipped, not queued: configuring email later
+does not send old notifications. Delivered rows are deleted after
+`NOTIFICATION_RETENTION_DAYS` (default 90). Stored and logged errors are short
+categories (for example `HTTP 404` or `SMTPRecipientsRefused`) and never
+contain webhook URLs or credentials.
+
+Request handlers can only insert into the outbox; only the background worker
+role (`api_worker`) reads and updates it.
+
+## Configuration
+
+Set these in `dt-api-secret` (see `geddes/k8s/01-secrets.yaml.example`) and
+restart the API. Teams alone is a supported setup: with no email transport
+configured, email is skipped for every notification. `GET /api/health` reports
+`runtime.admin_teams_notifications_configured` and
+`runtime.email_notifications_configured`.
+
+### Teams
+
+`ADMIN_TEAMS_WEBHOOK_URL`: the HTTP trigger URL of a Power Automate flow (or
+Logic App), the same kind used for the retrain alerts:
+
+1. Power Automate → Create → Instant cloud flow → trigger **When a Teams
+   webhook request is received** (or **When a HTTP request is received**).
+2. Add **Post card in a chat or channel**; post as Flow bot to the channel or
+   chat you want, and set the card to the trigger body (`triggerBody()`). The
+   API sends an Adaptive Card.
+3. Save and copy the trigger URL into `ADMIN_TEAMS_WEBHOOK_URL`.
+
+### Email
+
+Recipients are `ADMIN_NOTIFICATION_EMAILS` (comma separated) or, if unset,
+every active admin's email address in the platform. Choose one transport:
+
+- **SMTP relay:** `SMTP_HOST`, `SMTP_PORT` (default 587), `SMTP_SECURITY`
+  (`starttls` default, `ssl` or `none`; any other value is refused rather
+  than falling back to an unencrypted connection), `SMTP_FROM`, and `SMTP_USERNAME` /
+  `SMTP_PASSWORD` if the relay requires login. Use a relay that accepts mail
+  from the Geddes cluster.
+- **Power Automate:** `EMAIL_WEBHOOK_URL`, a flow triggered by **When a HTTP
+  request is received** with the schema below, followed by **Send an email
+  (V2)** using `to` (joined with `;`), `subject` and `html` as the body. It
+  sends from the account that owns the flow.
+
+  ```json
+  {"type": "object", "properties": {
+    "to": {"type": "array", "items": {"type": "string"}},
+    "subject": {"type": "string"},
+    "text": {"type": "string"},
+    "html": {"type": "string"}}}
+  ```
+
+If both are set, SMTP is used.
+
+## Role requests
+
+Everyone joins as a Researcher. On **Settings → Your role** a user picks the
+role they need and explains why; they can have one pending request, can
+cancel it, and can send at most three requests a day. Administrators decide on **Admin → Role Requests**, optionally with
+a note. Approval applies the same rules as changing a role on the Users page
+(an admin cannot demote themselves or the last admin) and is recorded in the
+role audit. Changing the role of an invited user who has not signed in yet
+keeps them "invited", so their first sign-in still notifies the admins. The requester sees the decision on their Settings page (and receives an
+email when an email transport is configured), and the web app picks up the
+new role on their next page load.
+
+Row-level security on `role_change_requests` (always enforced) lets users see
+and cancel only their own requests; only admins see all requests and decide
+them. When duplicate accounts are merged, their requests move to the
+remaining account and at most one stays pending.
+
+The sign-in identity sync (`GET /api/admin/users/{email}/role`) is
+server-to-server only: the web proxy refuses it and the API rejects calls
+that carry a user identity, so users cannot create or merge other people's
+accounts or fake sign-in notifications.
+
+## Deploying
+
+The new tables are created by the schema migrations: run the
+`dt-db-migrate` Job (`geddes/k8s/09-db-migrate-job.yaml`) with the new image
+before rolling out the API.

api/db_migrations.py

@@ -44,7 +44,7 @@
 
 
 # Reported by /api/health.
-RLS_REPORTED_TABLES = ("projects", "project_members") + RLS_ENFORCED_TABLES
+RLS_REPORTED_TABLES = ("projects", "project_members", "role_change_requests") + RLS_ENFORCED_TABLES
 
 
 def rls_enforcement_mode() -> str:
@@ -190,6 +190,8 @@ def run_schema_migrations() -> None:
         ensure_user_role_audit_table_pg,
     )
     from ml_response_cache import ensure_ml_endpoint_cache_table_pg
+    from platform_notifications import ensure_platform_notifications_table_pg
+    from role_requests import ensure_role_change_requests_table_pg
 
     lock_conn = get_pg_migration_connection()
     try:
@@ -220,6 +222,8 @@ def run_schema_migrations() -> None:
         ensure_data_uploads_kind_column_pg()
         ensure_data_uploads_processing_started_at_pg()
         ensure_ml_endpoint_cache_table_pg()
+        ensure_platform_notifications_table_pg()
+        ensure_role_change_requests_table_pg()
         # Last: the policies reference tables and columns created above.
         ensure_row_level_security_pg()
     finally:

api/main.py

@@ -87,6 +87,10 @@ def _runtime_informational_checks() -> dict[str, bool]:
     return {
         "postgres_superuser_password_configured": _env_present("PG_SUPER_PASS"),
         "postgres_worker_role_configured": _env_present("PG_WORKER_PASS"),
+        "admin_teams_notifications_configured": _env_present("ADMIN_TEAMS_WEBHOOK_URL"),
+        "email_notifications_configured": (
+            _env_present("SMTP_HOST") or _env_present("EMAIL_WEBHOOK_URL")
+        ),
     }
 
 
@@ -177,14 +181,16 @@ def _run_startup_migrations() -> None:
 
 
 def _start_background_jobs() -> None:
-    """Recover interrupted background work and start the proposal job sweeper."""
+    """Recover interrupted background work and start the background sweepers."""
     from metadata_pg import recover_stale_db3_uploads_pg, start_proposal_job_sweeper
+    from platform_notifications import start_notification_sweeper
 
     try:
         recover_stale_db3_uploads_pg()
     except Exception:
         logger.warning("Could not recover stale DB3 uploads at startup", exc_info=True)
     start_proposal_job_sweeper()
+    start_notification_sweeper()
 
 
 # ── Startup / shutdown ───────────────────────────────────────────────────────

api/metadata_pg.py

@@ -1445,6 +1445,42 @@ def get_equipment_pg(domain_id: str) -> dict[str, Any] | None:
         conn.close()
 
 
+def _enqueue_equipment_review_notification(
+    cur,
+    *,
+    registration_id: str,
+    domain_id: str,
+    equipment_name: str,
+    user: PlatformUser,
+    is_change: bool,
+) -> None:
+    """Tell administrators an equipment registration is waiting for review."""
+    from platform_notifications import enqueue_notification
+
+    action = "changes to" if is_change else "a new registration for"
+    enqueue_notification(
+        cur,
+        kind="equipment_review_requested",
+        # Called once per review cycle (see the callers); registrations are
+        # upserted, so the id alone would repeat across cycles.
+        dedupe_key=f"equipment_review_requested:{registration_id}:{uuid.uuid4().hex}",
+        subject=f"Equipment awaiting approval: {equipment_name}",
+        message=(
+            f"{user.name or user.id} submitted {action} {equipment_name}. "
+            "It needs an administrator's review before it can be used."
+        ),
+        facts=[
+            ("Equipment", equipment_name),
+            ("Equipment ID", domain_id),
+            ("Submitted by", user.name or user.id),
+            ("Email", user.email),
+            ("Organization", user.organization),
+            ("Type", "Changes to existing equipment" if is_change else "New registration"),
+        ],
+        link_path="/admin/reviews",
+    )
+
+
 def register_equipment_pg(
     *,
     payload: dict[str, Any],
@@ -1607,10 +1643,15 @@ def register_equipment_pg(
             )
 
             cur.execute(
-                "SELECT id FROM equipment_registrations WHERE domain_id = %s",
+                # FOR UPDATE serializes concurrent submissions for this equipment;
+                # the previous status decides whether a new review cycle starts.
+                "SELECT id, status FROM equipment_registrations WHERE domain_id = %s FOR UPDATE",
                 (domain_id,),
             )
             existing_registration = cur.fetchone()
+            already_pending_review = bool(
+                existing_registration and existing_registration.get("status") == "pending"
+            )
             features_count = len(features)
             if existing_registration:
                 cur.execute(
@@ -1683,7 +1724,24 @@ def register_equipment_pg(
                 )
                 registration_row = cur.fetchone()
 
+        # Notify when a registration enters review. A resubmission while it is
+        # still pending coalesces into the open review the admins were
+        # already told about.
+        if record_status == "pending" and registration_row and not already_pending_review:
+            with conn.cursor() as cur:
+                _enqueue_equipment_review_notification(
+                    cur,
+                    registration_id=str(registration_row["id"]),
+                    domain_id=domain_id,
+                    equipment_name=str(payload.get("name") or domain_id),
+                    user=user,
+                    is_change=False,
+                )
         conn.commit()
+        if record_status == "pending" and registration_row and not already_pending_review:
+            from platform_notifications import kick_notification_delivery
+
+            kick_notification_delivery()
         return {
             "registration_id": str(registration_row["id"]) if registration_row else None,
             "domain_id": domain_id,
@@ -1889,10 +1947,15 @@ def update_equipment_pg(
             )
 
             cur.execute(
-                "SELECT id FROM equipment_registrations WHERE domain_id = %s",
+                # FOR UPDATE serializes concurrent submissions for this equipment;
+                # the previous status decides whether a new review cycle starts.
+                "SELECT id, status FROM equipment_registrations WHERE domain_id = %s FOR UPDATE",
                 (domain_id,),
             )
             existing_registration = cur.fetchone()
+            already_pending_review = bool(
+                existing_registration and existing_registration.get("status") == "pending"
+            )
             if existing_registration:
                 cur.execute(
                     """
@@ -1952,7 +2015,24 @@ def update_equipment_pg(
                 )
                 registration_row = cur.fetchone()
 
+        # Notify when a registration enters review. A resubmission while it is
+        # still pending coalesces into the open review the admins were
+        # already told about.
+        if record_status == "pending" and registration_row and not already_pending_review:
+            with conn.cursor() as cur:
+                _enqueue_equipment_review_notification(
+                    cur,
+                    registration_id=str(registration_row["id"]),
+                    domain_id=domain_id,
+                    equipment_name=str(payload.get("name") or domain_id),
+                    user=user,
+                    is_change=True,
+                )
         conn.commit()
+        if record_status == "pending" and registration_row and not already_pending_review:
+            from platform_notifications import kick_notification_delivery
+
+            kick_notification_delivery()
         return {
             "registration_id": str(registration_row["id"]) if registration_row else None,
             "domain_id": domain_id,