0pt/services/api/orchestrator/main.py

"""
0pt data engine — FastAPI orchestrator (docs/04). Compliant sourcing only.
Runs on a VPS/worker host, NOT on Vercel (long jobs). Amounts in öre.
Reconciles the Next.js /api routes with the Python data engine: the Next layer
is the customer-facing edge; this service runs the actual pipeline.
"""
from __future__ import annotations

import asyncio
import hmac
import os
import uuid
from pathlib import Path
from typing import Any

import yaml
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel

from sources.connectors import opensanctions_screen, rdw_vehicle
from workers.pipeline import run_job

app = FastAPI(title="0pt data engine", version="0.1")
app.add_middleware(
    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"],
)

_SOURCES = yaml.safe_load((Path(__file__).resolve().parent.parent / "sources" / "sources.yaml").read_text(encoding="utf-8"))["sources"]


def _coerce_int(v: Any, default: int) -> int:
    """int(v) raises on decimal-formatted strings/floats; coerce via float() first
    and fall back to `default` on any unparseable, client-controlled input instead
    of a 500."""
    try:
        return int(float(v))
    except (TypeError, ValueError):
        return default

# ── estimate model (mirrors apps/web lib/data.computeEstimate) ───────────────
FIELD_RATE = 240
ENTITY_MULT = {"Bolag": 1.0, "Beslutsfattare": 1.4, "Fordon": 1.2}
FIELD_WEIGHT = {"identity": 1.0, "lei": 0.8, "financials": 1.6, "ubo": 1.4, "contact": 1.8, "intent": 1.5}
ENRICH_WEIGHT = {"dedup": 0.1, "crm": 0.2, "sanctions": 0.15, "nace": 0.1, "tech": 0.15}


class EstimateIn(BaseModel):
    markets: list[str] = []
    entity_type: str = "Bolag"
    fields: list[str] = []
    enrichment: list[str] = []
    volume: int = 1000
    frequency: str = "Engångs"
    confidence_target: int = 92


def compute_estimate(b: EstimateIn) -> dict[str, Any]:
    markets = max(1, len(b.markets))
    fields = set(b.fields) | {"identity"}
    fw = sum(FIELD_WEIGHT.get(f, 0.5) for f in fields) or 1
    em = ENTITY_MULT.get(b.entity_type, 1.0)
    cm = 1 + max(0, b.confidence_target - 90) * 0.03
    ew = sum(ENRICH_WEIGHT.get(e, 0) for e in b.enrichment)
    base = fw * markets * FIELD_RATE * (b.volume / 1000) * em * cm
    price = round(base * (1 + ew) / 5) * 5
    deposit = round(price / 2)
    days = min(14, max(1, round(1 + markets * 0.4 + b.volume / 6000 + len(fields) * 0.25)))
    avg_level = sum(3 if f in ("identity", "lei") else 2 if f in ("financials", "ubo") else 1 for f in fields) / len(fields)
    ceiling = round(82 + avg_level * 5)
    est_conf = min(b.confidence_target, ceiling)
    return {
        "total_ore": price * 100, "deposit_ore": deposit * 100, "balance_ore": (price - deposit) * 100,
        "est_days": days, "est_confidence": est_conf / 100, "recurring": b.frequency != "Engångs",
    }


# ── in-memory store (swapped for asyncpg when DATABASE_URL is set) ───────────
_ORDERS: dict[str, dict[str, Any]] = {}
_JOBS: dict[str, dict[str, Any]] = {}
# P014: one asyncio.Lock per order_id, held for the duration of a /run call.
# Process-local only (matches the honest in-memory-store limitation this file
# already documents), but closes the actual race a bare `if oid in _JOBS`
# check would leave open: two /run requests for the same order_id arriving
# back to back both await run_job() before either has stored a result, so a
# read-then-write check alone would let both through. See `run()` below.
_JOB_LOCKS: dict[str, asyncio.Lock] = {}


def _lock_for(oid: str) -> asyncio.Lock:
    lock = _JOB_LOCKS.get(oid)
    if lock is None:
        lock = asyncio.Lock()
        _JOB_LOCKS[oid] = lock
    return lock


class OrderIn(BaseModel):
    flow: str = "custom"
    config: dict[str, Any] = {}


# ── internal-token gate (fail-closed) ────────────────────────────────────────
# The edge sends `X-Internal-Token` on every engineFetch (lib/engine.ts), but
# until 2026-09-15 this service never VERIFIED it — a deployed engine accepted
# anonymous traffic from anyone who could reach the port. The gate below
# enforces it when INTERNAL_TOKEN is set: every route except /health (the
# liveness probe) must carry the exact token or gets 401. When INTERNAL_TOKEN
# is unset the gate is inert (local dev / docker compose without auth wiring);
# the production deployment ALWAYS sets it (see infra/docker-compose.yml).
_INTERNAL_TOKEN = os.getenv("INTERNAL_TOKEN", "").strip()
_PUBLIC_PATHS = {"/health", "/docs", "/openapi.json", "/redoc"}


