#!/usr/bin/env python3
"""Crypto v2 client + transactional receiver. Python 3.10+, standard library."""

import hashlib, hmac, json, os, re, sqlite3, sys, tempfile, uuid
from pathlib import Path
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qsl

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "python"))
from client import request, make_sign

ROOT = Path(__file__).resolve().parents[1]
FIELDS = "callback_version event_id event_seq status mch_id pt_order sh_order money pay_money goods_desc time ptype network is_testnet asset address decimals expected_amount_raw amount_raw tx_hash event_index payment_detected confirmed expires_at reason sign".split()
IDENTITY = "mch_id pt_order sh_order money pay_money ptype network is_testnet asset address decimals expected_amount_raw".split()


def config():
    c = json.loads((ROOT / "config.json").read_text())
    c.update(os.environ)
    return c


def db(c):
    directory = Path(c.get("PAY_STATE_DIR", ROOT / ".state/python"))
    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
    d = sqlite3.connect(directory / "orders.sqlite", timeout=5, isolation_level=None)
    d.executescript((ROOT / "schema.sql").read_text())
    return d


def units(value, decimals):
    if (
        not re.fullmatch(r"(0|[1-9][0-9]*)(\.[0-9]+)?", value)
        or not 0 <= int(decimals) <= 36
    ):
        raise ValueError("Invalid amount")
    whole, _, fraction = value.partition(".")
    if len(fraction) > int(decimals):
        raise ValueError("Too many decimals")
    return str(int(whole + fraction.ljust(int(decimals), "0")))


def save_response(d, number, r, query=False):
    local = d.execute(
        "SELECT money,ptype,expected FROM orders WHERE sh_order=?", (number,)
    ).fetchone()
    if not local:
        raise ValueError("Unknown local order; never register from callback")
    data = r["data"]
    if query:
        if data["out_order_sn"] != number or len(data["orders"]) != 1:
            raise ValueError("Ambiguous query")
        data = data["orders"][0]
    c = data["crypto"]
    if str(c.get("callback_version")) != "2":
        raise ValueError("Order did not enable v2 callbacks")
    ptype = str(c["ptype"] if query else data["ptype"])
    if ptype != local[1] or units(str(data["money"]), 2) != units(local[0], 2):
        raise ValueError("Response identity mismatch")
    if units(local[0], c["decimals"]) != str(c["expected_units"]):
        raise ValueError("Response quantity mismatch")
    e = {
        k: str(c[k]) for k in ["network", "is_testnet", "asset", "address", "decimals"]
    }
    e.update(
        mch_id=str(c["merchant_id"]),
        pt_order=data["order_sn"],
        sh_order=number,
        money=str(data["money"]),
        pay_money=str(data.get("pay_money", data["money"])),
        ptype=ptype,
        expected_amount_raw=str(c["expected_units"]),
    )
    if local[2] and json.loads(local[2]) != e:
        raise ValueError("Saved order snapshot cannot change")
    d.execute("UPDATE orders SET expected=? WHERE sh_order=?", (json.dumps(e), number))


def apply(d, p, key):
    if set(p) != set(FIELDS) or any(not isinstance(v, str) for v in p.values()):
        raise ValueError("Invalid fields")
    if not re.fullmatch("[a-f0-9]{32}", p["sign"]) or not hmac.compare_digest(
        make_sign(p, key), p["sign"]
    ):
        raise ValueError("Invalid signature")
    if (
        p["callback_version"] != "2"
        or not re.fullmatch("cn_[a-f0-9]{32}", p["event_id"])
        or not re.fullmatch("[1-9][0-9]{0,9}", p["event_seq"])
        or int(p["event_seq"]) > 2147483647
    ):
        raise ValueError("Invalid event")
    status = p["status"]
    seq = int(p["event_seq"])
    if (
        status not in ["expired", "confirming", "confirmed"]
        or p["payment_detected"] not in ["0", "1"]
        or p["confirmed"] != ("1" if status == "confirmed" else "0")
        or (status != "expired" and p["payment_detected"] != "1")
    ):
        raise ValueError("Invalid phase")
    if p["payment_detected"] == "1":
        if (
            p["amount_raw"] != p["expected_amount_raw"]
            or not p["tx_hash"]
            or not p["event_index"]
        ):
            raise ValueError("Transfer mismatch")
    elif p["amount_raw"] or p["tx_hash"] or p["event_index"]:
        raise ValueError("Unexpected transfer")
    stream = "test" if p["reason"] == "test_callback" else "payment"
    digest = hashlib.sha256(
        "&".join(k + "=" + p[k] for k in sorted(p) if k != "sign").encode()
    ).hexdigest()
    d.execute("BEGIN IMMEDIATE")
    try:
        row = d.execute(
            "SELECT expected FROM orders WHERE sh_order=?", (p["sh_order"],)
        ).fetchone()
        if (
            not row
            or not row[0]
            or any(json.loads(row[0]).get(k) != p[k] for k in IDENTITY)
        ):
            raise ValueError("Order identity mismatch")
        old = d.execute(
            "SELECT digest FROM events WHERE event_id=?", (p["event_id"],)
        ).fetchone()
        if old:
            if old[0] != digest:
                raise ValueError("Conflicting replay")
            d.execute("COMMIT")
            return
        if d.execute(
            "SELECT 1 FROM events WHERE sh_order=? AND stream=? AND seq=?",
            (p["sh_order"], stream, seq),
        ).fetchone():
            raise ValueError("Conflicting sequence")
        d.execute(
            "INSERT OR IGNORE INTO states(sh_order,stream) VALUES (?,?)",
            (p["sh_order"], stream),
        )
        previous = d.execute(
            "SELECT seq,status FROM states WHERE sh_order=? AND stream=?",
            (p["sh_order"], stream),
        ).fetchone()
        if seq > previous[0] and previous[1] == "confirmed" and status != "confirmed":
            raise ValueError("State regression")
        d.execute(
            "INSERT INTO events VALUES (?,?,?,?,?)",
            (p["event_id"], p["sh_order"], stream, seq, digest),
        )
        if seq > previous[0]:
            d.execute(
                "UPDATE states SET seq=?,status=? WHERE sh_order=? AND stream=?",
                (seq, status, p["sh_order"], stream),
            )
            if status == "confirmed":
                # Business intent and event/state commit atomically. Test outbox never fulfils real orders.
                d.execute(
                    "INSERT OR IGNORE INTO outbox VALUES (?,?,?)",
                    (p["sh_order"], stream, p["event_id"]),
                )
        d.execute("COMMIT")
    except Exception:
        d.execute("ROLLBACK")
        raise


