OpenFUT/scripts/settlement-staging.py

#!/usr/bin/env python3
"""ISOLATED staging harness for the Core SOLD-settlement path.

Exercises the REAL `openfut-core` binary over REAL HTTP (`POST /economy/settle-sale`)
against a THROWAWAY SQLite database in a fresh temp directory, so the transfer-market
sold path can be validated with no FIFA client, no UTAS host, and no production state.

Isolation guarantees (all enforced below, not merely documented):
  * The database is created by `tempfile.mkdtemp()` and deleted on exit (`--keep` opts out).
  * The listen port is chosen by binding 127.0.0.1:0 and reading the port back; the
    PRODUCTION ports 8099 (utas-host), 8199 (oracle) and 18080 (prod Core) are in a
    hard deny-list and are never bound or contacted.
  * Nothing under /home/alex/openfut-promotion/state/ (the live DBs) is read or written.
  * Core runs with an explicit LISTEN_ADDR / DATABASE_URL / DATA_DIR and cwd set to the
    temp directory, so no relative path can escape into the repo or a live database.

Canonical two-party fixture reproduced here:
    seller 1,000 coins owning `item-x`; buyer 20,000; gross 15,000; fee 750.
    Expected: buyer 20,000 -> 5,000; seller 1,000 -> 15,250; owner seller -> buyer;
    exactly ONE row for `item-x`; modelled coins 21,000 -> 20,250 (delta == fee).

Usage:
    python3 scripts/settlement-staging.py [--keep] [--no-build]

Exit code 0 iff every assertion passes.
"""

from __future__ import annotations

import argparse
import http.client
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request

# --- fixed facts about the repo (read from openfut-core/src/{main,config}.rs) -------
# Config::from_env(): LISTEN_ADDR, DATABASE_URL, DATA_DIR, DB_MAX_CONNECTIONS.
# Plain `openfut-core` (no subcommand) runs its own migrations, then serves axum.
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CORE_BIN = os.path.join(REPO_ROOT, "target", "release", "openfut-core")
CORE_DATA_DIR = os.path.join(REPO_ROOT, "openfut-core", "data")

# Ports owned by the live production test. Never bind, never speak to.
FORBIDDEN_PORTS = frozenset({8099, 8199, 18080})

GAME = "fifa17"
TS = "2026-01-01T00:00:00Z"
SELLER_PROFILE, SELLER_CLUB = "prof-seller", "club-seller"
BUYER_PROFILE, BUYER_CLUB = "prof-buyer", "club-buyer"
ITEM_ID, CARD_ID = "item-x", "def-x"
SELLER_START, BUYER_START = 1_000, 20_000
GROSS, FEE = 15_000, 750
PROCEEDS = GROSS - FEE

# `app::build` runs a content preflight that rejects any owned card whose card_id is
# not a loaded CardDefinition, so the fixture's definition ships as a one-entry
# production content pack (shape: openfut-core/src/models/card.rs::CardDefinition).
CARD_PACK = [
    {
        "id": CARD_ID,
        "name": "Staging Fixture",
        "overall": 82,
        "position": "ST",
        "nation": "Testland",
        "league": "Staging League",
        "club": "Fixture FC",
        "pace": 80,
        "shooting": 80,
        "passing": 80,
        "dribbling": 80,
        "defending": 40,
        "physical": 75,
        "rarity": "gold",
        "image_path": None,
    }
]

READY_TIMEOUT_S = 30.0


# --- output helpers -----------------------------------------------------------------


def banner(title: str) -> None:
    print()
    print("=" * 72)
    print(f"== {title}")
    print("=" * 72)


class Checks:
    """Every expectation is printed where it happens AND tallied, so one failure never
    hides the rest and the RESULT block stays a summary."""

    def __init__(self) -> None:
        self.results: list[tuple[str, bool, str]] = []

    def _record(self, label: str, ok: bool, detail: str) -> bool:
        self.results.append((label, ok, detail))
        print(f"  [{'PASS' if ok else 'FAIL'}] {label}: {detail}")
        return ok

    def expect(self, label: str, actual, expected) -> bool:
        return self._record(
            label, actual == expected, f"expected {expected!r}, got {actual!r}"
        )

    def expect_true(self, label: str, ok: bool, detail: str) -> bool:
        return self._record(label, bool(ok), detail)

    def report(self) -> bool:
        failed = [label for label, ok, _ in self.results if not ok]
        print(f"  {len(self.results) - len(failed)}/{len(self.results)} checks passed")
        for label, ok, detail in self.results:
            if not ok:
                print(f"  [FAIL] {label}: {detail}")
        return not failed


