"""Webhook checks: registration, delivery, retries. Modes: "quick" (default) and "retry" (about 14 minutes).

Configuration comes from environment variables (see README.md in this folder):
  TABBY_SIM_BASE   simulator base URL            (default http://localhost:4030/tabby-sim)
  TABBY_SIM_KEY    Authorization header value    (default "Bearer sk_test_000000000", the simulator's test key)
  TABBY_SIM_RECEIVER      receiver base URL (default http://localhost:9199)
  TABBY_SIM_RECEIVER_LOG  file the receiver's output is redirected to (default receiver.log)
  TABBY_SIM_DB            database name, read through the mysql client (credentials from ~/.my.cnf)
"""
import os
import sys

DB = os.environ.get("TABBY_SIM_DB") or sys.exit("set TABBY_SIM_DB: the database name, e.g. TABBY_SIM_DB=mydb")
import json, subprocess, sys, time, urllib.request, urllib.error, urllib.parse

BASE = os.environ.get("TABBY_SIM_BASE", "http://localhost:4030/tabby-sim")
KEY = os.environ.get("TABBY_SIM_KEY", "Bearer sk_test_000000000")
RECEIVER = os.environ.get("TABBY_SIM_RECEIVER", "http://localhost:9199")
RECEIVER_LOG = os.environ.get("TABBY_SIM_RECEIVER_LOG", "receiver.log")
RUN = str(int(time.time()))[-6:]


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *a, **k):
        return None


opener = urllib.request.build_opener(NoRedirect)


def call(method, path, body=None, form=None, auth=True, merchant=None, raw_body=None):
    data, headers = None, {}
    if body is not None:
        data, headers = json.dumps(body).encode(), {"content-type": "application/json"}
    if raw_body is not None:
        data, headers = raw_body.encode(), {"content-type": "application/json"}
    if form is not None:
        data = urllib.parse.urlencode(form).encode()
    if auth:
        headers["Authorization"] = KEY
    if merchant is not None:
        headers["X-Merchant-Code"] = merchant
    req = urllib.request.Request(BASE + path, data=data, method=method, headers=headers)
    try:
        r = opener.open(req, timeout=30)
    except urllib.error.HTTPError as e:
        r = e
    raw = r.read().decode()
    try:
        parsed = json.loads(raw)
    except ValueError:
        parsed = raw
    return r.status if hasattr(r, "status") else r.code, parsed


def sql(query):
    out = subprocess.run(["mysql", DB, "-N", "-B", "-e", query],
                         capture_output=True, text=True)
    if out.returncode != 0:
        raise RuntimeError(out.stderr)
    return [line.split("\t") for line in out.stdout.splitlines() if line]


def deliveries(payment_id, webhook_id=None):
    where = f"payment_id='{payment_id}'" + (f" AND webhook_id={webhook_id}" if webhook_id else "")
    rows = sql(f"SELECT event, attempt, state, IFNULL(http_status,'-'), IFNULL(error,''), CAST(request_body AS CHAR), webhook_id, attempted_at FROM tabby_sim_webhook_deliveries WHERE {where} ORDER BY id")
    return [{"event": r[0], "attempt": int(r[1]), "state": r[2], "http": r[3], "error": r[4], "body": json.loads(r[5]) if r[5] and r[5] != "NULL" else None, "webhook_id": int(r[6]), "at": r[7]} for r in rows]


def wait_for(payment_id, count, timeout=15, webhook_id=None):
    end = time.time() + timeout
    rows = deliveries(payment_id, webhook_id)
    while len(rows) < count and time.time() < end:
        time.sleep(0.5)
        rows = deliveries(payment_id, webhook_id)
    return rows


def settle(payment_id, webhook_id=None, pause=4):
    time.sleep(pause)
    return deliveries(payment_id, webhook_id)


results = []