class _InternalTokenGate:
    """Raw-ASGI middleware: constant-time X-Internal-Token check, 401 on
    missing/wrong token. Written as a plain ASGI class (not
    @app.middleware('http')) so behavior does not depend on the
    BaseHTTPMiddleware dispatch quirks of the installed starlette."""

    def __init__(self, app) -> None:
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http" and _INTERNAL_TOKEN and scope["path"] not in _PUBLIC_PATHS:
            headers = {}
            for k, v in scope.get("headers") or []:
                headers.setdefault(k.decode("latin-1").lower(), v.decode("latin-1"))
            supplied = headers.get("x-internal-token", "")
            if not hmac.compare_digest(supplied, _INTERNAL_TOKEN):
                resp = JSONResponse(
                    status_code=401,
                    content={"error": {"code": "unauthorized", "message": "missing or invalid X-Internal-Token"}},
                )
                await resp(scope, receive, send)
                return
        await self.app(scope, receive, send)


app.add_middleware(_InternalTokenGate)


@app.get("/health")
async def health():
    return {
        "ok": True,
        "service": "0pt-data-engine",
        "db": bool(os.getenv("DATABASE_URL")),
        "sources": len(_SOURCES),
        # Commit-kedjan: samma SHA som image-taggen/OCI-labeln (Dockerfile ARG
        # GIT_SHA). 'unknown' gör ett oprovenienserat image synligt istället
        # för tyst falskt — samma policy som 0pt-motor/scripts/verify-image.sh.
        "revision": os.getenv("GIT_SHA", "unknown"),
    }


@app.get("/sources")
async def sources():
    return [{k: s.get(k) for k in ("source_id", "name", "market", "access", "licence", "cadence")} for s in _SOURCES]


@app.get("/sources/{market}")
async def sources_for(market: str):
    return [s for s in _SOURCES if s.get("market") in (market, "global")]


@app.get("/vehicles/{plate}")
async def vehicle(plate: str):
    try:
        v = await rdw_vehicle(plate)
    except Exception as exc:
        raise HTTPException(status_code=503, detail="vehicle source unavailable") from exc
    return {**(v or {"vehicle": None}), "source_status": "ok"}


@app.get("/screen")
async def screen(q: str):
    try:
        result = await opensanctions_screen(q)
        return {**result, "source_status": "ok"}
    except Exception as exc:
        raise HTTPException(status_code=503, detail="screening source unavailable") from exc


@app.post("/estimates")
async def estimates(body: EstimateIn):
    return compute_estimate(body)


@app.post("/orders", status_code=201)
async def create_order(body: OrderIn):
    est = compute_estimate(EstimateIn(
        markets=body.config.get("markets", []),
        entity_type=body.config.get("type", "Bolag"),
        fields=body.config.get("fieldIds", []),
        enrichment=body.config.get("enrichmentIds", []),
        volume=_coerce_int(body.config.get("volume", 1000), 1000),
        frequency=body.config.get("frequency", "Engångs"),
        confidence_target=_coerce_int(body.config.get("confidence", 92), 92),
    ))
    oid = str(uuid.uuid4())
    _ORDERS[oid] = {"order_id": oid, "flow": body.flow, "status": "draft", "config": body.config, **est}
    return {"order_id": oid, "status": "draft", "deposit_ore": est["deposit_ore"], "total_ore": est["total_ore"]}


@app.get("/orders")
async def list_orders():
    return list(_ORDERS.values())


@app.get("/orders/{oid}")
async def get_order(oid: str):
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    return {**o, "job": _JOBS.get(oid)}


class StatusIn(BaseModel):
    status: str


@app.patch("/orders/{oid}")
async def set_status(oid: str, body: StatusIn):
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    o["status"] = body.status
    return o


@app.post("/orders/{oid}/run")
async def run(oid: str):
    """
    P014: idempotent delivery. An order should be fulfilled/delivered ONCE.
    Before this fix, calling this endpoint twice for the same order_id (a
    retried request, a duplicate trigger, an operator double-click) re-ran
    the full compliant sourcing pipeline a second time and silently
    overwrote `_JOBS[oid]` with a second, independently-produced result --
    a genuinely different delivery (a second live fetch against upstream
    registries, a second QA pass with its own confidence draws), completely
    indistinguishable from the first at the API surface. This is the actual
    "duplicate delivery" gap P014's recon flagged as NOT MET: nothing
    downstream of order creation was idempotent at all. A job already
    recorded for this order_id is now returned unchanged; `run_job` is never
    invoked twice for the same order in this process's lifetime. The lock
    (see `_lock_for` above) closes the race between two requests arriving
    concurrently, not just sequential retries.
    """
    o = _ORDERS.get(oid)
    if not o:
        raise HTTPException(404, "order saknas")
    async with _lock_for(oid):
        if oid in _JOBS:
            return _JOBS[oid]
        o["status"] = "in_production"
        _JOBS[oid] = await run_job(oid, o["config"])
        o["status"] = "review"
    return _JOBS[oid]


@app.get("/orders/{oid}/status")
async def job_status(oid: str):
    return _JOBS.get(oid) or {"stage": "queued", "pct": 0, "state": "queued"}