#!/usr/bin/env python3
"""Wire-level verification of the seller-facing SOLD flow, in isolation.

Proves the harness produces a correct, authentic sold row BEFORE any operator time
is spent driving a real FIFA client. Brings up its own Core + utas-host on ephemeral
ports against throwaway databases, runs the real settlement through the
`staging_sell` binary, then reads every seller-facing surface under BOTH A/B
variants and exercises the bulk clear verb.

ISOLATION: production ports 8099, 8199, 18080, 8443, 42127, 42130, 42131, 4216,
8080, 8081 and 8094 are in a hard deny-list checked before every bind and every
request, and nothing under /home/alex/openfut-promotion/state/ is opened.

    python3 scripts/sold-wire-check.py [--keep]
"""
import argparse
import http.client
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import sys
import tempfile
import time

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
FORBIDDEN = {8099, 8199, 18080, 8443, 42127, 42130, 42131, 4216, 8080, 8081, 8094}
TS = "2026-01-01T00:00:00Z"
PERSONA = "33068179"
SELLER_CLUB = "club-seller-a"
BUYER_CLUB = "club-buyer-b"
ITEM = "core-disposable-x"
CARD = "def-disposable"
TRADE_ID = "900500150"
GROSS = 150

checks = []


def check(label, ok, detail=""):
    checks.append((label, bool(ok), detail))
    print(f"  [{'PASS' if ok else 'FAIL'}] {label}{(': ' + detail) if detail else ''}")
    return ok


def free_port():
    for _ in range(200):
        s = socket.socket()
        s.bind(("127.0.0.1", 0))
        p = s.getsockname()[1]
        s.close()
        if p not in FORBIDDEN and p > 1024:
            return p
    raise RuntimeError("no free port")