def check(name, ok, detail=""):
    results.append(bool(ok))
    print(("PASS  " if ok else "FAIL  ") + name + (f"   [{detail}]" if (detail and not ok) else ""), flush=True)


def new_payment(merchant, approve=True, phone="500000001", email="otp.success@tabby.ai"):
    body = {"merchant_code": merchant, "payment": {"amount": "100.00", "currency": "AED",
            "buyer": {"phone": phone, "email": email},
            "order": {"reference_id": "T-step5", "items": [{"title": "Phone", "quantity": 1, "unit_price": "100.00"}]}},
            "merchant_urls": {"success": "https://example.com/ok", "cancel": "https://example.com/cancel", "failure": "https://example.com/fail"}}
    _, d = call("POST", "/api/v2/checkout", body=body)
    sid, pid = d["id"], d["payment"]["id"]
    if approve:
        call("POST", f"/hpp/{sid}", form={"action": "approve", "otp": "8888"}, auth=False)
    return sid, pid


def register(merchant, url, title="X-Sign", value="secret-5"):
    body = {"url": url}
    if title:
        body["header"] = {"title": title, "value": value}
    return call("POST", "/api/v1/webhooks", body=body, merchant=merchant)


def P(pid, suffix=""):
    return f"/api/v2/payments/{pid}{suffix}"


def capture(pid, amount, ref):
    return call("POST", P(pid, "/captures"), body={"amount": amount, "reference_id": ref})


def refund(pid, amount, ref):
    return call("POST", P(pid, "/refunds"), body={"amount": amount, "reference_id": ref})