# --- port / process plumbing --------------------------------------------------------


def pick_free_port() -> int:
    """Bind 127.0.0.1:0, read the port back, release it. Production ports refused."""
    for _ in range(64):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("127.0.0.1", 0))
            port = s.getsockname()[1]
        if port not in FORBIDDEN_PORTS and port > 1024:
            return port
    raise RuntimeError("could not obtain a free non-production port")


def build_core() -> None:
    print("building openfut-core (release)...")
    subprocess.run(
        ["cargo", "build", "-p", "openfut-core", "--release"],
        cwd=REPO_ROOT,
        check=True,
    )


def pack_path(workdir: str) -> str:
    return os.path.join(workdir, "content-pack.json")


def write_content_pack(workdir: str) -> str:
    path = pack_path(workdir)
    with open(path, "w") as fh:
        json.dump(CARD_PACK, fh, indent=2)
    return path


def launch_core(db_path: str, port: int, workdir: str, log_path: str, label: str):
    """Start the real Core binary against the throwaway DB. Core runs its migrations."""
    if port in FORBIDDEN_PORTS:
        raise RuntimeError(f"refusing to bind production port {port}")
    env = dict(os.environ)
    env.update(
        {
            "LISTEN_ADDR": f"127.0.0.1:{port}",
            "DATABASE_URL": f"sqlite://{db_path}",
            "DATA_DIR": CORE_DATA_DIR,
            "OPENFUT_CONTENT_PACKS": pack_path(workdir),
            "RUST_LOG": "openfut_core=info",
        }
    )
    log = open(log_path, "ab", buffering=0)
    log.write(f"\n---- {label} on 127.0.0.1:{port} ----\n".encode())
    proc = subprocess.Popen(
        [CORE_BIN],
        cwd=workdir,  # temp dir: no relative path can reach a real database
        env=env,
        stdout=log,
        stderr=log,
    )
    proc._log = log  # type: ignore[attr-defined]
    return proc


def wait_ready(proc, port: int, log_path: str) -> None:
    deadline = time.monotonic() + READY_TIMEOUT_S
    last = ""
    while time.monotonic() < deadline:
        if proc.poll() is not None:
            raise RuntimeError(
                f"Core exited early with code {proc.returncode}\n{tail(log_path)}"
            )
        try:
            conn = http.client.HTTPConnection("127.0.0.1", port, timeout=2)
            conn.request("GET", "/health")
            resp = conn.getresponse()
            resp.read()
            conn.close()
            if resp.status == 200:
                return
            last = f"/health -> HTTP {resp.status}"
        except OSError as exc:
            last = f"{type(exc).__name__}: {exc}"
        time.sleep(0.1)
    raise RuntimeError(
        f"Core on 127.0.0.1:{port} not ready after {READY_TIMEOUT_S:.0f}s "
        f"(last: {last})\n{tail(log_path)}"
    )


def stop_core(proc) -> None:
    if proc is None or proc.poll() is not None:
        return
    proc.terminate()
    try:
        proc.wait(timeout=10)
    except subprocess.TimeoutExpired:
        proc.kill()
        proc.wait(timeout=10)
    finally:
        log = getattr(proc, "_log", None)
        if log is not None:
            log.close()


def tail(log_path: str, lines: int = 25) -> str:
    try:
        with open(log_path, "r", errors="replace") as fh:
            body = fh.read().splitlines()
    except OSError:
        return "(no core log)"
    return "--- core log tail ---\n" + "\n".join(body[-lines:])


# --- database -----------------------------------------------------------------------


def connect(db_path: str) -> sqlite3.Connection:
    conn = sqlite3.connect(db_path, timeout=10)
    conn.execute("PRAGMA busy_timeout = 10000")
    return conn