def req(port, method, path, body=None):
    assert port not in FORBIDDEN, f"refusing to contact production port {port}"
    c = http.client.HTTPConnection("127.0.0.1", port, timeout=20)
    headers = {"X-OpenFUT-Game": "fifa17"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    c.request(method, path, body=json.dumps(body) if body is not None else None,
              headers=headers)
    r = c.getresponse()
    raw = r.read()
    c.close()
    try:
        return r.status, json.loads(raw)
    except Exception:
        return r.status, raw.decode("utf-8", "replace")


def wait_http(port, path, timeout=45, proc=None, log=None):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if proc is not None and proc.poll() is not None:
            tail = ""
            if log and os.path.exists(log):
                tail = open(log).read()[-1500:]
            raise RuntimeError(f"process exited {proc.returncode}\n{tail}")
        try:
            st, _ = req(port, "GET", path)
            if st < 500:
                return
        except Exception:
            time.sleep(0.25)
    tail = open(log).read()[-1500:] if log and os.path.exists(log) else ""
    raise RuntimeError(f"{path} on {port} never became ready\n{tail}")


def seed(db):
    """Two identities by direct SQL: Seller A (the FIFA persona) and synthetic Buyer B."""
    con = sqlite3.connect(db)
    for prof, club, coins, game in (
        ("prof-seller-a", SELLER_CLUB, 1_000, "fifa17"),
        ("prof-buyer-b", BUYER_CLUB, 20_000, "buyer-game"),
    ):
        con.execute(
            "INSERT INTO profiles (id, username, game_id, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?)", (prof, prof, game, TS, TS))
        con.execute(
            "INSERT INTO clubs (id, profile_id, name, coins, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?)", (club, prof, club, coins, TS, TS))
    con.execute(
        "INSERT INTO owned_cards (id, club_id, card_id, is_loan, acquired_at) "
        "VALUES (?, ?, ?, 0, ?)", (ITEM, SELLER_CLUB, CARD, TS))
    con.commit()
    con.close()


def owner_of(db, item):
    con = sqlite3.connect(db)
    row = con.execute("SELECT club_id FROM owned_cards WHERE id = ?", (item,)).fetchone()
    n = con.execute("SELECT COUNT(*) FROM owned_cards WHERE id = ?", (item,)).fetchone()[0]
    coins = dict(con.execute("SELECT id, coins FROM clubs").fetchall())
    con.close()
    return (row[0] if row else None), n, coins


def start_core(tmp, port, log):
    db = os.path.join(tmp, "core.db")
    pack = os.path.join(tmp, "pack.json")
    with open(pack, "w") as f:
        # A top-level ARRAY: Core's content-pack loader expects a sequence, not a
        # map. Needed because the preflight refuses to start when an owned card
        # references a CardDefinitionId no pack defines.
        json.dump([{
            "id": CARD, "name": "Disposable", "overall": 75, "position": "ST",
            "nation": "Nation", "league": "League", "club": "Club",
            "pace": 75, "shooting": 75, "passing": 75, "dribbling": 75,
            "defending": 40, "physical": 70, "rarity": "gold",
            "image_path": None,
        }], f)
    env = dict(os.environ,
               LISTEN_ADDR=f"127.0.0.1:{port}",
               DATABASE_URL=f"sqlite://{db}",
               # Core's real data dir (read-only): it needs chemistry_styles.json
               # and friends. The throwaway DB and the content pack stay in tmp.
               DATA_DIR=os.path.join(REPO, "openfut-core", "data"),
               OPENFUT_CONTENT_PACKS=pack)
    # Migrate-only pass first: Core owns its schema, so the fixture cannot be
    # seeded into an empty file. Stop it before the external writer touches the db.
    with open(log, "w") as lf:
        p = subprocess.Popen([os.path.join(REPO, "target/release/openfut-core")],
                             cwd=tmp, env=env, stdout=lf, stderr=subprocess.STDOUT)
    wait_http(port, "/health", proc=p, log=log)
    p.terminate()
    p.wait(timeout=20)
    seed(db)
    with open(log, "a") as lf:
        p = subprocess.Popen([os.path.join(REPO, "target/release/openfut-core")],
                             cwd=tmp, env=env, stdout=lf, stderr=subprocess.STDOUT)
    wait_http(port, "/health", proc=p, log=log)
    return p, db


def start_host(tmp, port, core_port, log, variant, coins_processed="0",
               count_mode="active"):
    with open(os.path.join(tmp, "catalog.json"), "w") as f:
        # Minimal STAGING catalog. Deliberately NOT the production catalog, which
        # lives under /home/alex/openfut-promotion/state/ and must never be opened.
        json.dump({"schema_version": 1, "game": "fifa17",
                   "cards": {CARD: {"asset_id": 212188, "version": 0,
                                    "rareflag": 1, "kind": "player"}}}, f)
    env = dict(os.environ,
               OPENFUT_UTAS_HOST_ADDR=f"127.0.0.1:{port}",
               OPENFUT_CORE_URL=f"http://127.0.0.1:{core_port}",
               # Deliberately dead: any Python fallback must fail closed and be
               # visible, never silently serve production data.
               OPENFUT_UTAS_PYTHON_URL="http://127.0.0.1:9",
               OPENFUT_FIFA17_TABLES_DIR=os.path.join(REPO, "fifa17-recon/data/tables"),
               OPENFUT_IDENTITY_STORE=os.path.join(tmp, "identity.json"),
               OPENFUT_PERSONA_ID=PERSONA,
               OPENFUT_MARKET_DB=os.path.join(tmp, "market.db"),
               OPENFUT_PILE_DB=os.path.join(tmp, "pile.db"),
               OPENFUT_FIFA17_SOLD_EXPERIMENT=variant,
               OPENFUT_FIFA17_SOLD_COINS_PROCESSED=coins_processed,
               OPENFUT_FIFA17_SOLD_COUNT_MODE=count_mode,
               OPENFUT_FIFA17_CATALOG=os.path.join(tmp, "catalog.json"),
               RUST_LOG="info")
    with open(log, "w") as lf:
        p = subprocess.Popen([os.path.join(REPO, "target/release/openfut-utas-host")],
                             cwd=tmp, env=env, stdout=lf, stderr=subprocess.STDOUT)
    wait_http(port, "/ut/game/fifa17/tradePile/counts", proc=p, log=log)
    return p


def banner(t):
    print("\n" + "=" * 72)
    print(f"== {t}")
    print("=" * 72)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--keep", action="store_true")
    args = ap.parse_args()

    tmp = tempfile.mkdtemp(prefix="openfut-sold-wire-")
    procs = []
    try:
        core_port, host_port = free_port(), free_port()
        core_log = os.path.join(tmp, "core.log")
        host_log = os.path.join(tmp, "host.log")
        banner("ISOLATED STAGING (production untouched)")
        print(f"  tmp        : {tmp}")
        print(f"  core       : 127.0.0.1:{core_port}")
        print(f"  utas-host  : 127.0.0.1:{host_port}")
        print(f"  forbidden  : {sorted(FORBIDDEN)}")

        core, core_db = start_core(tmp, core_port, core_log)
        procs.append(core)
        host = start_host(tmp, host_port, core_port, host_log, "highest")
        procs.append(host)
        print("  both ready")

        bann = [l for l in open(host_log) if "sold-experiment" in l]
        check("host banner names the variant", any("bidState=highest" in l for l in bann),
              (bann[0].strip() if bann else "no banner"))

        banner("BEFORE — seller A owns the item, nothing listed")
        own, n, coins = owner_of(core_db, ITEM)
        print(f"  owner={own} instances={n} coins={coins}")
        st, pile = req(host_port, "GET", "/ut/game/fifa17/tradePile")
        st2, counts = req(host_port, "GET", "/ut/game/fifa17/tradePile/counts")
        print(f"  /tradePile total={pile.get('total')}  counts={json.dumps(counts)}")
        check("seller owns the item", own == SELLER_CLUB, str(own))
        check("no rows before listing", pile.get("total") == 0)
        check("sold counter starts at 0", counts.get("sold") == 0)

        banner(f"LIST — authentic active listing at {GROSS} coins")
        # Seed the listing directly into the staging market db: the client normally
        # does this via POST /auctionhouse, which needs a wire-id mapping we do not
        # have in this headless check. The LISTING SHAPE is identical either way.
        mdb = os.path.join(tmp, "market.db")
        con = sqlite3.connect(mdb)
        con.execute(
            "INSERT INTO listings (listing_id, card_id, core_item_id, wire_item_id, "
            "wire_resource_id, start_price, buy_now_price, owner, state, created_at, "
            "item_json, duration_secs) VALUES (?,?,?,?,?,?,?,?, 'active', ?, ?, ?)",
            (TRADE_ID, CARD, ITEM, 100000178, 212188, GROSS, GROSS, "CAGE",
             str(int(time.time() * 1000)), json.dumps({
                 "id": 100000178, "resourceId": 212188, "rating": 75,
                 "preferredPosition": "ST", "itemState": "forSale",
                 "untradeable": False, "assetId": 212188}), 3600))
        con.commit()
        con.close()
        st, pile = req(host_port, "GET", "/ut/game/fifa17/tradePile")
        st2, counts = req(host_port, "GET", "/ut/game/fifa17/tradePile/counts")
        row = pile["auctionInfo"][0]
        print(f"  active row: tradeState={row['tradeState']} bidState={row['bidState']} "
              f"expires={row['expires']} counts={json.dumps(counts)}")
        check("active row is active/none", row["tradeState"] == "active" and row["bidState"] == "none")
        check("counts.selling == 1 while active", counts.get("selling") == 1)
        check("counts.sold still 0 while active", counts.get("sold") == 0)

        banner("PURCHASE — synthetic Buyer B, through the REAL settlement path")
        out = subprocess.run(
            [os.path.join(REPO, "target/release/staging_sell"),
             "--market-db", mdb, "--core-url", f"http://127.0.0.1:{core_port}",
             "--trade-id", TRADE_ID, "--item", ITEM,
             "--seller", SELLER_CLUB, "--buyer", BUYER_CLUB, "--gross", str(GROSS)],
            capture_output=True, text=True, timeout=120)
        print("  " + "\n  ".join((out.stdout + out.stderr).strip().splitlines()))
        check("staging_sell succeeded", out.returncode == 0, f"exit {out.returncode}")

        own, n, coins = owner_of(core_db, ITEM)
        fee = GROSS * 5 // 100
        print(f"  owner={own} instances={n} coins={coins} fee={fee}")
        check("ownership transferred to buyer", own == BUYER_CLUB, str(own))
        check("exactly ONE authoritative instance", n == 1, str(n))
        check("buyer debited gross", coins.get(BUYER_CLUB) == 20_000 - GROSS,
              str(coins.get(BUYER_CLUB)))
        check("seller credited net", coins.get(SELLER_CLUB) == 1_000 + GROSS - fee,
              str(coins.get(SELLER_CLUB)))
        check("economy shrank by exactly the fee",
              21_000 - sum(coins.values()) == fee, str(21_000 - sum(coins.values())))

        banner("SOLD ROW — variant A: closed / highest")
        st, pile = req(host_port, "GET", "/ut/game/fifa17/tradePile")
        st2, counts = req(host_port, "GET", "/ut/game/fifa17/tradePile/counts")
        st3, status = req(host_port, "GET", f"/ut/game/fifa17/trade/status?tradeIds={TRADE_ID}")
        a_row = pile["auctionInfo"][0] if pile.get("auctionInfo") else {}
        print("  " + json.dumps(a_row, indent=2).replace("\n", "\n  "))
        print(f"  counts={json.dumps(counts)}")
        check("sold row is present in the pile", pile.get("total") == 1)
        check("tradeState closed", a_row.get("tradeState") == "closed")
        check("bidState highest (variant A)", a_row.get("bidState") == "highest")
        check("currentBid == sale price", a_row.get("currentBid") == GROSS)
        check("expires 0", a_row.get("expires") == 0)
        check("twelve atoms exactly", len(a_row) == 12, str(len(a_row)))
        check("counts.sold == 1", counts.get("sold") == 1)
        check("counts.selling == 0", counts.get("selling") == 0)
        check("/trade/status agrees", status["auctionInfo"][0]["tradeState"] == "closed"
              and status["auctionInfo"][0]["bidState"] == "highest")

        banner("VARIANT B — same state, restart host with closed / buyNow")
        host.terminate(); host.wait(timeout=20); procs.remove(host)
        host = start_host(tmp, host_port, core_port, host_log, "buyNow",
                          coins_processed="1", count_mode="active_plus_sold")
        procs.append(host)
        st, pileb = req(host_port, "GET", "/ut/game/fifa17/tradePile")
        st2, countsb = req(host_port, "GET", "/ut/game/fifa17/tradePile/counts")
        b_row = pileb["auctionInfo"][0]
        print(f"  bidState={b_row['bidState']} coinsProcessed={b_row['coinsProcessed']} "
              f"counts={json.dumps(countsb)}")
        check("bidState buyNow (variant B)", b_row.get("bidState") == "buyNow")
        check("coinsProcessed 1 when asked", b_row.get("coinsProcessed") == 1)
        check("count_mode active_plus_sold counts the sold row",
              countsb.get("count") == 1 and countsb.get("sold") == 1,
              json.dumps(countsb))
        differing = sorted(k for k in a_row if a_row.get(k) != b_row.get(k))
        check("A/B differ ONLY in bidState and coinsProcessed",
              differing == ["bidState", "coinsProcessed"], str(differing))

        banner("CLEAR — the PE-proven bulk verb DELETE .../trade/sold")
        pre_coins = owner_of(core_db, ITEM)[2]
        st, body = req(host_port, "DELETE", "/ut/delete/game/fifa17/trade/sold")
        print(f"  HTTP {st} body={json.dumps(body)}")
        st2, pilec = req(host_port, "GET", "/ut/game/fifa17/tradePile")
        st3, countsc = req(host_port, "GET", "/ut/game/fifa17/tradePile/counts")
        own2, n2, post_coins = owner_of(core_db, ITEM)
        print(f"  after clear: total={pilec.get('total')} counts={json.dumps(countsc)} "
              f"owner={own2} instances={n2}")
        check("clear acks 200 {}", st == 200 and body == {})
        check("sold row gone from the pile", pilec.get("total") == 0)
        check("counts.sold back to 0", countsc.get("sold") == 0)
        check("clear moved NO coins", pre_coins == post_coins, f"{pre_coins} -> {post_coins}")
        check("buyer still owns the item after clear", own2 == BUYER_CLUB, str(own2))
        check("still exactly one instance", n2 == 1, str(n2))
        st, again = req(host_port, "DELETE", "/ut/delete/game/fifa17/trade/sold")
        check("clearing again is a safe no-op", st == 200)
        cleared = [l for l in open(host_log) if "market-clear-sold" in l]
        check("clear is logged for capture", bool(cleared),
              cleared[-1].strip() if cleared else "no log line")

        banner("PRODUCTION UNTOUCHED")
        alive = subprocess.run(["ps", "-o", "pid=", "-p", "3631953"],
                               capture_output=True, text=True).stdout.strip()
        check("prod-host pid 3631953 still alive", alive == "3631953", alive or "gone")
        opened = subprocess.run(
            ["bash", "-lc",
             "ls -l /proc/*/fd 2>/dev/null | grep -c openfut-promotion || true"],
            capture_output=True, text=True).stdout.strip()
        print(f"  staging fds referencing production state: (informational) {opened}")

        banner("RESULT")
        passed = sum(1 for _, ok, _ in checks if ok)
        print(f"  {passed}/{len(checks)} checks passed")
        failed = [l for l, ok, _ in checks if not ok]
        if failed:
            print("  FAILED: " + "; ".join(failed))
        print("\n  " + ("ALL CHECKS PASSED" if not failed else "FAILURES PRESENT"))
        return 0 if not failed else 1
    finally:
        for p in procs:
            try:
                p.terminate(); p.wait(timeout=15)
            except Exception:
                try:
                    p.kill()
                except Exception:
                    pass
        if args.keep:
            print(f"\n  kept {tmp}")
        else:
            shutil.rmtree(tmp, ignore_errors=True)
            print(f"\n  removed {tmp}")


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