Phase 1: billing usage visibility, Stripe portal, tenant-auth binding

DEPLOY.md

@@ -36,6 +36,7 @@ The backend image runs in three roles, all from `deploy/docker-compose.yml`:
 - `WORKER_ROLE` — set per-container in `docker-compose.yml` (`web`/`worker`/`scheduler`); no need to set in Doppler.
 - `STRIPE_PORTAL_CONFIG_ID` — optional `bpc_...`; empty uses the Stripe dashboard default portal config. Used by the Phase 1 billing portal endpoint. Create the portal configuration in Stripe (enable subscription update to the localmate price, payment-method update, cancellation) — one-time dashboard/API step.
 - `DASHBOARD_URL` — `https://localmate.crewcircle.co`; used as the Stripe portal `return_url` base.
+- `SUPABASE_JWT_SECRET` — **Phase 1 (required for billing).** The Supabase project's JWT secret (Settings → API → JWT Settings in the Supabase dashboard). Used to verify dashboard bearer tokens and derive `client_id` from the authenticated user (tenant-auth binding, C8/D20). When unset, the legacy `require_auth` falls back to anonymous for old callers, but the strict `/billing/*` endpoints reject with 401 — so it must be set in prd.
 
 ### Migrations (Phase 0)
 
@@ -64,6 +65,21 @@ It asserts, and FAILS the build otherwise:
 
 Record the gate's `ALL PHASE 0 STAGING GATE CHECKS PASSED` output on the PR before merge.
 
+### Phase 1 — billing visibility + tenant auth
+
+**Migration:** apply `supabase/migrations/013_user_client_map.sql` on top of `012`. It creates `user_client_map (user_id text primary key, client_id uuid references clients(id))` with RLS + a `service_role_all` policy (matches `001_clients.sql`). One row per Supabase auth user → owned client; populate it at signup (or backfill existing users to their `clients` row by email). Until a user has a row, `/billing/*` returns 403.
+
+**Tenant-auth binding (C8/D20):** `GET /billing/usage` and `POST /billing/portal` derive `client_id` from the authenticated identity via `user_client_map` (never from the request body/query), so a user can only reach their own tenant. `SUPABASE_JWT_SECRET` must be set in Doppler — with it empty these endpoints reject 401 (the legacy `require_auth` still falls back to anonymous for old non-client-scoped callers).
+
+**One-time Stripe Billing Portal configuration:**
+1. Stripe Dashboard → Settings → Billing → Customer portal → Add configuration (or via API: `stripe.billing_portal.Configuration.create(...)`).
+2. Enable: subscription update (to the localmate price), payment-method update, subscription cancellation.
+3. Copy the configuration id (`bpc_...`).
+4. Set `STRIPE_PORTAL_CONFIG_ID` in Doppler (`localmate/prd`). Leave empty to use the Stripe dashboard default portal config.
+5. `DASHBOARD_URL` controls where the user lands after leaving the portal (the `return_url`).
+
+Portal-driven subscription changes (plan/card/cancel) flow back through the existing Stripe webhook handlers (`customer.subscription.updated` / `deleted`) which already update `clients.subscription_status` — no new reconciliation path.
+
 
 ---
 

backend/.env.example

@@ -4,6 +4,8 @@
 SUPABASE_URL=
 SUPABASE_ANON_KEY=
 SUPABASE_SERVICE_ROLE_KEY=
+# Phase 1 — tenant auth (Supabase JWT secret; required for /billing/* client-scoped endpoints)
+SUPABASE_JWT_SECRET=
 STRIPE_SECRET_KEY=
 STRIPE_PRICE_ID=
 STRIPE_WEBHOOK_SECRET=

backend/jobs/trial_emails.py

@@ -1,8 +1,17 @@
 """Trial milestone emails — day1, day7, day13, and expired.
 
-APScheduler-compatible async job. Runs hourly (wired via CronTrigger(hour=9)
-in scheduler.py). Uses trial_emails_sent table for idempotency: one email
-per (client_id, day_number) pair, ever.
+APScheduler-compatible async job. Runs daily (wired via CronTrigger(hour=9) in
+scheduler.py, which enqueues the ``run_trial_emails_daily`` arq task). Uses the
+``trial_emails_sent`` table for idempotency: one email per (client_id,
+day_number) pair, ever.
+
+Outbound sends are durable (C4): instead of calling the Resend senders directly,
+this job enqueues a ``send_email_task`` arq job (the Phase 0 durable wrapper) so
+a Resend transport failure retries with backoff and dead-letters on exhaustion
+rather than being silently swallowed. The idempotency row is recorded BEFORE the
+enqueue and rolled back if the enqueue fails, so a failed enqueue can never leave
+a row that would block a future send (which would cause a silent drop); a
+permanently-failed send lands in ``dead_letter`` for operator replay.
 """
 
 import logging