def seed(db_path: str) -> None:
    """Direct-SQL fixture. Column sets match openfut-core/migrations/0001_initial.sql
    (+ 0016 game_dimension's profiles.game_id).

    Two profiles share one game_id, which services::profile::create_profile would
    refuse -- that limit is a service rule, not a schema constraint, and the settle
    route never resolves the active profile when both club ids are named explicitly.
    """
    conn = connect(db_path)
    try:
        with conn:
            conn.executemany(
                "INSERT INTO profiles (id, username, created_at, updated_at, game_id) "
                "VALUES (?, ?, ?, ?, ?)",
                [
                    (SELLER_PROFILE, "seller", TS, TS, GAME),
                    (BUYER_PROFILE, "buyer", TS, TS, GAME),
                ],
            )
            conn.executemany(
                "INSERT INTO clubs (id, profile_id, name, coins, created_at, updated_at) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                [
                    (SELLER_CLUB, SELLER_PROFILE, "Seller FC", SELLER_START, TS, TS),
                    (BUYER_CLUB, BUYER_PROFILE, "Buyer FC", BUYER_START, TS, TS),
                ],
            )
            conn.execute(
                "INSERT INTO owned_cards (id, club_id, card_id, is_loan, acquired_at) "
                "VALUES (?, ?, ?, 0, ?)",
                (ITEM_ID, SELLER_CLUB, CARD_ID, TS),
            )
    finally:
        conn.close()


def snapshot(db_path: str) -> dict:
    conn = connect(db_path)
    try:
        cur = conn.cursor()
        seller = cur.execute(
            "SELECT coins FROM clubs WHERE id = ?", (SELLER_CLUB,)
        ).fetchone()[0]
        buyer = cur.execute(
            "SELECT coins FROM clubs WHERE id = ?", (BUYER_CLUB,)
        ).fetchone()[0]
        owners = [
            row[0]
            for row in cur.execute(
                "SELECT club_id FROM owned_cards WHERE id = ?", (ITEM_ID,)
            )
        ]
        total = cur.execute("SELECT COALESCE(SUM(coins), 0) FROM clubs").fetchone()[0]
        return {
            "seller_coins": seller,
            "buyer_coins": buyer,
            "owner": owners[0] if owners else None,
            "item_rows": len(owners),
            "total_coins": total,
        }
    finally:
        conn.close()


def print_snapshot(title: str, snap: dict, extra: dict | None = None) -> None:
    banner(title)
    print(f"  seller club {SELLER_CLUB!r:>14} coins : {snap['seller_coins']:>8,}")
    print(f"  buyer  club {BUYER_CLUB!r:>14} coins : {snap['buyer_coins']:>8,}")
    print(f"  owner of {ITEM_ID!r:>17}       : {snap['owner']}")
    print(f"  rows in owned_cards for {ITEM_ID!r}  : {snap['item_rows']}")
    print(f"  total modelled coins (SUM clubs) : {snap['total_coins']:>8,}")
    for key, value in (extra or {}).items():
        print(f"  {key:<32} : {value}")


# --- HTTP ---------------------------------------------------------------------------


