podium-multi-location-router

What this skill does

# Podium Multi-Location Router

## Overview

Route Podium API calls across multiple physical locations and operate the routing layer in production. This is not a tutorial on Podium's location model — it is the per-location dispatch code your integration runs when Sydney and Burleigh Heads share a single OAuth app, when an agency onboards five new stores in one afternoon, and when a compliance auditor walks in and asks "which location's webchat received that contact write at 14:07 UTC last Tuesday."

The six production failures this skill prevents:

1. **Writes to wrong location silently** — Sydney's contact write lands on Burleigh Heads' contact list because the calling code resolved `location_uid` from a stale lookup or hard-coded the wrong UID. The Podium API accepts it, returns 200, and the data sits in the wrong location with no error surface. Naive integrations discover this when a customer reports "I'm getting review requests for a store I've never been to."
2. **Credential cross-contamination** — Sydney's OAuth token gets reused for a write to Burleigh Heads. The Podium API does not reject this because the OAuth app is org-scoped and both locations live under the same org. The write succeeds; the audit trail attributes it to Sydney's credential; post-incident forensics cannot tell whether the write was authorized by Sydney's operator or by a bug.
3. **Audit trail missing location-ID** — log lines say `wrote contact name=Jane Doe` without saying which `location_uid` received the write. When a compliance question lands months later, the integration cannot answer "which location received this customer's data?" The answer determines GDPR data-subject scope and PCI cardholder-data exposure.
4. **Bulk onboarding race condition** — onboarding 5 new locations in one operation spawns 5 OAuth authorization flows. One fails partway (a user closed the consent tab, a redirect URI mismatched), the orchestrator continues with the rest, and the credentials map ends up with 4 valid entries + 1 dangling half-record. Subsequent operations against the half-onboarded location either crash or, worse, fall through to a default credential and write to the wrong place.
5. **Location-ID verification skipped on write** — the application code passes an arbitrary `location_uid` into the Podium call. Podium accepts any well-formed UID and returns 403 only if the current token genuinely has no access. The 403 is swallowed by an upstack handler ("just a stale auth, will retry"), and the integration silently stops working for one location while the rest keep flowing.
6. **Rate-limits not isolated per location** — the integration uses a single shared token bucket. Sydney runs a holiday review-request burst at 2pm and burns the org-wide quota; Burleigh Heads' webchats start returning 429 with no explanation visible to the Burleigh Heads operator who has done nothing wrong.

## Prerequisites

- Python 3.10+
- `podium-auth` skill installed; this skill consumes its `PodiumAuth` and `PodiumOrgRouter` classes
- `podium-rate-limit-survival` skill installed; this skill consumes its per-bucket budget primitive
- Podium OAuth app with `locations.read` scope granted (verifying a `location_uid` requires `GET /v4/locations`)
- A writable directory for the audit log (default `./audit-log/podium-router.jsonl`)
- A persistent location-credential map (file, AWS Secrets Manager, GCP Secret Manager, or SOPS-encrypted YAML)

## Instructions

Build in this order. Each section neutralizes one production failure mode.

### 1. Per-location credential isolation (neutralizes cross-contamination)

The router holds one `PodiumAuth` instance per `location_uid`, never per org. Two locations under the same org get two separate auth instances because the audit trail must attribute every call to a specific location and a specific credential record — even if the underlying OAuth refresh token happens to be shared at the org level.

```python
import asyncio, time
from dataclasses import dataclass, field
from typing import Optional
from podium_auth import PodiumAuth, PodiumOrgRouter, PodiumAuthError

@dataclass
class LocationCredential:
    location_uid: str
    org_slug: str
    client_id: str
    client_secret: str
    refresh_token_file: str
    verified_at: float = 0.0   # unix seconds; 0 = never verified

class LocationRouter:
    def __init__(self, credentials: list[LocationCredential], audit_log_path: str):
        self._creds: dict[str, LocationCredential] = {c.location_uid: c for c in credentials}
        self._auths: dict[str, PodiumAuth] = {}
        self._audit_log_path = audit_log_path
        self._verify_lock = asyncio.Lock()

    def get_client(self, location_uid: str) -> "PodiumLocationClient":
        cred = self._creds.get(location_uid)
        if not cred:
            raise UnknownLocationError(location_uid)
        if location_uid not in self._auths:
            self._auths[location_uid] = PodiumAuth(
                client_id=cred.client_id,
                client_secret=cred.client_secret,
                refresh_token=load_refresh_token(cred.refresh_token_file),
            )
        return PodiumLocationClient(
            location_uid=location_uid,
            auth=self._auths[location_uid],
            router=self,
        )
```

The map key is `location_uid`, never `org_slug`. Two locations under the same org get two map entries with the same `client_id`/`client_secret` but distinct `refresh_token_file` paths. This is the line that prevents Sydney's token from being reused for a Burleigh Heads write — the router has no syntactic path from a `location_uid` to a different location's auth.

### 2. Pre-flight location-ID verification (neutralizes silent wrong-location writes)

Before any write, the router confirms the `location_uid` exists in the set returned by `GET /v4/locations` for the current credential. Cache the verification with a short TTL — typically 1 hour — so a re-org of locations on the Podium side eventually propagates without re-verifying every call.

```python
VERIFY_TTL_SECONDS = 3600

async def ensure_location_verified(self, location_uid: str) -> None:
    cred = self._creds[location_uid]
    if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
        return
    async with self._verify_lock:
        if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
            return
        auth = self._auths[location_uid]
        token = await auth.get_token()
        async with httpx.AsyncClient(timeout=10) as c:
            r = await c.get(
                "https://api.podium.com/v4/locations",
                headers={"Authorization": f"Bearer {token}"},
            )
        if r.status_code != 200:
            raise LocationVerificationError(location_uid, r.status_code, r.text)
        ids = {loc["uid"] for loc in r.json().get("locations", [])}
        if location_uid not in ids:
            raise LocationNotInScopeError(location_uid, sorted(ids))
        cred.verified_at = time.time()
```

`LocationNotInScopeError` is the loud version of "Podium returned 403 silently." It fires before the call is made, so the caller cannot accidentally write to a location their token does not own. The 1-hour TTL keeps the verification cost negligible while bounding the staleness window.

### 3. Structured audit log on every call (neutralizes audit-trail gaps)

Every API call routed through this skill emits one JSON line to the audit log. The fields are fixed: `{ts, location_uid, org_slug, endpoint, method, status, request_id, latency_ms}`. No tokens, no request bodies, no PII — just the routing fingerprint.

```python
import json, os, time, uuid

def emit_audit(self, *, location_uid: str, endpoint: str, method: str,
               status: int, latency_ms: float, request_id: str) -> None:
    cred = self._creds[location_uid]
    record = {
        "ts": time.time(),
        "location_uid": location_uid,
        "org_slug": cred.org_slug,
        "endpoint": endpoint,
        "method"