def quick():
    # ---------------------------------------------------------------- A: registration
    mA = f"wh5a{RUN}"
    st, d = register(mA, f"{RECEIVER}/ok")
    wA = int(d["id"]) if st == 200 else None
    check("register -> 200 {id (string), url, is_test true, header}", st == 200 and sorted(d) == ["header", "id", "is_test", "url"] and isinstance(d["id"], str) and d["is_test"] is True and d["header"] == {"title": "X-Sign", "value": "secret-5"} and d["url"] == f"{RECEIVER}/ok", str(d))
    st, d = call("GET", "/api/v1/webhooks", merchant=mA)
    check("GET lists it", st == 200 and isinstance(d, list) and len(d) == 1 and d[0]["id"] == str(wA), str(d))
    st, d = call("GET", "/api/v1/webhooks", merchant=f"nobody{RUN}")
    check("GET for a merchant code with none -> 200 null", st == 200 and d is None, f"{st} {d!r}")
    st, d = register(f"wh5nohdr{RUN}", f"{RECEIVER}/ok", title=None)
    check("register without header -> 200, header null", st == 200 and d["header"] is None, str(d))
    for label, kwargs in (("no key", dict(auth=False)),):
        st, d = call("POST", "/api/v1/webhooks", body={"url": f"{RECEIVER}/ok"}, merchant=mA, **kwargs)
        check("POST without key -> 401 'authorization'", st == 401 and d == {"status": "error", "errorType": "not_authorized", "error": "authorization"}, str(d))
    st, d = call("GET", "/api/v1/webhooks", merchant=mA, auth=False)
    check("GET without key -> 401 'authorization'", st == 401 and d.get("error") == "authorization")
    bad = {"status": "error", "errorType": "bad_data", "error": "bad_request"}
    st, d = call("POST", "/api/v1/webhooks", body={"url": f"{RECEIVER}/ok"})
    check("POST without X-Merchant-Code -> 400 bad_request", (st, d) == (400, bad), str(d))
    st, d = call("GET", "/api/v1/webhooks")
    check("GET without X-Merchant-Code -> 400 bad_request", (st, d) == (400, bad), str(d))
    for label, body in (("missing url", {}), ("url not a url", {"url": "not-a-url"}), ("ftp url", {"url": "ftp://x.y/z"}),
                        ("header not an object", {"url": f"{RECEIVER}/ok", "header": "x"}),
                        ("header title with a space", {"url": f"{RECEIVER}/ok", "header": {"title": "Bad Name", "value": "v"}})):
        st, d = call("POST", "/api/v1/webhooks", body=body, merchant=mA)
        check(f"register {label} -> 400 bad_request", (st, d) == (400, bad), f"{st} {d}")
    st, d = call("POST", "/api/v1/webhooks", raw_body="{not json", merchant=mA)
    check("register with broken JSON -> 400", st == 400, f"{st} {d}")
    st, d = call("GET", "/api/v1/webhooks", merchant=mA)
    check("none of the rejected registrations was stored", len(d) == 1, str(d))

    # ---------------------------------------------------------------- B: the full flow on one webhook
    sid, pid = new_payment(mA)
    rows = wait_for(pid, 1)
    check("approve -> 1 delivery 'authorize', success, HTTP 200, attempt 1", len(rows) == 1 and rows[0]["event"] == "authorize" and rows[0]["state"] == "success" and rows[0]["http"] == "200" and rows[0]["attempt"] == 1 and rows[0]["webhook_id"] == wA, str(rows))
    body = rows[0]["body"] if rows else {}
    check("...payload keys are exactly the documented ones", sorted(body) == sorted("id created_at expires_at closed_at status is_test is_expired amount currency order captures refunds meta token".split()), str(sorted(body)))
    check("...status 'authorized', id = payment, order only reference_id, empty captures/refunds, closed_at null, amount '100'",
          body.get("status") == "authorized" and body.get("id") == pid and body.get("order") == {"reference_id": "T-step5"} and body.get("captures") == [] and body.get("refunds") == [] and body.get("closed_at") is None and body.get("amount") == "100" and body.get("is_expired") is False and body.get("is_test") is True and body.get("token") is None)
    n0 = len(rows)

    capture(pid, "60", "cap-1")
    rows = wait_for(pid, n0 + 1)
    last = rows[-1] if len(rows) > n0 else {}
    check("partial capture 60 -> delivery 'capture', status 'authorized', 1 capture in the body", len(rows) == n0 + 1 and last.get("event") == "capture" and last["body"]["status"] == "authorized" and len(last["body"]["captures"]) == 1 and last["body"]["captures"][0]["amount"] == "60", str(last))
    n1 = len(rows)

    capture(pid, "60", "cap-1")
    capture(pid, "50", "cap-2")           # 400: over the remaining 40
    call("GET", P(pid))
    check("replayed capture, failed (400) capture and GET -> NO new delivery", len(settle(pid)) == n1)

    # Tabby lists Update as a webhook event (status as before): a PUT that changes the reference sends one
    call("PUT", P(pid), body={"order": {"reference_id": "NEW-REF"}})
    rows = wait_for(pid, n1 + 1)
    last = rows[-1] if len(rows) > n1 else {}
    check("PUT changing the reference -> delivery 'update', status 'authorized', the new reference", len(rows) == n1 + 1 and last.get("event") == "update" and last["body"]["status"] == "authorized" and last["body"]["order"] == {"reference_id": "NEW-REF"}, str(last))
    n1 = len(rows)

    capture(pid, "40", "cap-3")
    rows = wait_for(pid, n1 + 2)
    cap_row = rows[-2] if len(rows) >= n1 + 2 else {}
    last = rows[-1] if len(rows) >= n1 + 2 else {}
    # Tabby's documented sequence for a full capture: 'authorized' (with the capture), then 'closed'
    check("capture of the rest -> TWO deliveries: 'capture' then 'close'", len(rows) == n1 + 2 and cap_row.get("event") == "capture" and last.get("event") == "close", str([r["event"] for r in rows]))
    check("...the 'capture' one: status 'authorized', 2 captures, closed_at null", cap_row.get("body", {}).get("status") == "authorized" and len(cap_row["body"]["captures"]) == 2 and cap_row["body"]["closed_at"] is None, str(cap_row))
    check("...the 'close' one: status 'closed', 2 captures, closed_at set", last.get("body", {}).get("status") == "closed" and len(last["body"]["captures"]) == 2 and last["body"]["closed_at"], str(last))
    check("...the PUT reference_id shows in both webhook orders", cap_row.get("body", {}).get("order") == {"reference_id": "NEW-REF"} and last.get("body", {}).get("order") == {"reference_id": "NEW-REF"}, str((cap_row.get("body", {}).get("order"), last.get("body", {}).get("order"))))
    n2 = len(rows)

    capture(pid, "40", "cap-3")           # retry of the closing capture: replay
    check("retry of the closing capture -> NO new delivery", len(settle(pid)) == n2)

    refund(pid, "30", "ref-1")
    rows = wait_for(pid, n2 + 1)
    last = rows[-1] if len(rows) > n2 else {}
    check("refund 30 -> 'refund', status 'closed', 1 refund of 30 in the body", len(rows) == n2 + 1 and last.get("event") == "refund" and last["body"]["status"] == "closed" and len(last["body"]["refunds"]) == 1 and last["body"]["refunds"][0]["amount"] == "30", str(last))
    n3 = len(rows)
    refund(pid, "30", "ref-1")            # replay
    refund(pid, "500", "ref-2")           # 400
    check("replayed refund and failed (400) refund -> NO new delivery", len(settle(pid)) == n3)
    check("the payment got exactly 6 deliveries: authorize, capture, update, capture, close, refund", [r["event"] for r in deliveries(pid)] == ["authorize", "capture", "update", "capture", "close", "refund"], str([r["event"] for r in deliveries(pid)]))
    log = open(RECEIVER_LOG).read()
    check("the receiver saw the custom header X-Sign: secret-5", f"/ok | X-Sign: secret-5 | " in log and pid in log)

    # ---------------------------------------------------------------- C: other events
    sid, pid = new_payment(mA, approve=False)
    call("POST", f"/hpp/{sid}", form={"action": "reject"}, auth=False)
    rows = wait_for(pid, 1)
    check("HPP Reject -> 1 delivery 'reject', body status 'rejected'", len(rows) == 1 and rows[0]["event"] == "reject" and rows[0]["body"]["status"] == "rejected" and rows[0]["state"] == "success", str(rows))

    sid, pid = new_payment(mA, approve=False, email="otp.rejected@tabby.ai")
    call("POST", f"/hpp/{sid}", form={"action": "approve", "otp": "8888"}, auth=False)
    rows = wait_for(pid, 1)
    check("otp.rejected@tabby.ai + Approve -> 'reject' webhook", len(rows) == 1 and rows[0]["event"] == "reject" and rows[0]["body"]["status"] == "rejected", str(rows))

    sid, pid = new_payment(mA)
    wait_for(pid, 1)
    call("POST", P(pid, "/close"))
    rows = wait_for(pid, 2)
    check("close of an AUTHORIZED payment -> 'authorize' then 'close' (status 'closed', closed_at set)", [r["event"] for r in rows] == ["authorize", "close"] and rows[-1]["body"]["status"] == "closed" and rows[-1]["body"]["closed_at"], str([r["event"] for r in rows]))
    call("POST", P(pid, "/close"))
    check("closing again (400) -> NO new delivery", len(settle(pid)) == 2)

    sid, pid = new_payment(mA, approve=False)
    call("POST", f"/hpp/{sid}", form={"action": "cancel"}, auth=False)
    rows = settle(pid)
    state = call("GET", P(pid))[1]["status"]
    print(f"INFO  Cancel -> payment {state}; {len(rows)} delivery row(s) (TABBY_SIM_SEND_EXPIRED_WEBHOOK is off by default => expect 0)")
    check("HPP Cancel -> EXPIRED, and no webhook while TABBY_SIM_SEND_EXPIRED_WEBHOOK is off", state == "EXPIRED" and len(rows) == 0, str(rows))

    sid, pid = new_payment(mA, approve=False, phone="500000002")
    check("checkout rejected at creation (payment stays CREATED) -> no webhook", len(settle(pid)) == 0)

    sid, pid = new_payment(f"other{RUN}")
    check("a payment of another merchant code gets nothing from this webhook", len(settle(pid)) == 0)

    sid, pid = new_payment(mA, approve=False)
    check("a CREATED payment (never approved) -> no webhook", len(settle(pid)) == 0)

    # ---------------------------------------------------------------- D: a registered webhook only sees later events
    mD = f"wh5d{RUN}"
    sid, pid = new_payment(mD)            # approved before any webhook exists
    time.sleep(2)
    register(mD, f"{RECEIVER}/ok")
    capture(pid, "10", "c")
    rows = wait_for(pid, 1)
    check("webhook registered after the approve: gets the later capture only (no 'authorize')", [r["event"] for r in rows] == ["capture"], str([r["event"] for r in rows]))

    # ---------------------------------------------------------------- E: two webhooks, one payment
    mE = f"wh5e{RUN}"
    _, d1 = register(mE, f"{RECEIVER}/ok"); _, d2 = register(mE, f"{RECEIVER}/ok", title=None)
    sid, pid = new_payment(mE)
    rows = wait_for(pid, 2)
    check("two webhooks -> each event is delivered once to each", sorted(r["webhook_id"] for r in rows) == sorted([int(d1["id"]), int(d2["id"])]) and all(r["event"] == "authorize" for r in rows), str(rows))

    # ---------------------------------------------------------------- F: a 201 is not a success
    mF = f"wh5f{RUN}"
    st, d = register(mF, f"{RECEIVER}/created")
    sid, pid = new_payment(mF)
    rows = wait_for(pid, 1)
    check("receiver answers 201 -> delivery 'failed', http 201, error text, attempt 1", len(rows) == 1 and rows[0]["state"] == "failed" and rows[0]["http"] == "201" and rows[0]["attempt"] == 1 and "expected 200" in rows[0]["error"], str(rows))

    # ---------------------------------------------------------------- G: connection refused
    mG = f"wh5g{RUN}"
    register(mG, "http://localhost:9/nothing-listens-here")
    sid, pid = new_payment(mG)
    rows = wait_for(pid, 1)
    check("nothing listening -> delivery 'failed', no http status, an error text", len(rows) == 1 and rows[0]["state"] == "failed" and rows[0]["http"] == "-" and rows[0]["error"], str(rows))

    # ---------------------------------------------------------------- H: the limit of 4
    mH = f"wh5h{RUN}"
    codes = [register(mH, f"{RECEIVER}/ok?n={i}")[0] for i in range(1, 5)]
    check("4 registrations -> all 200", codes == [200, 200, 200, 200], str(codes))
    st, d = register(mH, f"{RECEIVER}/ok?n=5")
    check("5th registration -> 400 bad_request", (st, d) == (400, bad), f"{st} {d}")
    st, d = call("GET", "/api/v1/webhooks", merchant=mH)
    check("...and the list still has exactly 4", len(d) == 4, str(len(d)))
    st, d = register(f"{mH}-other", f"{RECEIVER}/ok")
    check("another merchant code still has its own 4 slots", st == 200, str(d))

    # ---------------------------------------------------------------- I: concurrent registrations never exceed 4
    import threading
    mI = f"wh5i{RUN}"
    out = []
    ts = [threading.Thread(target=lambda i=i: out.append(register(mI, f"{RECEIVER}/ok?c={i}")[0])) for i in range(8)]
    [t.start() for t in ts]; [t.join() for t in ts]
    st, d = call("GET", "/api/v1/webhooks", merchant=mI)
    check("8 parallel registrations -> exactly 4 succeed, 4 are refused, 4 stored", sorted(out) == [200] * 4 + [400] * 4 and len(d) == 4, f"{sorted(out)} stored={len(d)}")


