---
name: talo-stack-python
description: Talo Pay bank-transfer payments (Argentina, ARS, CVU/alias) through the Talo Partner Hub in Python 3.12+, with FastAPI as the main path, Django 5.2 LTS / 6.x as the second and a Flask note. Covers an httpx2 hub client with typed errors and safe retries, server-side checkout with external_id = order id and a polling order page, raw-body HMAC-signed webhooks (await request.body(), request.body + csrf_exempt, request.get_data()), durable exactly-once handling in Postgres (unique event ids and a fulfil-once conditional update in one transaction), reconciliation by cron (script or management command), deploy notes for Railway, Render, Fly.io and a VPS, and pytest tests plus the sandbox checklist. Use when the project uses FastAPI, Django or Flask and the user wants to charge by bank transfer (CVU/alias, ARS) with Talo or Talo Pay.
---

# Talo Pay on python

This kit implements the 10-rule contract at https://talopartnerhub.vercel.app/docs/stacks.md in Python, on top of the core skills `talo-payments` (API, events) and `talo-go-live` (keys, sandbox plan, production). Hub: https://talopartnerhub.vercel.app (API facts: /docs/api.md, /docs/webhooks.md).
All code below ran (pytest, Django's test runner, a live uvicorn + mock hub) on Python 3.12 and 3.13, FastAPI 0.162 / Starlette 1.10, httpx2 2.16 (Pydantic's maintained continuation of httpx, same API; Starlette's TestClient prefers it), psycopg 3.3 + psycopg-pool 3.3, Django 6.1 and 5.2 LTS, django-tasks-db 0.13, Flask 3.1, Postgres 18.

## 1. Setup

```bash
pip install "fastapi[standard]" "psycopg[binary,pool]" httpx2                                # FastAPI (brings uvicorn, python-dotenv)
pip install "django>=5.2" "psycopg[binary]" httpx2 dj-database-url django-tasks-db gunicorn  # Django; on 5.2 also django-tasks
```

Django reads the URL with `DATABASES = {"default": dj_database_url.config(conn_max_age=600, conn_health_checks=True)}`.

```bash
TALO_HUB_API_KEY=tph_test_...      # npx talo-hub keys create --name <project> --scope charge --dotenv .env
TALO_HUB_WEBHOOK_SECRET=whsec_...  # printed once by: npx talo-hub webhooks add https://<host>/webhooks/talo
DATABASE_URL=postgresql://user:password@host:5432/shop
TALO_HUB_DEV_FORWARD=1             # local .env only (section 7)
```

Server-side only: `.env` locally (gitignore `.env` and `.env.*`; commit a value-less `.env.example`), the host's env vars when deployed; read with `os.environ`, never in templates, JS or logs. The browser only gets `FIELDS` (status, cvu, alias, amount, paid, expires_at).

## 2. Hub client (server-only)

`shop/talo_hub.py` (Django: `payments/talo_hub.py`; one file for every framework):
```python
"""Talo Partner Hub client + webhook verifier. Server-side only."""
import asyncio, hashlib, hmac, json, os, re, time
from decimal import Decimal
from urllib.parse import quote

import httpx2

HUB_URL = "https://talopartnerhub.vercel.app"
PAID = frozenset({"SUCCESS", "OVERPAID"})  # fulfil once; UNDERPAID is NOT paid
RANK = {"PENDING": 0, "UNDERPAID": 1, "EXPIRED": 2, "SUCCESS": 3, "OVERPAID": 4}  # statuses only move forward
TIMEOUT = httpx2.Timeout(8.0, connect=3.0)
CREATE_TIMEOUT = httpx2.Timeout(25.0, connect=3.0)  # Talo can take 15-20 s to create a payment

class TaloHubError(Exception):  # the hub answered { "error": { code, message, details? } }
    def __init__(self, status: int, code: str, message: str, details: dict | None = None):
        super().__init__(f"Talo Partner Hub {status} {code}: {message}")
        self.status, self.code, self.details = status, code, details or {}

def _request(method: str, path: str, body: dict | None = None, timeout: httpx2.Timeout = TIMEOUT) -> dict:  # KeyError if no key
    return {"method": method, "url": path, "json": body, "timeout": timeout, "headers": {"Authorization": f"Bearer {os.environ['TALO_HUB_API_KEY']}"}}

def _parse(res: httpx2.Response) -> dict:
    try:
        body = res.json()
    except ValueError:
        body = None
    if res.is_success and isinstance(body, dict):
        return body
    err = (body.get("error") if isinstance(body, dict) else None) or {}
    raise TaloHubError(res.status_code, err.get("code", "unknown"), err.get("message", res.reason_phrase), err.get("details"))

def _backoff(exc: Exception, attempt: int, attempts: int) -> float:  # retry network errors and 502 only
    if attempt == attempts or not (isinstance(exc, httpx2.TransportError) or getattr(exc, "status", 0) == 502):
        raise exc
    return 0.5 * 2 ** (attempt - 1)

class TaloHub:  # Django, Flask, scripts: blocking
    client = httpx2.Client

    def __init__(self, transport: httpx2.BaseTransport | httpx2.AsyncBaseTransport | None = None):
        self.http = self.client(base_url=HUB_URL, timeout=TIMEOUT, transport=transport)

    def _call(self, method: str, path: str, body: dict | None = None, attempts: int = 3, timeout: httpx2.Timeout = TIMEOUT) -> dict:
        for attempt in range(1, attempts + 1):
            try:
                return _parse(self.http.request(**_request(method, path, body, timeout)))
            except (httpx2.TransportError, TaloHubError) as exc:
                time.sleep(_backoff(exc, attempt, attempts))

    def create_payment(self, amount: Decimal, external_id: str, motive: str | None = None):
        body = {"amount": float(amount.quantize(Decimal("0.01"))), "external_id": external_id}  # same id on retry
        return self._call("POST", "/api/v1/payments", body | {"motive": motive} if motive else body, timeout=CREATE_TIMEOUT)

    def get_payment(self, payment_id: str, attempts: int = 3):
        return self._call("GET", f"/api/v1/payments/{quote(payment_id, safe='')}", attempts=attempts)

class AsyncTaloHub(TaloHub):  # FastAPI: same methods, awaited (`await hub.get_payment(...)`)
    client = httpx2.AsyncClient

    async def _call(self, method: str, path: str, body: dict | None = None, attempts: int = 3, timeout: httpx2.Timeout = TIMEOUT) -> dict:
        for attempt in range(1, attempts + 1):
            try:
                return _parse(await self.http.request(**_request(method, path, body, timeout)))
            except (httpx2.TransportError, TaloHubError) as exc:
                await asyncio.sleep(_backoff(exc, attempt, attempts))
```

Retries only network errors (`httpx2.TransportError`, timeouts included) and `502 talo_unavailable`, with the same external_id, so a retried create returns the same payment. Anything else raises `TaloHubError` with `.status` / `.code` (`unauthorized`, `insufficient_scope`, `invalid_amount`, `talo_rejected` + `.details["talo_message"]`, `talo_reconnect_required`: the account owner runs `npx talo-hub login` again): log it, show the buyer a generic "try again". The webhook re-read uses `attempts=1` to answer within the hub's 10 s.

## 3. Checkout (FastAPI)

```python
import logging, os
from contextlib import asynccontextmanager

import httpx2
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from psycopg_pool import AsyncConnectionPool

from .payments import apply_payment, send_receipt
from .talo_hub import AsyncTaloHub, TaloHubError, read_event

log = logging.getLogger("talo")
FIELDS = ("status", "cvu", "alias", "amount", "paid", "expires_at")  # all the browser ever gets

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with AsyncConnectionPool(os.environ["DATABASE_URL"], kwargs={"autocommit": True}) as pool:
        app.state.pool, app.state.hub = pool, AsyncTaloHub()
        yield
        await app.state.hub.http.aclose()

app = FastAPI(lifespan=lifespan)

def external_id(order_id: int, attempt: int) -> str:  # order-1001, then order-1001-2 after EXPIRED ...
    return f"order-{order_id}" if attempt == 1 else f"order-{order_id}-{attempt}"

@app.post("/orders/{order_id}/pay")
async def pay(order_id: int, request: Request):  # also check the order belongs to this session/user
    pool, hub = request.app.state.pool, request.app.state.hub
    async with pool.connection() as conn:
        cur = await conn.execute("SELECT amount, payment_status, payment_attempt FROM orders WHERE id = %s", (order_id,))
        if (order := await cur.fetchone()) is None:
            raise HTTPException(404, "order not found")
    amount, status, attempt = order  # the amount comes from YOUR database, never from the request
    attempt += status == "EXPIRED"  # an expired payment can't be paid: new payment, new external_id
    try:
        p = await hub.create_payment(amount, external_id(order_id, attempt), motive=f"Pedido #{order_id}")
        if p["status"] == "EXPIRED":  # the hub knew before its webhook reached us: next external_id now
            attempt += 1
            p = await hub.create_payment(amount, external_id(order_id, attempt), motive=f"Pedido #{order_id}")
    except (TaloHubError, httpx2.HTTPError):
        log.exception("could not create the Talo payment for order %s", order_id)
        raise HTTPException(503, "No pudimos generar los datos de pago. Probá de nuevo en un momento.")
    async with pool.connection() as conn:
        await conn.execute(
            "UPDATE orders SET payment_id = %s, payment_status = %s, payment_attempt = %s, cvu = %s, alias = %s,"
            " expires_at = %s WHERE id = %s",
            (p["id"], p["status"], attempt, p["cvu"], p["alias"], p["expires_at"], order_id),
        )
    return {k: p.get(k) for k in FIELDS}

@app.get("/orders/{order_id}/payment")  # polled by the order page (same ownership check); never fulfils
async def payment_status(order_id: int, request: Request):
    async with request.app.state.pool.connection() as conn:
        cur = await conn.execute("SELECT payment_id FROM orders WHERE id = %s AND payment_id IS NOT NULL", (order_id,))
        if (row := await cur.fetchone()) is None:
            raise HTTPException(404, "no payment for this order")
    try:
        p = await request.app.state.hub.get_payment(row[0], attempts=1)
    except (TaloHubError, httpx2.HTTPError):
        raise HTTPException(503, "status unavailable, try again")
    return {k: p.get(k) for k in FIELDS}
```

Order page (any template engine), polling every 3 s and backing off to 10 s, stopping at a final status:
```html
<p>Transferí <strong>{{ amount }}</strong> al CVU <code>{{ cvu }}</code> (alias <code>{{ alias }}</code>) antes del {{ expires_at }}.</p>
<p id="status">Esperando la transferencia…</p>
<script>
  const ars = new Intl.NumberFormat("es-AR", { style: "currency", currency: "ARS" });
  const show = (text) => (document.getElementById("status").textContent = text);
  async function poll(orderId, delay = 3000) {
    try {
      const p = await (await fetch(`/orders/${orderId}/payment`, { cache: "no-store" })).json();
      if (p.status === "UNDERPAID") show(`Recibimos ${ars.format(p.paid)}: faltan ${ars.format(p.amount - p.paid)}.`);
      if (p.status === "SUCCESS" || p.status === "OVERPAID") return show("¡Pago recibido! Ya preparamos tu pedido.");
      if (p.status === "EXPIRED") return show("El pago venció: generá uno nuevo."); // POST /orders/<id>/pay again
    } catch {} // network blip: keep polling
    setTimeout(() => poll(orderId, Math.min(delay * 1.5, 10000)), delay);
  }
  poll({{ order.id }});
</script>
```

After EXPIRED a "generar nuevo pago" button POSTs `/orders/<id>/pay` again: `payment_attempt` goes up and the external_id becomes `order-<id>-2`, also when the hub answers EXPIRED before its webhook arrived. If the expired payment had `paid > 0`, that money arrived: refund it (`POST /api/v1/payments/{id}/refunds`) or charge only `amount - paid`, never the full amount twice. The page only displays; the webhook and the reconciliation job fulfil, never the browser.

## 4. Webhook endpoint

End of `talo_hub.py`: the verifier and the framework-neutral first steps (verify, parse, ack non-payment events):
```python
_T, _V1 = re.compile(r"[0-9]{1,12}"), re.compile(r"[0-9a-f]{64}")

def verify_signature(raw_body: bytes, header: str | None, secret: str, *, now: float | None = None) -> bool:
    """talo-hub-signature: t=<unix>,v1=hex(HMAC-SHA256(secret, b"<t>." + raw body)); |now - t| <= 300 s."""
    pairs = [part.strip().partition("=") for part in (header or "").split(",")]
    fields = {key: value for key, _, value in pairs}
    if len(fields) != len(pairs) or not _T.fullmatch(fields.get("t", "")) or not _V1.fullmatch(fields.get("v1", "")):
        return False  # missing, repeated or malformed t / v1 (v1 = 64 lowercase hex)
    if abs((time.time() if now is None else now) - int(fields["t"])) > 300:
        return False
    expected = hmac.new(secret.encode(), fields["t"].encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, fields["v1"])

def accept_unsigned_forward(headers) -> bool:  # `events listen --forward-to` doesn't sign: dev flag + sandbox key only
    return (os.environ.get("TALO_HUB_DEV_FORWARD") == "1" and headers.get("talo-hub-forwarded") == "1"
            and os.environ.get("TALO_HUB_API_KEY", "").startswith("tph_test_"))

def read_event(raw: bytes, headers) -> tuple[int, int | None, str | None]:
    """Verify + parse, for any framework. Returns (status, event_id, payment_id): answer `status` now unless it is 0."""
    if not accept_unsigned_forward(headers):
        if not (secret := os.environ.get("TALO_HUB_WEBHOOK_SECRET")):
            return 500, None, None  # misconfigured: a 5xx makes the hub retry later
        if not verify_signature(raw, headers.get("talo-hub-signature"), secret):
            return 400, None, None
    try:
        event = json.loads(raw)
        event_id, event_type = int(event["id"]), str(event["type"])
        payment_id = event["data"]["payment"]["id"] if event_type.startswith("payment.") else None
    except (ValueError, KeyError, TypeError):
        return 400, None, None
    return (0 if payment_id else 200), event_id, payment_id  # 200 now: webhook.test, account.updated
```

FastAPI: raw bytes → `read_event` → re-read with GET (503 on failure: the hub retries) → one transaction → 2xx → slow work after the response (`BackgroundTasks` is in-process; work that must survive a crash goes to a queue such as Celery, RQ or arq):
```python
@app.post("/webhooks/talo")  # the exact path you register: no trailing slash, no auth dependency
async def talo_webhook(request: Request, background: BackgroundTasks):
    raw = await request.body()  # exact bytes: no Pydantic body model, no request.json() before this
    status, event_id, payment_id = read_event(raw, request.headers)
    if status:  # 400 bad signature/body, 500 secret missing, 200 non-payment event
        return JSONResponse({"received": status == 200}, status_code=status)
    try:
        payment = await request.app.state.hub.get_payment(payment_id, attempts=1)  # fresh data, not the snapshot
    except (TaloHubError, httpx2.HTTPError):
        log.exception("re-read of %s failed; the hub will retry event %s", payment_id, event_id)
        return JSONResponse({"received": False}, status_code=503)
    async with request.app.state.pool.connection() as conn:
        order_id = await apply_payment(conn, payment, event_id)
    if order_id is not None:  # fulfilled by THIS delivery
        background.add_task(send_receipt, order_id)  # runs after the response is sent
    return {"received": True}
```

What breaks it in FastAPI:
- A Pydantic body model or `await request.json()` instead of `await request.body()` (the HMAC covers the exact bytes), or a middleware that rewrites/decompresses bodies.
- Auth in front of the path: router-level `dependencies=[Depends(...)]`, session/auth middleware. Declare the route on the app itself.
- `HTTPSRedirectMiddleware`: uvicorn trusts `X-Forwarded-Proto` only from `--forwarded-allow-ips` (default 127.0.0.1), so behind Railway/Render/Fly it sees http and answers 307. Leave it out (the platform already forces https) or set `--forwarded-allow-ips` to the proxy.
- `redirect_slashes` (default on): POST `/webhooks/talo/` gets a 307. Register exactly the declared path.

Django, `payments/views.py`. Below the webhook, `pay` and `payment_status` port the FastAPI routes 1:1, EXPIRED re-check included (`get_object_or_404`, `services.hub.create_payment(order.amount, ...)`, `Order.objects.filter(pk=order.pk).update(..., expires_at=parse_datetime(p["expires_at"]))`), and keep CSRF on. URLs: `path("webhooks/talo", views.talo_webhook)` (register exactly that, no trailing slash), `path("orders/<int:order_id>/pay", views.pay)`, `path("orders/<int:order_id>/payment", views.payment_status)`.
```python
import logging

import httpx2
from django.contrib.auth.decorators import login_not_required
from django.http import JsonResponse
from django.shortcuts import get_object_or_404
from django.utils.dateparse import parse_datetime
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_GET, require_POST

from . import services
from .models import Order
from .talo_hub import TaloHubError, read_event

log = logging.getLogger("talo")
FIELDS = ("status", "cvu", "alias", "amount", "paid", "expires_at")  # all the browser ever gets

@csrf_exempt  # the hub has no CSRF token: the HMAC signature is the authentication
@login_not_required  # Django 5.1+: needed when LoginRequiredMiddleware is on
@require_POST
def talo_webhook(request):
    status, event_id, payment_id = read_event(request.body, request.headers)  # request.body = the raw bytes
    if status:  # 400 bad signature/body, 500 secret missing, 200 non-payment event
        return JsonResponse({"received": status == 200}, status=status)
    try:
        payment = services.hub.get_payment(payment_id, attempts=1)  # fresh data, not the snapshot
    except (TaloHubError, httpx2.HTTPError):
        log.exception("re-read of %s failed; the hub will retry event %s", payment_id, event_id)
        return JsonResponse({"received": False}, status=503)
    services.apply_payment(payment, event_id)  # commits; the receipt task is enqueued on commit
    return JsonResponse({"received": True})
```

What breaks it in Django:
- No `@csrf_exempt` → 403 (`CsrfViewMiddleware`); with `LoginRequiredMiddleware` (5.1+) and no `@login_not_required` → 302 to login.
- `APPEND_SLASH`: pattern `webhooks/talo/` hit at `/webhooks/talo` raises RuntimeError under DEBUG and answers 301 in production, losing the POST. Pattern and registered URL must be identical.
- Anything that reads the stream first (`request.read()` in a middleware) makes `request.body` raise `RawPostDataException`.
- Deploy settings: the host in `ALLOWED_HOSTS` (else 400 `DisallowedHost`); with `SECURE_SSL_REDIRECT = True` also `SECURE_PROXY_SSL_HEADER = ("HTTP_X_FORWARDED_PROTO", "https")`, or every delivery gets a 301.

Flask 3.1: in a `@app.post("/webhooks/talo")` view call `read_event(request.get_data(), request.headers)` (`get_data()` is the raw body), decorate it with `@csrf.exempt` if Flask-WTF's `CSRFProtect` is on, then follow the Django view with the blocking `TaloHub` and a sync copy of `apply_payment` (psycopg `Connection`, same SQL, `with conn.transaction():`).

## 5. Durable storage

`schema.sql` (psycopg path; run it from your migration tool). Money is `numeric(12, 2)` → `Decimal`, never float:
```sql
CREATE TABLE IF NOT EXISTS orders (
  id              bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  amount          numeric(12, 2) NOT NULL CHECK (amount > 0),  -- ARS, set by your server from the cart
  payment_id      text UNIQUE,                                 -- "PAR-talo-hub-<external_id>"
  payment_status  text,                                        -- PENDING | UNDERPAID | SUCCESS | OVERPAID | EXPIRED
  payment_attempt integer NOT NULL DEFAULT 1,                  -- external_id = order-<id>, then order-<id>-2 ...
  cvu             text,
  alias           text,
  expires_at      timestamptz,
  fulfilled_at    timestamptz,                                 -- set exactly once
  created_at      timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS orders_unfulfilled ON orders (created_at) WHERE fulfilled_at IS NULL;

CREATE TABLE IF NOT EXISTS talo_events (
  event_id    bigint PRIMARY KEY,                              -- the webhook body's integer "id"
  received_at timestamptz NOT NULL DEFAULT now()
);
```

`shop/payments.py`, shared by the webhook and the reconciliation job:
```python
import logging

from psycopg import AsyncConnection

from .talo_hub import PAID, RANK

log = logging.getLogger("talo")

async def apply_payment(conn: AsyncConnection, payment: dict, event_id: int | None = None) -> int | None:
    """Webhook + reconciliation; `payment` was just re-read with GET. Returns the order id if fulfilled NOW."""
    async with conn.transaction():  # ONE transaction: event id + status + fulfilment
        if event_id is not None:
            cur = await conn.execute("INSERT INTO talo_events (event_id) VALUES (%s) ON CONFLICT DO NOTHING", (event_id,))
            if cur.rowcount == 0:
                return None  # already handled: commit, answer 2xx
        status = payment["status"]
        await conn.execute(  # forward only: a slow, older re-read can't move SUCCESS back to PENDING
            "UPDATE orders SET payment_status = %s WHERE payment_id = %s AND payment_status = ANY(%s)",
            (status, payment["id"], [s for s, rank in RANK.items() if rank < RANK.get(status, -1)]),
        )
        if status not in PAID:
            return None  # UNDERPAID (the buyer transfers the rest), PENDING, EXPIRED: never fulfil
        cur = await conn.execute(
            "UPDATE orders SET fulfilled_at = now() WHERE payment_id = %s AND fulfilled_at IS NULL RETURNING id",
            (payment["id"],),
        )
        return row[0] if (row := await cur.fetchone()) else None

async def send_receipt(order_id: int) -> None:  # slow side effects: email, stock, shipping
    log.info("order %s paid: email the buyer, reserve stock, ship", order_id)
```

Django, same logic with the ORM: `get_or_create` on the event's primary key absorbs the concurrent-duplicate IntegrityError, and `transaction.on_commit` enqueues the receipt only if the commit happened. `payments/models.py` mirrors `schema.sql`: `Order` with `amount = DecimalField(max_digits=12, decimal_places=2)` (never FloatField), `payment_id = CharField(max_length=150, unique=True, null=True)`, `payment_status` (`max_length=10`, default `""`), `payment_attempt` (default 1), `cvu`, `alias`, `expires_at`, `fulfilled_at` (null), `created_at` (`auto_now_add`); and `TaloEvent` with `event_id = BigIntegerField(primary_key=True)`: that constraint IS the dedupe. Add `"django_tasks_db"` (on 5.2 also `"django_tasks"`) and the app to `INSTALLED_APPS`, set `TASKS = {"default": {"BACKEND": "django_tasks_db.DatabaseBackend"}}`, `makemigrations` + `migrate`. `payments/services.py`:
```python
import logging
from functools import partial

from django.db import transaction
from django.utils import timezone

try:
    from django.tasks import task  # Django 6.x
except ImportError:  # Django 5.2 LTS: pip install django-tasks
    from django_tasks import task

from .models import Order, TaloEvent
from .talo_hub import PAID, RANK, TaloHub

hub = TaloHub()  # one per process

@task
def send_receipt(order_id: int) -> None:  # slow side effects run in the worker: email, stock, shipping
    logging.getLogger("talo").info("order %s paid: email the buyer, reserve stock, ship", order_id)

def apply_payment(payment: dict, event_id: int | None = None) -> int | None:
    """Webhook + reconciliation; `payment` was just re-read with GET. Returns the order id if fulfilled NOW."""
    with transaction.atomic():  # ONE transaction: event id + status + fulfilment
        if event_id is not None:
            _, created = TaloEvent.objects.get_or_create(event_id=event_id)  # handles the IntegrityError race
            if not created:
                return None  # already handled: commit, answer 2xx
        status, orders = payment["status"], Order.objects.filter(payment_id=payment["id"])
        orders.filter(payment_status__in=[s for s, r in RANK.items() if r < RANK.get(status, -1)]).update(payment_status=status)
        if status not in PAID or not orders.filter(fulfilled_at__isnull=True).update(fulfilled_at=timezone.now()):
            return None  # UNDERPAID (never fulfil), or fulfilled before
        order_id = orders.values_list("pk", flat=True).get()
        transaction.on_commit(partial(send_receipt.enqueue, order_id))  # enqueued only if the COMMIT happens
        return order_id
```

Never keep handled event ids or "fulfilled" flags in a set, dict, `lru_cache` or module global: each gunicorn/uvicorn worker, restart and machine has its own copy, so a retried event is processed twice or a fulfilment is lost. SQLite: tests only.

## 6. Reconciliation

FastAPI, `shop/reconcile.py`:
```python
async def reconcile(hub: AsyncTaloHub, conn: psycopg.AsyncConnection) -> int:
    cur = await conn.execute(  # PENDING, UNDERPAID, or paid but not fulfilled; 15 days > the longest expiry (14)
        "SELECT payment_id FROM orders WHERE payment_id IS NOT NULL AND fulfilled_at IS NULL AND payment_status <> 'EXPIRED'"
        " AND created_at > now() - interval '15 days'"
    )
    fulfilled = 0
    for (payment_id,) in await cur.fetchall():
        try:
            payment = await hub.get_payment(payment_id)
        except (TaloHubError, httpx2.HTTPError):
            logging.exception("reconcile: could not read %s", payment_id)
            continue
        if (order_id := await apply_payment(conn, payment)) is not None:  # same logic as the webhook
            await send_receipt(order_id)
            fulfilled += 1
    return fulfilled

async def main() -> None:
    hub = AsyncTaloHub()
    async with await psycopg.AsyncConnection.connect(os.environ["DATABASE_URL"], autocommit=True) as conn:
        logging.info("reconcile: fulfilled %s orders", await reconcile(hub, conn))
    await hub.http.aclose()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop)  # psycopg async needs a selector loop (Windows)
```

Django, `payments/management/commands/talo_reconcile.py`: a `BaseCommand` whose `handle` takes the same orders (`Order.objects.filter(payment_id__isnull=False, fulfilled_at__isnull=True, created_at__gte=timezone.now() - timedelta(days=15)).exclude(payment_status="EXPIRED")`) and runs `services.apply_payment(services.hub.get_payment(pid))` for each inside `try/except (TaloHubError, httpx2.HTTPError)`, so one failing read doesn't stop the rest; receipts go through `on_commit` + `db_worker` as in the webhook. Run it as `python manage.py talo_reconcile`.

Every 10–15 min (`*/10 * * * *`, UTC), one run at a time:
- Railway: a second service from the repo with Settings → Cron Schedule; start command `python -m shop.reconcile` or `python manage.py talo_reconcile` (it must exit; runs ≥ 5 min apart).
- Render: a Cron Job with that command and the web service's environment group.
- Fly.io: a process group `cron = "supercronic /app/crontab"`, then `fly scale count cron=1`.
- VPS: crontab or a systemd timer with the app's environment.

## 7. Local development

```bash
uvicorn shop.main:app --env-file .env --port 8000  # Windows: add --loop asyncio:SelectorEventLoop (psycopg async)
python -m dotenv -f .env run -- python manage.py runserver 8000  # Django, needs "python-dotenv[cli]"; plus manage.py db_worker
npx talo-hub events listen --forward-to http://localhost:8000/webhooks/talo
```

Relayed events are unsigned and carry `talo-hub-forwarded: 1`; `accept_unsigned_forward` takes them only with `TALO_HUB_DEV_FORWARD=1` **and** a `tph_test_` key. Never set the flag when deployed (a forged event can at worst trigger a re-read).

## 8. Deploy

- Start: `uvicorn shop.main:app --host 0.0.0.0 --port $PORT --workers 2` (Fly: `PORT` must equal fly.toml's `internal_port`; set it in `[env]`). Django: `gunicorn shop.wsgi --bind 0.0.0.0:$PORT --workers 3`, `python manage.py migrate` as the pre-deploy/release command, `python manage.py db_worker` as its own worker service or process.
- Railway, Render and Fly.io give a public https URL. Register the final one (custom domain, no www / http→https / slash redirect) with no basic auth, IP allowlist, bot challenge or preview protection in front of `/webhooks/talo`.
- VPS: app on 127.0.0.1 behind nginx with a valid certificate; `location = /webhooks/talo { proxy_pass http://127.0.0.1:8000; }` without `auth_basic` or `return` (uvicorn trusts nginx's `X-Forwarded-*` from 127.0.0.1 by default).
- Per environment: dev/staging get the `tph_test_` key and their own endpoint + `whsec_`; only production gets the `tph_live_` key and its endpoint's secret (`npx talo-hub login --production`, then `keys create ... --production` and `webhooks add <url> --production`). Never share a secret between environments.

## 9. Test it

`tests/test_verify.py` (pytest):
```python
import hashlib, hmac, time

import pytest

from shop.talo_hub import verify_signature  # Django: from payments.talo_hub import verify_signature

SECRET, T = "whsec_test_secret", 1793000000
BODY = b'{"id":1,"type":"webhook.test","created_at":"2026-10-26T00:00:00.000Z","data":{"message":"Talo Partner Hub test event","environment":"sandbox"}}'
V1 = "ee8e039a9e5e9ce3eac17705f1133bfd58776d46f5163e01ad6a42d2cc540b9e"

def test_vector():
    assert verify_signature(BODY, f"t={T},v1={V1}", SECRET, now=T)  # now= pins the clock for the fixed vector
    assert verify_signature(BODY, f"t={T},v1={V1}", SECRET, now=T + 300)

@pytest.mark.parametrize("body,header,now", [
    (BODY + b" ", f"t={T},v1={V1}", T),  # tampered body
    (BODY, f"t={T},v1={V1}", T + 301), (BODY, f"t={T},v1={V1}", T - 301), (BODY, f"t={T},v1={V1}", None),  # stale t
    (BODY, f"t={T},v1={V1[:-2]}", T), (BODY, f"t={T},v1={V1.upper()}", T), (BODY, f"t={T},v1={'z' * 64}", T),  # bad v1
    (BODY, f"t={T},v1={V1},v1={V1}", T), (BODY, f"v1={V1}", T), (BODY, "", T), (BODY, None, T),  # bad/missing header
])
def test_rejects(body, header, now):
    assert not verify_signature(body, header, SECRET, now=now)

def test_current_time_and_wrong_secret():
    t = int(time.time())
    v1 = hmac.new(SECRET.encode(), f"{t}.".encode() + BODY, hashlib.sha256).hexdigest()
    assert verify_signature(BODY, f"t={t},v1={v1}", SECRET)
    assert not verify_signature(BODY, f"t={t},v1={v1}", "whsec_other")
```

Route tests: inject `AsyncTaloHub(transport=httpx2.MockTransport(handler))` (Django: patch `services.hub` with `TaloHub(transport=...)`), sign bodies with the current time, use a real database, and assert: valid → 200 and fulfilled once; same event again → 200, not fulfilled twice; bad signature → 400, nothing stored; UNDERPAID → not fulfilled; failed re-read → 503 and the event id not stored. Tests never call the real hub.

Sandbox (`tph_test_` key, deployed or tunnelled https URL registered):
1. POST `/orders/<id>/pay`, then `npx talo-hub sandbox pay <payment_id>` → SUCCESS in ~10–25 s, fulfilled once, the page shows it.
2. New order, `npx talo-hub sandbox pay <id> --amount <less>` → UNDERPAID, not fulfilled, the page asks for the rest; pay the rest → fulfilled. Another with `--amount <more>` → OVERPAID, fulfilled.
3. The same event twice: locally (dev flag on), POST one event's body from `npx talo-hub events list --payment <id> --json` twice with `talo-hub-forwarded: 1` → 2xx both times, one fulfilment; signed duplicates are covered by the route tests.
4. `npx talo-hub webhooks test <id>` → delivered; `npx talo-hub doctor` → ready.

## Checklist

- [ ] `TALO_HUB_API_KEY` (scope charge) and `TALO_HUB_WEBHOOK_SECRET` only in server env; `.env*` gitignored
- [ ] Payment created on the server from the order's `Decimal`; `external_id` = order id, `-2`, `-3`… after EXPIRED
- [ ] Webhook verifies the raw bytes (`await request.body()` / `request.body` / `request.get_data()`): 300 s, 64-hex v1, `compare_digest`, 400
- [ ] Webhook path outside CSRF, login and auth dependencies, no redirect; exact https URL registered
- [ ] Unsigned forwarded events only with `TALO_HUB_DEV_FORWARD=1` and a `tph_test_` key
- [ ] Payment re-read with GET before acting; 503 when that fails
- [ ] One transaction: unique event id, forward-only status, `fulfilled_at IS NULL` conditional update
- [ ] Slow work after the response (BackgroundTasks, `django.tasks` + `on_commit`, a queue)
- [ ] Reconciliation scheduled on the host, one instance
- [ ] Verifier test vector green; sandbox exact / less / more / duplicate / ping / doctor done
- [ ] `tph_live_` key and production secret only in production