@@ -11,22 +20,16 @@
 import pytz
 
 from db import get_db
-from services.resend_email import (
-    send_trial_day1_email,
-    send_trial_day7_email,
-    send_trial_day13_email,
-    send_trial_expired_email,
-)
 
 AEST = pytz.timezone("Australia/Sydney")
 logger = logging.getLogger(__name__)
 
-# days_since_trial_start -> (send_fn, needs_client_id_arg).
-# send_trial_expired_email differs: (to, business_name) only — handled separately.
-_DAY_DISPATCH: dict[int, tuple] = {
-    1:  (send_trial_day1_email, True),
-    7:  (send_trial_day7_email, True),
-    13: (send_trial_day13_email, True),
+# days_since_trial_start -> Resend email kind (see services.resend_email._content).
+# The expired email differs (no client_id arg) and is handled separately.
+_DAY_KIND: dict[int, str] = {
+    1:  "trial_day1",
+    7:  "trial_day7",
+    13: "trial_day13",
 }
 
 
@@ -48,7 +51,28 @@ def _record_send(db, client_id: str, day_number: int) -> None:
     ).execute()
 
 
-async def run_trial_emails() -> None:
+def _unrecord_send(db, client_id: str, day_number: int) -> None:
+    """Remove a tentatively-recorded idempotency row (enqueue-failure rollback)."""
+    (
+        db.table("trial_emails_sent")
+        .delete()
+        .eq("client_id", client_id)
+        .eq("day_number", day_number)
+        .execute()
+    )
+
+
+async def run_trial_emails(pool) -> None:
+    """Enqueue durable trial-milestone emails for all active-trial clients.
+
+    ``pool`` is the arq Redis pool (``ctx["redis"]`` from the
+    ``run_trial_emails_daily`` cron task). When absent the run is skipped — the
+    next run with a live pool picks it up.
+    """
+    if pool is None:
+        logger.warning("run_trial_emails: no arq pool available — skipping this run")
+        return
+
     db = get_db()
     now = datetime.now(AEST)
 
@@ -79,46 +103,55 @@ async def run_trial_emails() -> None:
             ).astimezone(AEST)
             days_since = (now.date() - trial_start.date()).days
 
-            if days_since in _DAY_DISPATCH:
-                fn, _needs_id = _DAY_DISPATCH[days_since]
+            if days_since in _DAY_KIND:
                 if not _already_sent(db, client_id, days_since):
+                    # Record the idempotency row FIRST so a crash between
+                    # enqueue-success and record does not cause a duplicate
+                    # delivery on the next run. If the enqueue fails, roll the
+                    # row back so the next run can retry (no silent drop).
+                    _record_send(db, client_id, days_since)
                     try:
-                        await fn(client["email"], client["business_name"], client_id)
-                        _record_send(db, client_id, days_since)
+                        await pool.enqueue_job(
+                            "send_email_task",
+                            _DAY_KIND[days_since],
+                            client["email"],
+                            client["business_name"],
+                            client_id,
+                        )
                         sent += 1
                         logger.info(
-                            "Sent day-%d email to %s (%s)",
+                            "Enqueued day-%d email to %s (%s)",
                             days_since, client["email"], client_id,
                         )
-                    except Exception as exc:
-                        logger.error(
-                            "send_trial_day%d_email failed for %s: %s",
-                            days_since, client_id, exc,
-                        )
+                    except Exception:
+                        _unrecord_send(db, client_id, days_since)
+                        raise
 
             raw_ends = client.get("trial_ends_at")
             if raw_ends:
                 trial_ends = datetime.fromisoformat(
                     raw_ends.replace("Z", "+00:00")
                 ).astimezone(AEST)
                 if now > trial_ends and not _already_sent(db, client_id, -1):
+                    # Record-first / rollback-on-enqueue-failure (see above).
+                    _record_send(db, client_id, -1)
                     try:
