Phase 3: multi-location bi-directional menu sync, canonical store, Square OAuth, images

DEPLOY.md

@@ -10,13 +10,77 @@ 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.
+- `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)
+
+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 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.
+
+
 ---
 
 ## Phase A — You (one-time, ~10 min)

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=
@@ -17,6 +19,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

@@ -25,10 +25,28 @@ class Settings(BaseSettings):
     sentry_dsn:                str = ""
     project_id:                str = "localmate"
     environment:               str = "prod"
+    # Square — global token deprecated for production (per-client OAuth is
+    # authoritative via square_oauth.get_valid_token). Kept for sandbox/dev fallback.
     square_access_token:       str = ""
     square_environment:        str = "sandbox"
+    square_app_id:             str = ""
+    square_app_secret:         str = ""
+    square_oauth_redirect_path: str = "/auth/square-callback"
+    square_webhook_signature_key: str = ""
+    menu_images_bucket:        str = "menu-images"
     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_...)
+
+    # --- Phase 4: GBP Pub/Sub provisioning (D15-B full automation) ---
+    gcp_project_id:            str = ""       # GCP project hosting the Pub/Sub topic
+    gcp_sa_json:               str = ""       # service-account JSON key (topic-admin) for Pub/Sub REST
+    gbp_pubsub_topic_name:     str = "gbp-reviews"  # shared topic short name
+
     class Config:
         env_file = ".env.local"
         env_file_encoding = "utf-8"

backend/jobs/competitor_watch.py

@@ -7,16 +7,24 @@
 
 from db import get_db
 from services.claude import generate_competitor_brief
+from services.structured_extract import (
+    extract_structured,
+    detect_prices_from_text,
+    diff_structured,
+)
 from utils.retry import retry_on_failure
 
 logger = logging.getLogger(__name__)
 
 
 @retry_on_failure()
-async def snapshot_website(url: str) -> tuple[str, str]:
-    """Fetch a competitor URL, strip non-content elements, and return (md5_hash, clean_text).
+async def snapshot_website(url: str) -> tuple[str, str, str]:
+    """Fetch a competitor URL, strip non-content elements, and return (md5_hash, clean_text, raw_html).
 
-    Returns ("", "") when the key is missing or the response is empty.
+    The raw HTML is returned so the caller can run ``extract_structured`` on
+    the original (with JSON-LD ``<script>`` blocks intact, before stripping).
+
+    Returns ("", "", "") when the key is missing or the response is empty.
     """
     async with httpx.AsyncClient() as client:
         resp = await client.get(
@@ -26,45 +34,82 @@ async def snapshot_website(url: str) -> tuple[str, str]:
             timeout=15,
         )
         resp.raise_for_status()
-        soup = BeautifulSoup(resp.text, "html.parser")
+        raw_html = resp.text
+        soup = BeautifulSoup(raw_html, "html.parser")
         for tag in soup(["script", "style", "nav", "footer"]):
             tag.decompose()
         clean_text = " ".join(soup.get_text().split())
         md5_hex = hashlib.md5(clean_text.encode()).hexdigest()
-        return md5_hex, clean_text
+        return md5_hex, clean_text, raw_html
+
+
+def _build_structured(raw_html: str, clean_text: str) -> dict:
+    """Extract structured data from HTML, with regex text fallback for prices."""
+    structured = extract_structured(raw_html)
+    # Regex fallback: supplement prices when JSON-LD found none.
+    if not structured["prices"]:
+        text_prices = detect_prices_from_text(clean_text)
+        if text_prices:
+            structured["prices"] = text_prices
+    return structured
+
+
+def _format_structured_diff(diff: list[dict]) -> list[str]:
+    """Format structured diffs into human-readable lines for the brief."""
+    lines = []
+    for d in diff:
+        kind = d["kind"]
+        name = d["name"]
+        if kind == "changed":
+            lines.append(f"{name} price ${d['old']} → ${d['new']}")
+        elif kind == "added":
+            lines.append(f"{name} added at ${d['new']}")
+        elif kind == "removed":
+            lines.append(f"{name} removed (was ${d['old']})")
+    return lines
 
 
 async def detect_changes(client_id: str, competitor_url: str) -> dict | None:
-    """Compare the latest snapshot hash against a fresh fetch.
+    """Compare the latest snapshot against a fresh fetch.
 
     Returns None when the content is unchanged or the fetch failed.
-    Returns a dict with change details on difference, including the new snapshot row id.
-
-    The caller is responsible for calling ``run_competitor_snapshots_all_clients``
-    which invokes this per client/URL pair.
+    Returns a dict with change details on difference, including:
+      - ``structured_diff``: field-level diffs from JSON-LD / price extraction
+      - ``snapshot_id``: the new snapshot row id
+      - ``prev_text`` / ``curr_text``: text snippets (fallback for the brief)
     """
     db = get_db()
 
     last_resp = (
         db.table("competitor_snapshots")
-        .select("content_hash, content_text")
+        .select("content_hash, content_text, structured_data")
         .eq("client_id", client_id)
         .eq("competitor_url", competitor_url)
         .order("captured_at", desc=True)
         .limit(1)
         .execute()
     )
 
-    new_hash, new_text = await snapshot_website(competitor_url)
+    new_hash, new_text, raw_html = await snapshot_website(competitor_url)
     if not new_hash:
         return None
 
+    new_structured = _build_structured(raw_html, new_text)
+
     last_text = ""
+    last_structured: dict = {}
+    structured_diff: list[dict] = []
     if last_resp.data:
         last_row = last_resp.data[0]
         if last_row["content_hash"] == new_hash:
             return None
         last_text = last_row.get("content_text") or ""
+        last_structured = last_row.get("structured_data") or {}
+        try:
+            structured_diff = diff_structured(last_structured, new_structured)
+        except Exception as e:
+            logger.warning("diff_structured failed for %s: %s", competitor_url, e)
+            structured_diff = []
 
     insert_resp = (
         db.table("competitor_snapshots")
@@ -73,6 +118,8 @@ async def detect_changes(client_id: str, competitor_url: str) -> dict | None:
             "competitor_url": competitor_url,
             "content_hash": new_hash,
             "content_text": new_text,
+            "structured_data": new_structured,
+            "structured_diff": structured_diff,
         })
         .execute()
     )
@@ -84,6 +131,7 @@ async def detect_changes(client_id: str, competitor_url: str) -> dict | None:
         "prev_text": last_text,
         "curr_text": new_text,
         "snapshot_id": new_id,
+        "structured_diff": structured_diff,
     }
 
 
@@ -107,8 +155,11 @@ async def run_competitor_snapshots_all_clients() -> None:
 
     1. Snapshot each URL listed in ``competitor_urls``.
     2. If changes are detected, collect them into a list.
-    3. Call ``generate_competitor_brief`` once per changed client.
-    4. Persist the brief text and extracted threat level on each new snapshot row.
+    3. Build ``changes_summary`` from structured diffs first (concrete
+       field/value changes), falling back to text snippets when no structured
+       signal exists.
+    4. Call ``generate_competitor_brief`` once per changed client.
+    5. Persist the brief text and extracted threat level on each new snapshot row.
 
     Each client is wrapped in its own try/except so one failure never crashes
     the entire job run.
@@ -153,11 +204,20 @@ async def run_competitor_snapshots_all_clients() -> None:
 
             parts = []
             for c in changes:
-                parts.append(
-                    f"Competitor: {c['url']}\n"
-                    f"Previous snippet: {c['prev_text'][:300]}\n"
-                    f"Current snippet:  {c['curr_text'][:300]}"
-                )
+                # Build summary from structured diffs first (concrete, specific).
+                structured_lines = _format_structured_diff(c.get("structured_diff", []))
+                if structured_lines:
+                    parts.append(
+                        f"Competitor: {c['url']}\nStructured changes:\n"
+                        + "\n".join(structured_lines)
+                    )
+                else:
+                    # Fall back to text snippets when no structured signal exists.
+                    parts.append(
+                        f"Competitor: {c['url']}\n"
+                        f"Previous snippet: {c['prev_text'][:300]}\n"
+                        f"Current snippet:  {c['curr_text'][:300]}"
+                    )
             changes_summary = "\n\n".join(parts)
 
             brief = await _generate_brief_safe(business_name, changes_summary)