def inspect(d):
    return {
        t: [dict(zip([x[0] for x in q.description], r)) for r in q.fetchall()]
        for t in ["states", "events", "outbox"]
        for q in [d.execute("SELECT * FROM " + t)]
    }


def selftest():
    f = json.loads((ROOT / "fixtures.json").read_text())
    with tempfile.TemporaryDirectory() as temp:
        d = db({"PAY_STATE_DIR": temp})
        for e in f["orders"]:
            d.execute(
                "INSERT INTO orders VALUES (?,?,?,?)",
                (e["sh_order"], e["money"], e["ptype"], json.dumps(e)),
            )
        for case in f["cases"]:
            before = inspect(d)
            try:
                apply(d, case["payload"], f["key"])
                ok = True
            except ValueError:
                ok = False
            assert ok == case["accept"], case["name"]
            if not ok:
                assert before == inspect(d), case["name"] + " rollback"
        assert inspect(d) == f["final"], "Final persistence mismatch"
        for pt in ["2911", "2912", "2913"]:
            created = json.loads((ROOT / f"responses/{pt}-create.json").read_text())
            queried = json.loads((ROOT / f"responses/{pt}-query.json").read_text())
            number = queried["data"]["out_order_sn"]
            d.execute(
                "INSERT INTO orders(sh_order,money,ptype) VALUES (?,?,?)",
                (number, "1.00", pt),
            )
            save_response(d, number, created)
            save_response(d, number, queried, True)
        d.close()
    print(
        "PASS " + str(len(f["cases"])) + " callback cases, rollback, state and outbox"
    )


def serve(c, port):
    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            code = 400
            try:
                n = int(self.headers.get("Content-Length", "0"))
                if (
                    self.path != "/notify"
                    or not 0 < n <= 16384
                    or not self.headers.get("Content-Type", "")
                    .lower()
                    .startswith("application/x-www-form-urlencoded")
                ):
                    raise ValueError("Invalid HTTP request")
                pairs = parse_qsl(
                    self.rfile.read(n).decode("utf-8"),
                    keep_blank_values=True,
                    strict_parsing=True,
                )
                if len(pairs) != len(dict(pairs)):
                    raise ValueError("Duplicate fields")
                d = db(c)
                try:
                    apply(d, dict(pairs), c["PAY_KEY"])
                finally:
                    d.close()
                code = 200
            except ValueError:
                pass
            except Exception:
                code = 503
            self.send_response(code)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            self.wfile.write(b"success" if code == 200 else b"fail")

        def log_message(self, *_):
            pass

    print(
        "Local receiver /notify; TLS proxy required. Use inspect to see verified records.",
        flush=True,
    )
    ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()


def main():
    c = config()
    action = sys.argv[1] if len(sys.argv) > 1 else "help"
    if action == "selftest":
        selftest()
        return
    if action == "serve":
        serve(c, int(sys.argv[2]) if len(sys.argv) > 2 else 8081)
        return
    d = db(c)
    try:
        if action == "inspect":
            r = inspect(d)
        elif action == "balance":
            r = request("querybalance", {}, c)
        elif action in ["create", "demo"]:
            if "YOUR-MERCHANT-HOST" in c["PAY_NOTIFY_URL"]:
                raise ValueError("Set PAY_NOTIFY_URL to your HTTPS receiver first")
            number = sys.argv[2] if len(sys.argv) > 2 else "CV2" + uuid.uuid4().hex[:26]
            if c["PAY_PTYPE"] not in ["2911", "2912", "2913", "2901", "2902", "2903"]:
                raise ValueError("Invalid crypto channel")
            units(c["PAY_MONEY"], 2)
            d.execute(
                "INSERT INTO orders(sh_order,money,ptype) VALUES (?,?,?)",
                (number, c["PAY_MONEY"], c["PAY_PTYPE"]),
            )
            print("Saved merchant order: " + number, flush=True)
            r = request(
                "index",
                dict(
                    ptype=c["PAY_PTYPE"],
                    money=c["PAY_MONEY"],
                    order_sn=number,
                    goods_desc="",
                    client_ip="",
                    format="json",
                    notify_url=c["PAY_NOTIFY_URL"],
                    crypto_callback_version="2",
                ),
                c,
            )
            save_response(d, number, r)
            if action == "demo":
                r["query_result"] = request("query", {"out_order_sn": number}, c)
        elif action == "query":
            number = sys.argv[2]
            r = request("query", {"out_order_sn": number}, c)
            save_response(d, number, r, True)
        else:
            print(
                "selftest | balance | demo | create [ORDER] | query ORDER | serve [PORT] | inspect"
            )
            return
        print(json.dumps(r, ensure_ascii=False, indent=2))
    finally:
        d.close()


if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        print(
            "ERROR: " + str(e) + "; query saved order before retrying creation.",
            file=sys.stderr,
        )
        sys.exit(1)