Phase 0: arq task queue, durable webhooks, single-active scheduler

DEPLOY.md

@@ -10,13 +10,61 @@ Target domain: **localmate.crewcircle.co** (dashboard) / **api.localmate.crewcir
 | Layer | Host | Domain | Trigger |
 |---|---|---|---|
 | Next.js dashboard | **Vercel** | `localmate.crewcircle.co` | GitHub Actions `deploy.yml` on push to `main` |
-| FastAPI + APScheduler | **Docker Compose + Caddy** on shared DigitalOcean Sydney droplet (`170.64.183.45`) | `api.localmate.crewcircle.co` | `scripts/deploy_backend.sh` via Doppler |
+| FastAPI (web) + arq worker + scheduler + Redis | **Docker Compose + Caddy** on shared DigitalOcean Sydney droplet (`170.64.183.45`) | `api.localmate.crewcircle.co` | `scripts/deploy_backend.sh` via Doppler |
 | Postgres | **Supabase** | `*.supabase.co` | Pulumi provisioner |
 | Secrets | **Doppler** `localmate/prd` (inherits `crewcircle-master/prod`) | — | Doppler CLI |
 | DNS | **Cloudflare** | `crewcircle.co` zone | `scripts/finish_infra.sh` via Doppler |
 
 The backend shares the same droplet as **TaxFlowAI** — Caddy routes `api.localmate.crewcircle.co` to the localmate backend container and `api.taxflow.crewcircle.com.au` to the taxflow backend container. No new droplet needed.
 