def post_settle(port: int, body: dict) -> tuple[int, str]:
    if port in FORBIDDEN_PORTS:
        raise RuntimeError(f"refusing to POST to production port {port}")
    req = urllib.request.Request(
        f"http://127.0.0.1:{port}/economy/settle-sale",
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json", "X-OpenFUT-Game": GAME},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            return resp.status, resp.read().decode(errors="replace")
    except urllib.error.HTTPError as exc:  # 4xx/5xx carry the rejection body
        return exc.code, exc.read().decode(errors="replace")


def show_response(status: int, body: str) -> dict | None:
    print(f"  HTTP {status}")
    try:
        parsed = json.loads(body)
    except json.JSONDecodeError:
        print(f"  body (not JSON): {body!r}")
        return None
    print("  body:")
    for line in json.dumps(parsed, indent=2, sort_keys=True).splitlines():
        print(f"    {line}")
    return parsed


# --- the run ------------------------------------------------------------------------


def run(port: int, db_path: str, workdir: str, log_path: str, checks: Checks) -> None:
    settle_body = {
        "item_id": ITEM_ID,
        "gross": GROSS,
        "fee": FEE,
        "seller_club_id": SELLER_CLUB,
        "buyer_club_id": BUYER_CLUB,
    }

    print(f"  content pack : {write_content_pack(workdir)} (defines {CARD_ID})")

    banner("MIGRATE (real Core creates the throwaway schema)")
    # Core owns its schema (sqlx migrations, embedded at compile time), so the
    # fixture cannot be seeded into an empty file. Start Core once purely to
    # migrate, stop it, THEN seed: the external writer and Core's WAL pool never
    # coexist, which is the safest ordering.
    migrator = launch_core(db_path, port, workdir, log_path, "migrate pass")
    try:
        wait_ready(migrator, port, log_path)
        print(f"  migrations applied; Core answered /health on 127.0.0.1:{port}")
    finally:
        stop_core(migrator)
    print("  migrate pass stopped")

    banner("SEED (canonical two-party fixture, direct SQL)")
    seed(db_path)
    print(f"  {SELLER_CLUB}: {SELLER_START:,} coins, owns {ITEM_ID} ({CARD_ID})")
    print(f"  {BUYER_CLUB}: {BUYER_START:,} coins")

    banner(f"LAUNCH Core on 127.0.0.1:{port} (db {db_path})")
    server = launch_core(db_path, port, workdir, log_path, "serve pass")
    try:
        wait_ready(server, port, log_path)
        print(f"  ready: GET /health -> 200 (pid {server.pid})")

        before = snapshot(db_path)
        print_snapshot("BEFORE", before)

        banner("PURCHASE  POST /economy/settle-sale")
        print(f"  request: {json.dumps(settle_body)}")
        status, body = post_settle(port, settle_body)
        receipt = show_response(status, body)

        after = snapshot(db_path)
        print_snapshot(
            "AFTER",
            after,
            {
                "fee withheld and destroyed": f"{FEE:,}",
                f"DUPLICATE COUNT for {ITEM_ID!r}": after["item_rows"],
                "coins destroyed (before-after)": f"{before['total_coins'] - after['total_coins']:,}",
            },
        )

        banner("EXPECTATIONS")
        checks.expect("purchase HTTP status", status, 200)
        checks.expect("buyer coins", after["buyer_coins"], BUYER_START - GROSS)
        checks.expect("seller coins", after["seller_coins"], SELLER_START + PROCEEDS)
        checks.expect("item owner is buyer club", after["owner"], BUYER_CLUB)
        checks.expect("duplicate count == 1", after["item_rows"], 1)
        checks.expect(
            "total coins before", before["total_coins"], SELLER_START + BUYER_START
        )
        checks.expect(
            "total coins after",
            after["total_coins"],
            SELLER_START + BUYER_START - FEE,
        )
        buyer_debit = before["buyer_coins"] - after["buyer_coins"]
        seller_credit = after["seller_coins"] - before["seller_coins"]
        checks.expect_true(
            "conservation buyer_debit == seller_credit + fee",
            buyer_debit == seller_credit + FEE,
            f"{buyer_debit:,} == {seller_credit:,} + {FEE:,}",
        )
        if receipt is None:
            checks.expect_true("receipt is JSON", False, "response body was not JSON")
        else:
            checks.expect("receipt.item_id", receipt.get("item_id"), ITEM_ID)
            checks.expect("receipt.card_id", receipt.get("card_id"), CARD_ID)
            checks.expect(
                "receipt.seller_club_id", receipt.get("seller_club_id"), SELLER_CLUB
            )
            checks.expect(
                "receipt.buyer_club_id", receipt.get("buyer_club_id"), BUYER_CLUB
            )
            checks.expect("receipt.gross", receipt.get("gross"), GROSS)
            checks.expect("receipt.fee", receipt.get("fee"), FEE)
            checks.expect("receipt.proceeds", receipt.get("proceeds"), PROCEEDS)
            checks.expect(
                "receipt.seller_balance",
                receipt.get("seller_balance"),
                after["seller_coins"],
            )
            checks.expect(
                "receipt.buyer_balance",
                receipt.get("buyer_balance"),
                after["buyer_coins"],
            )
            checks.expect(
                "receipt.squad_slots_freed", receipt.get("squad_slots_freed"), 0
            )

        banner("RETRY (identical request must be rejected, nothing may move)")
        retry_status, retry_body = post_settle(port, settle_body)
        show_response(retry_status, retry_body)
        replay = snapshot(db_path)
        print_snapshot("AFTER RETRY", replay)
        checks.expect_true(
            "retry rejected (4xx)",
            400 <= retry_status < 500,
            f"HTTP {retry_status}",
        )
        checks.expect("retry left buyer coins", replay["buyer_coins"], after["buyer_coins"])
        checks.expect(
            "retry left seller coins", replay["seller_coins"], after["seller_coins"]
        )
        checks.expect("retry left owner", replay["owner"], after["owner"])
        checks.expect("retry left total coins", replay["total_coins"], after["total_coins"])
        checks.expect("retry left one row", replay["item_rows"], 1)

        # The identical retry above is refused by the AFFORDABILITY guard, because
        # settle_sale debits before it touches ownership and the buyer no longer holds
        # 15,000. That alone never exercises the ownership CAS that is the actual
        # replay guard, so replay the same item at a price the buyer CAN afford: the
        # only thing left to stop it is "item not owned by the named seller".
        banner("REPLAY GUARD (affordable re-settle must still fail on ownership)")
        cheap = dict(settle_body, gross=100, fee=5)
        print(f"  request: {json.dumps(cheap)}")
        guard_status, guard_body = post_settle(port, cheap)
        show_response(guard_status, guard_body)
        guarded = snapshot(db_path)
        print_snapshot("AFTER REPLAY GUARD", guarded)
        checks.expect("replay guard rejects with 404", guard_status, 404)
        checks.expect_true(
            "replay guard cites ownership",
            "not owned" in guard_body,
            f"body: {guard_body}",
        )
        checks.expect(
            "replay guard left buyer coins", guarded["buyer_coins"], after["buyer_coins"]
        )
        checks.expect(
            "replay guard left seller coins",
            guarded["seller_coins"],
            after["seller_coins"],
        )
        checks.expect("replay guard left owner", guarded["owner"], after["owner"])
        checks.expect(
            "replay guard left total coins",
            guarded["total_coins"],
            after["total_coins"],
        )
        checks.expect("replay guard left one row", guarded["item_rows"], 1)
    finally:
        stop_core(server)
        print("\n  Core stopped")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument(
        "--keep", action="store_true", help="keep the temp dir (and its DB + core log)"
    )
    parser.add_argument(
        "--no-build", action="store_true", help="use target/release/openfut-core as-is"
    )
    args = parser.parse_args()

    if not args.no_build:
        build_core()
    if not os.path.isfile(CORE_BIN):
        print(f"core binary not found: {CORE_BIN}", file=sys.stderr)
        return 2

    port = pick_free_port()
    workdir = tempfile.mkdtemp(prefix="openfut-settlement-staging-")
    db_path = os.path.join(workdir, "staging.db")
    log_path = os.path.join(workdir, "core.log")

    banner("SETTLEMENT STAGING HARNESS (isolated; production untouched)")
    print(f"  temp dir     : {workdir}")
    print(f"  throwaway db : {db_path}")
    print(f"  core binary  : {CORE_BIN}")
    print(f"  port         : {port} (production {sorted(FORBIDDEN_PORTS)} never bound)")

    checks = Checks()
    try:
        run(port, db_path, workdir, log_path, checks)
    except Exception as exc:  # report, then still clean up
        banner("HARNESS ERROR")
        print(f"  {type(exc).__name__}: {exc}")
        checks.expect_true("harness completed", False, f"{type(exc).__name__}: {exc}")
    finally:
        if args.keep:
            print(f"\n  --keep: leaving {workdir} in place")
        else:
            shutil.rmtree(workdir, ignore_errors=True)
            print(f"\n  removed {workdir}")

    banner("RESULT")
    ok = checks.report()
    print()
    print(f"  {'ALL CHECKS PASSED' if ok else 'FAILURES PRESENT'}")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())