-                        await send_trial_expired_email(
-                            client["email"], client["business_name"]
+                        await pool.enqueue_job(
+                            "send_email_task",
+                            "trial_expired",
+                            client["email"],
+                            client["business_name"],
                         )
-                        _record_send(db, client_id, -1)
                         sent += 1
                         logger.info(
-                            "Sent expired email to %s (%s)",
+                            "Enqueued expired email to %s (%s)",
                             client["email"], client_id,
                         )
-                    except Exception as exc:
-                        logger.error(
-                            "send_trial_expired_email failed for %s: %s",
-                            client_id, exc,
-                        )
+                    except Exception:
+                        _unrecord_send(db, client_id, -1)
+                        raise
 
         except Exception as exc:
             logger.error("trial_emails: unexpected error for client %s: %s", client_id, exc)
 
-    logger.info("run_trial_emails: processed %d clients, sent %d emails", processed, sent)
+    logger.info("run_trial_emails: processed %d clients, enqueued %d emails", processed, sent)

backend/main.py

@@ -8,7 +8,7 @@
 from config import settings
 from db import init_db
 from scheduler import create_scheduler
-from routers import auth, webhooks, drafts
+from routers import auth, webhooks, drafts, billing
 
 logging.basicConfig(level=logging.INFO)
 
@@ -75,6 +75,7 @@ async def lifespan(app: FastAPI):
 app.include_router(auth.router, prefix="/auth")
 app.include_router(webhooks.router, prefix="/webhooks")
 app.include_router(drafts.router, prefix="/drafts")
+app.include_router(billing.router, prefix="/billing")
 try:
     from routers import approve
     app.include_router(approve.router, prefix="/approve")

backend/middleware/auth.py

@@ -1,21 +1,18 @@
 import logging
 import jwt
-from fastapi import Depends, HTTPException, Request
+from fastapi import HTTPException, Request
 from config import settings
+from db import get_db
 
 logger = logging.getLogger(__name__)
 
 
-async def verify_supabase_jwt(request: Request) -> dict:
-    """Verify Supabase JWT from Authorization Bearer header. Returns decoded payload.
+async def _decode_bearer(request: Request) -> dict:
+    """Extract and decode the Supabase JWT from the Authorization Bearer header.
 
-    When supabase_jwt_secret is empty (e.g. CI / local dev stubs), logs a
-    warning and returns an anonymous payload so tests don't break.
+    Raises 401 when the header is missing; 403 when the token is invalid.
+    Shared by :func:`verify_supabase_jwt` and :func:`verify_supabase_jwt_strict`.
     """
-    if not settings.supabase_jwt_secret:
-        logger.warning("SUPABASE_JWT_SECRET not configured — allowing anonymous access")
-        return {"sub": "anonymous", "email": None}
-
     auth_header = request.headers.get("Authorization")
     if not auth_header or not auth_header.startswith("Bearer "):
         raise HTTPException(status_code=401, detail="Missing auth")
@@ -33,4 +30,63 @@ async def verify_supabase_jwt(request: Request) -> dict:
     return payload
 
 
+async def verify_supabase_jwt(request: Request) -> dict:
+    """Verify Supabase JWT from Authorization Bearer header. Returns decoded payload.
+
+    When supabase_jwt_secret is empty (e.g. CI / local dev stubs), logs a
+    warning and returns an anonymous payload so tests don't break.
+    """
+    if not settings.supabase_jwt_secret:
+        logger.warning("SUPABASE_JWT_SECRET not configured — allowing anonymous access")
+        return {"sub": "anonymous", "email": None}
+
+    return await _decode_bearer(request)
+
+
 require_auth = verify_supabase_jwt
+
+
+async def verify_supabase_jwt_strict(request: Request) -> dict:
+    """Strict JWT verification — raises 401 when SUPABASE_JWT_SECRET is unset.
+
+    Unlike :func:`verify_supabase_jwt` (kept for backward compat), this variant
+    NEVER falls back to an anonymous payload when the secret is empty. It is the
+    basis for tenant-scoped endpoints (C8/D20) where an unauthenticated caller
+    must not reach client data.
+    """
+    if not settings.supabase_jwt_secret:
+        raise HTTPException(status_code=401, detail="Auth not configured")
+
+    return await _decode_bearer(request)
+
+
+async def require_client_id(request: Request) -> str:
+    """Derive the caller's ``client_id`` from the authenticated identity (C8/D20).
+
+    The client_id is resolved from the JWT subject via the ``user_client_map``
+    table — NEVER from the request body or query string — so a user can only
+    ever reach their own tenant. All Phase 1+ client-scoped endpoints
+    (``/billing/usage``, ``/billing/portal``, …) inject this as a dependency.
+
+    Raises:
+        401 — auth not configured, or missing/invalid token.
+        403 — authenticated user has no client binding (cross-tenant denial).
+    """
+    payload = await verify_supabase_jwt_strict(request)
+    user_id = payload.get("sub")
+    if not user_id:
+        raise HTTPException(status_code=401, detail="Invalid identity")
+
+    db = get_db()
+    resp = (
+        db.table("user_client_map")
+        .select("client_id")
+        .eq("user_id", user_id)
+        .maybe_single()
+        .execute()
+    )
+    if not resp or not resp.data:
+        # Authenticated but owns no client — deny (cross-tenant protection).
+        raise HTTPException(status_code=403, detail="No client bound to this user")
+
+    return resp.data["client_id"]