+### Task queue containers (Phase 0)
+
+The backend image runs in three roles, all from `deploy/docker-compose.yml`:
+
+| Container | `WORKER_ROLE` | Command | Purpose |
+|---|---|---|---|
+| `localmate-backend` | `web` | `uvicorn main:app` | FastAPI API; enqueues jobs onto arq. Does **not** run APScheduler. |
+| `localmate-worker` | `worker` | `arq worker.WorkerSettings` | Executes queued tasks (inbound webhook processing + durable outbound sends). Scale with `docker compose up -d --scale localmate-worker=N`. |
+| `localmate-scheduler` | `scheduler` | `uvicorn main:app` | **Single-active** enqueue-only APScheduler (NOT HA). Each cron trigger pushes an arq job. Must remain a single instance so crons fire exactly once. |
+| `localmate-redis` | — | `redis-server --appendonly yes --maxmemory 128mb --maxmemory-policy noeviction` | arq broker/result store. Persistent (AOF), non-evicting, memory-capped (128MB) on the shared 1GB droplet. Named volume `localmate-redis-data`. |
+
+**Durability:** inbound webhooks (Stripe/GBP/menu) are persisted to `webhook_events` before enqueue; a reconciliation job (`reconcile_webhooks`, every 5 min) re-enqueues rows stuck `pending`. Exhausted retries (inbound and outbound) land in `dead_letter`. Redis is non-evicting so job state is never silently dropped.
+
+### New Doppler vars (`localmate/prd`)
+
+- `REDIS_URL` — `redis://localmate-redis:6379/0` (compose service DNS name). **Required.**
+- `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.
+
+### Migrations (Phase 0)
+
+Apply `supabase/migrations/010_webhook_events.sql`, `011_dead_letter.sql`, `012_booking_credentials.sql` on top of `009`. Each new table has RLS enabled with a `service_role_all` policy (matches `001_clients.sql`). The sequence applies cleanly on top of `009` and is reversible (drop the two new tables / the added `clients` columns).
+
+### Rollback (queue)
+
+`docker compose stop localmate-worker localmate-scheduler localmate-redis` reverts to web-only. Inbound webhooks still persist to `webhook_events` as `pending` (nothing crashes); processing resumes when the worker + Redis come back and the reconciler re-enqueues the backlog.
+
+### Phase 0 staged integration gate (C10 — MUST PASS before merge)
+
+Unit tests mock Redis/Postgres, which cannot prove the durability behaviour. Before the Phase 0 PR merges to prod, run the executable gate — it spins up throwaway `postgres:15-alpine` + `redis:7-alpine` containers and **exits non-zero on any failed acceptance check** (nothing to eyeball):
+
+```bash
+bash scripts/phase0_staging_gate.sh
+```
+
+It asserts, and FAILS the build otherwise:
+
+1. **Migrations** — Phase 0 migrations `010`/`011`/`012` apply cleanly on the reconstructed baseline; `webhook_events` + `dead_letter` exist with a `service_role_all` RLS policy; `012` added the five booking-credential columns on `clients`. (A pre-existing baseline-migration quirk already live in prod is a warning, not a Phase 0 failure.)
+2. **Worker drain / restart** — an enqueued inbound job is drained by a real arq worker to `status='done'`.
+3. **Retry + dead-letter** — a permanently-failing job is retried up to `MAX_TRIES` (real `arq.Retry` semantics) and then lands as exactly one `dead_letter` row.
+4. **Reconciliation** — a stale `pending` row is re-enqueued and a stale `processing` lease is reset to `pending` and re-enqueued.
+5. **Exactly-once / dedupe** — enqueuing the same event twice with the deterministic `_job_id` yields a single job (arq drops the duplicate).
+6. **Redis durability (D9-A)** — the gate refuses to pass unless Redis is `noeviction` + AOF, so job state is never silently dropped.
+
+Record the gate's `ALL PHASE 0 STAGING GATE CHECKS PASSED` output on the PR before merge.
+
+
 ---
 
 ## Phase A — You (one-time, ~10 min)

backend/.env.example

@@ -17,6 +17,11 @@ DATAFORSEO_LOGIN=
 DATAFORSEO_PASSWORD=
 GBP_CLIENT_ID=
 GBP_CLIENT_SECRET=
+# Phase 0 — task queue / worker / billing portal
+REDIS_URL=redis://localhost:6379/0
+WORKER_ROLE=web
+STRIPE_PORTAL_CONFIG_ID=
+DASHBOARD_URL=http://localhost:3000
 BASE_DOMAIN=crewcircle.com.au
 PROJECT_ID=local-biz-au
 ENVIRONMENT=prod

backend/Dockerfile

@@ -22,7 +22,15 @@ COPY . .
 #   DATAFORSEO_LOGIN, DATAFORSEO_PASSWORD
 #   GBP_CLIENT_ID, GBP_CLIENT_SECRET
 #   BASE_DOMAIN, PROJECT_ID
+#   REDIS_URL                (arq broker; e.g. redis://localmate-redis:6379/0)
+#   WORKER_ROLE              (web|worker|scheduler; set per-container in compose)
+#   STRIPE_PORTAL_CONFIG_ID  (optional; Stripe billing portal config bpc_...)
+#   DASHBOARD_URL            (optional; portal return_url base)
 #   SENTRY_DSN (optional)
+#
+# Run modes (same image, different command):
+#   web / scheduler : uv run uvicorn main:app --host 0.0.0.0 --port 8000
+#   worker          : uv run arq worker.WorkerSettings
 
 EXPOSE 8000
 

backend/config.py

@@ -29,6 +29,12 @@ class Settings(BaseSettings):
     square_environment:        str = "sandbox"
     supabase_jwt_secret:       str = ""
 
+    # --- Phase 0: queue / worker / billing-portal infra ---
+    redis_url:                 str = "redis://localhost:6379/0"
+    worker_role:               str = "web"   # "web" | "worker" | "scheduler"
+    dashboard_url:             str = ""       # Stripe portal return_url base
+    stripe_portal_config_id:   str = ""       # Stripe portal configuration id (bpc_...)
+
     class Config:
         env_file = ".env.local"
         env_file_encoding = "utf-8"

backend/main.py

@@ -23,11 +23,38 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     await init_db()
-    scheduler = create_scheduler()
-    scheduler.start()
-    app.state.scheduler = scheduler
+    app.state.scheduler = None
+    app.state.arq = None
+
+    from task_queue import get_arq_pool
+
+    log = logging.getLogger(__name__)
+    if settings.worker_role == "scheduler":
+        # Dedicated single-active scheduler container: enqueue-only APScheduler
+        # + an arq pool to push jobs. NOT HA — must be a single instance (C4).
+        app.state.arq = await get_arq_pool()
+        scheduler = create_scheduler()
+        scheduler.start()
+        app.state.scheduler = scheduler
+        log.info("Started enqueue-only scheduler (role=scheduler)")
+    else:
+        # web role: create an arq pool for enqueuing from request handlers.
+        # Do NOT start APScheduler here (prevents duplicate cron fire across
+        # web replicas). The 'worker' role runs via the arq CLI, not uvicorn.
+        try:
+            app.state.arq = await get_arq_pool()
+        except Exception as e:
+            log.warning("arq pool init failed (web role): %s", e)
+
     yield
-    scheduler.shutdown()
+
+    if app.state.scheduler is not None:
+        app.state.scheduler.shutdown()
+    if app.state.arq is not None:
+        try:
+            await app.state.arq.close()
+        except Exception:
+            pass
 
 
 app = FastAPI(title="LocalMate", lifespan=lifespan)

backend/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "stripe>=9.0.0",
     "httpx>=0.27.0",
     "apscheduler>=3.10.0",
+    "arq>=0.26.0",
     "pytz>=2024.1",
     "pydantic-settings>=2.0.0",
     "resend>=2.0.0",
@@ -26,4 +27,9 @@ dev-dependencies = [
     "pytest>=8.0.0",
     "pytest-asyncio>=0.23.0",
     "httpx>=0.27.0",
+    "asyncpg>=0.29.0",
+    "psycopg[binary]>=3.1.0",
 ]
+
+[tool.pytest.ini_options]
+pythonpath = ["."]