def retry():
    from datetime import datetime
    mR = f"wh5r{RUN}"
    st, d = register(mR, f"{RECEIVER}/fail")
    wR = int(d["id"])
    t0 = time.time()
    sid, pid = new_payment(mR)
    print(f"retry test: payment {pid}, webhook {wR}, receiver answers 500 on /fail; watching up to ~15 minutes", flush=True)
    seen = set()
    last_new = t0
    while time.time() - t0 < 16 * 60:
        for r in deliveries(pid, wR):
            if r["attempt"] not in seen:
                seen.add(r["attempt"]); last_new = time.time()
                print(f"  attempt {r['attempt']} logged at {r['at']}  state={r['state']} http={r['http']}", flush=True)
        if len(seen) >= 5 and time.time() - last_new > 150:
            break
        time.sleep(5)
    rows = deliveries(pid, wR)
    ts = [datetime.strptime(r["at"], "%Y-%m-%d %H:%M:%S") for r in rows]
    gaps = [int((b - a).total_seconds()) for a, b in zip(ts, ts[1:])]
    print("\nDELIVERY LOG (tabby_sim_webhook_deliveries)")
    print("  attempt | attempted_at        | state  | http | gap since previous")
    for i, r in enumerate(rows):
        print(f"  {r['attempt']:>7} | {r['at']} | {r['state']:<6} | {r['http']:>4} | {'-' if i == 0 else str(gaps[i-1]) + ' s'}")
    check("5 attempts in total, numbered 1..5", [r["attempt"] for r in rows] == [1, 2, 3, 4, 5], str([r["attempt"] for r in rows]))
    check("every attempt 'failed' with HTTP 500", bool(rows) and all(r["state"] == "failed" and r["http"] == "500" for r in rows))
    check("no 6th attempt (waited >2 minutes after the 5th)", len(rows) == 5)
    check("every gap is between about 1 and 4 minutes (55-250 s)", len(gaps) == 4 and all(55 <= g <= 250 for g in gaps), f"gaps={gaps}")
    check("gaps are non-decreasing (+-5 s)", len(gaps) == 4 and all(b >= a - 5 for a, b in zip(gaps, gaps[1:])), f"gaps={gaps}")
    job = sql("SELECT id, queue, state, attempt, max_attempts, JSON_LENGTH(errors) FROM oban_jobs WHERE worker='DaProductApp.TabbySim.WebhookWorker' AND JSON_UNQUOTE(JSON_EXTRACT(args,'$.payment_id'))='" + pid + "'")
    print("\nOBAN JOB (id, queue, state, attempt, max_attempts, errors):", job)
    raw = sql("SELECT CAST(errors AS CHAR) FROM oban_jobs WHERE worker='DaProductApp.TabbySim.WebhookWorker' AND JSON_UNQUOTE(JSON_EXTRACT(args,'$.payment_id'))='" + pid + "'")[0][0]
    n_unknown, n_http500 = raw.count("unknown worker"), raw.count("HTTP 500")
    print(f"   job errors: {n_http500} x 'HTTP 500', {n_unknown} x 'unknown worker'")
    check("the job ran in queue tabby_sim_webhooks and ended 'discarded' at attempt 5 of 5", bool(job) and job[0][1] == "tabby_sim_webhooks" and job[0][2] == "discarded" and job[0][3] == "5" and job[0][4] == "5", str(job))
    check("no 'unknown worker' error, and exactly 5 recorded 'HTTP 500' errors", n_unknown == 0 and n_http500 == 5, f"unknown={n_unknown} http500={n_http500}")
    log = open(RECEIVER_LOG).read()
    check("the receiver saw exactly 5 POSTs for this payment", log.count(pid) == 5, f"count={log.count(pid)}")


if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "quick"
    quick() if mode == "quick" else retry()
    print(f"\n{sum(results)}/{len(results)} passed")