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 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 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] = {}


@app.get("/health")
async def health():
    return {"ok": True, "service": "0pt-data-engine", "db": bool(os.getenv("DATABASE_URL")), "sources": len(_SOURCES)}


@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:
        v = None
    return v or {"vehicle": None}


@app.get("/screen")
async def screen(q: str):
    try:
        return await opensanctions_screen(q)
    except Exception:
        return {"matches": [], "source": "opensanctions"}


@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"}