"""Expiry checks: two timers, one expired webhook, expire on read, pre-scoring-rejected, the job stopping. Modes: 1, 2, all.

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_LOG  file the receiver's output is redirected to (default receiver.log)
  TABBY_SIM_RECEIVER      receiver base URL (default http://localhost:9199)
  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
from datetime import datetime, timezone

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")
HOOK = RECEIVER + "/ok"
RUN = str(int(time.time()))[-6:]
WORKER = "DaProductApp.TabbySim.ExpiryWorker"
PENDING = "('available','scheduled','retryable')"


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=False):
    data, headers = None, {}
    if body is not None:
        data, headers = json.dumps(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
    text = r.read().decode()
    status = r.status if hasattr(r, "status") else r.code
    if raw:
        return status, text
    try:
        return status, json.loads(text)
    except ValueError:
        return status, text


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]


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 info(msg):
    print("      " + msg, flush=True)


def now():
    return datetime.now(timezone.utc)


def stamp():
    return now().strftime("%H:%M:%S")


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


def checkout(merchant, phone="500000001"):
    body = {"merchant_code": merchant, "payment": {"amount": "100.00", "currency": "AED", "buyer": {"phone": phone, "email": "otp.success@tabby.ai"},
            "order": {"reference_id": "T-step6", "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"}}
    st, d = call("POST", "/api/v2/checkout", body=body)
    assert st == 200, (st, d)
    return d["id"], d["payment"]["id"]


def db_status(pid):
    r = sql(f"SELECT status, is_expired, expires_at, updated_at, session_status FROM tabby_sim_payments WHERE id='{pid}'")
    return {"status": r[0][0], "is_expired": r[0][1], "expires_at": r[0][2], "updated_at": r[0][3], "session_status": r[0][4]}


def deliveries(pid):
    rows = sql(f"SELECT event, attempt, state, IFNULL(http_status,'-'), CAST(request_body AS CHAR) FROM tabby_sim_webhook_deliveries WHERE payment_id='{pid}' ORDER BY id")
    return [{"event": r[0], "attempt": int(r[1]), "state": r[2], "http": r[3], "body": json.loads(r[4])} for r in rows]


def jobs(min_id):
    return sql(f"SELECT id, state, scheduled_at, IFNULL(attempted_at,'-'), IFNULL(completed_at,'-') FROM oban_jobs WHERE worker='{WORKER}' AND id>{min_id} ORDER BY id")


def pending_jobs():
    return int(sql(f"SELECT COUNT(*) FROM oban_jobs WHERE worker='{WORKER}' AND state IN {PENDING}")[0][0])


def receiver_lines(pid):
    return [l for l in open(RECEIVER_LOG).read().splitlines() if pid in l]


def sleep_until(t, label=""):
    wait = (t - time.time())
    if wait > 0:
        info(f"waiting {wait:.0f}s {label}")
        time.sleep(wait)


def err400(resp, text):
    st, d = resp
    return st == 400 and isinstance(d, dict) and d.get("errorType") == "bad_data" and d.get("error") == text


def batch1():
    print(f"== batch 1 (E1, E3, E4, E5), started {stamp()} UTC", flush=True)
    max_job = int(sql("SELECT IFNULL(MAX(id),0) FROM oban_jobs")[0][0])
    old_created = int(sql("SELECT COUNT(*) FROM tabby_sim_payments WHERE status='CREATED'")[0][0])
    info(f"{old_created} CREATED payments already in the table (overdue ones are expired by the first run)")
    for m in ("exp1", "exp3"):
        st, d = register(m + RUN)
        check(f"register webhook for {m}{RUN}", st == 200, str(d))

    t0 = time.time()
    sidA, A = checkout("exp1" + RUN)                       # E1
    sidC1, C1 = checkout("exp3" + RUN)                     # E3 control (normal)
    sidC2, C2 = checkout("exp3" + RUN, phone="500000002")  # E3 pre-scoring rejected
    sidD, D = checkout("exp4" + RUN)                       # E4 (no webhook registered)
    F = [checkout("exp5" + RUN)[1] for _ in range(3)]      # E5, three quick checkouts
    info(f"created A={A} C1={C1} C2(rejected up front)={C2} D={D} F={F}")

    # ---- E5 part 1: never two pending jobs
    check("E5: after 7 quick checkouts there is exactly ONE pending expiry job", pending_jobs() == 1, f"pending={pending_jobs()}")

    # ---- E1: timers
    st, a = call("GET", f"/api/v2/payments/{A}")
    created_at, expires_at = a["created_at"], a["expires_at"]
    gap = (datetime.fromisoformat(expires_at.replace("Z", "+00:00")) - datetime.fromisoformat(created_at.replace("Z", "+00:00"))).total_seconds()
    check("E1: payment is CREATED right after checkout, expires_at = created_at + 2 min", a["status"] == "CREATED" and gap == 120, f"{a['status']} gap={gap}")
    st, page = call("GET", f"/hpp/{sidA}", auth=False, raw=True)
    check("E1: hosted page is usable at once (200, Approve button)", st == 200 and "Approve" in page and "Session expired" not in page, str(st))

    sleep_until(t0 + 72, "until the 1-minute session timer has passed")
    st, page = call("GET", f"/hpp/{sidA}", auth=False, raw=True)
    check("E1: after 1 min the hosted page says 'Session expired' (HTTP 410)", st == 410 and "Session expired" in page, f"{st}")
    st, _ = call("POST", f"/hpp/{sidA}", form={"action": "approve", "otp": "8888"}, auth=False, raw=True)
    check("E1: approve after the session timer -> 410", st == 410, str(st))
    st, a = call("GET", f"/api/v2/payments/{A}")
    check("E1: ...while retrieve still says CREATED (two separate timers)", a["status"] == "CREATED", a["status"])

    # ---- the job must do the work: nobody reads A, C1, C2, F* from the API until the DB shows EXPIRED
    exp_dt = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
    sleep_until(t0 + 110, "(no API reads of the payments, so only the background job can expire them)")
    seen = {}
    deadline = time.time() + 120
    while time.time() < deadline and not all(k in seen for k in ("A", "C1", "C2", "F0")):
        for key, pid in (("A", A), ("C1", C1), ("C2", C2), ("F0", F[0])):
            if key not in seen and db_status(pid)["status"] == "EXPIRED":
                seen[key] = now()
        time.sleep(2)
    for key, pid in (("A", A), ("C1", C1), ("C2", C2), ("F0", F[0])):
        late = (seen[key] - exp_dt).total_seconds() if key in seen else None
        info(f"{key}: EXPIRED in the DB {'%.0f' % late + 's after expires_at' if late is not None else 'NEVER'}")
    check("E1: the background job expired A on its own (nobody read it), within ~65 s of expires_at", "A" in seen and (seen["A"] - exp_dt).total_seconds() <= 65, str(seen))
    check("E5: ...and expired the pre-scoring-rejected, the control and the three exp5 payments too", all(k in seen for k in ("C1", "C2", "F0")), str(list(seen)))
    time.sleep(6)   # let the webhook go out

    ra = db_status(A)
    check("E1: stored as EXPIRED with is_expired = 1", ra["status"] == "EXPIRED" and ra["is_expired"] == "1", str(ra))
    # ---- webhooks, exactly once
    da = deliveries(A)
    check("E1: exactly ONE delivery-log row for A (event expire, attempt 1, success, HTTP 200)", len(da) == 1 and da[0]["event"] == "expire" and da[0]["attempt"] == 1 and da[0]["state"] == "success" and da[0]["http"] == "200", str(da))
    body = da[0]["body"] if da else {}
    check("E1: webhook body: status 'expired', is_expired true, expired_at = the payment's expires_at", body.get("status") == "expired" and body.get("is_expired") is True and body.get("expired_at") == expires_at and body.get("id") == A, str({k: body.get(k) for k in ("status", "is_expired", "expired_at", "id")}) + f" expires_at={expires_at}")
    check("E1: ...closed_at null, captures/refunds empty, order only reference_id", body.get("closed_at") is None and body.get("captures") == [] and body.get("refunds") == [] and body.get("order") == {"reference_id": "T-step6"})
    check("E1: the receiver got exactly one POST for A, with the custom header", len(receiver_lines(A)) == 1 and "X-Sign: secret-6" in (receiver_lines(A) or [""])[0], str(len(receiver_lines(A))))

    # ---- reading afterwards must not send it again
    for _ in range(3):
        call("GET", f"/api/v2/payments/{A}")
    call("GET", "/api/v2/payments?limit=20")
    call("PUT", f"/api/v2/payments/{A}", body={"order": {"reference_id": "x"}})
    time.sleep(5)
    check("E1: after more retrieves, a list and a PUT there is STILL exactly one webhook for A", len(deliveries(A)) == 1 and len(receiver_lines(A)) == 1, f"rows={len(deliveries(A))} lines={len(receiver_lines(A))}")
    st, a = call("GET", f"/api/v2/payments/{A}")
    check("E1: retrieve shows EXPIRED", a["status"] == "EXPIRED", a["status"])

    # ---- E3: pre-scoring-rejected
    r2, r1 = db_status(C2), db_status(C1)
    check("E3: pre-scoring-rejected payment (session rejected) expired: EXPIRED, is_expired 1", r2["session_status"] == "rejected" and r2["status"] == "EXPIRED" and r2["is_expired"] == "1", str(r2))
    check("E3: ...and sent NO webhook (0 delivery rows, 0 receiver lines)", len(deliveries(C2)) == 0 and len(receiver_lines(C2)) == 0, f"rows={len(deliveries(C2))}")
    check("E3: control payment of the same merchant code DID send one expired webhook", r1["status"] == "EXPIRED" and len(deliveries(C1)) == 1 and deliveries(C1)[0]["body"]["status"] == "expired", str(deliveries(C1)))

    # ---- E4: nothing works on an EXPIRED payment
    for label, pid in (("D (expired by the job)", D), ("A", A)):
        check(f"E4 {label}: capture -> 400 'could not capture not authorized payments'", err400(call("POST", f"/api/v2/payments/{pid}/captures", body={"amount": "100.00", "reference_id": "cap-1"}), "could not capture not authorized payments"))
        check(f"E4 {label}: close -> 400 'could not close not authorized payments'", err400(call("POST", f"/api/v2/payments/{pid}/close"), "could not close not authorized payments"))
        check(f"E4 {label}: refund -> 400 'can not refund payment that has no captures and not closed'", err400(call("POST", f"/api/v2/payments/{pid}/refunds", body={"amount": "10.00", "reference_id": "ref-1"}), "can not refund payment that has no captures and not closed"))
        check(f"E4 {label}: update -> 400 'could not update not authorized payments'", err400(call("PUT", f"/api/v2/payments/{pid}", body={"order": {"reference_id": "NEW"}}), "could not update not authorized payments"))
    st, a = call("GET", f"/api/v2/payments/{D}")
    check("E4: EXPIRED is terminal: still EXPIRED after those calls", a["status"] == "EXPIRED")

    # ---- E5: the job stopped
    time.sleep(70)
    left = int(sql("SELECT COUNT(*) FROM tabby_sim_payments WHERE status='CREATED' AND expires_at > UTC_TIMESTAMP()")[0][0])
    js = jobs(max_job)
    print("      expiry jobs created during batch 1 (id, state, scheduled_at, attempted_at, completed_at):", flush=True)
    for j in js:
        print("        ", j, flush=True)
    check("E5: no unexpired CREATED payment is left", left == 0, str(left))
    check("E5: the expiry job CHAIN HAS STOPPED: no available/scheduled/retryable job remains", pending_jobs() == 0, f"pending={pending_jobs()}")
    check("E5: every job of the chain ended 'completed' (none failed or discarded)", bool(js) and all(j[1] == "completed" for j in js), str([j[1] for j in js]))
    check("E5: the chain ran several times (not one run) before stopping", len(js) >= 2, str(len(js)))
    old_left = int(sql("SELECT COUNT(*) FROM tabby_sim_payments WHERE status='CREATED'")[0][0])
    check("E5: the 28 overdue CREATED rows from earlier days were expired too (none CREATED left)", old_left == 0, f"{old_left} CREATED left")
    return {"A": A}


def batch2():
    print(f"\n== batch 2 (E2, expire on read without the job), started {stamp()} UTC", flush=True)
    m = "exp2" + RUN
    check("register webhook for " + m, register(m)[0] == 200)
    t0 = time.time()
    sid1, B1 = checkout(m)
    sid2, B2 = checkout(m)
    pend = pending_jobs()
    info(f"after two checkouts: {pend} pending expiry job(s)")
    sql(f"DELETE FROM oban_jobs WHERE worker='{WORKER}' AND state IN {PENDING}")
    check("E2: the pending expiry job was removed (0 pending) - nothing can expire the payments in the background", pending_jobs() == 0)
    st, b1 = call("GET", f"/api/v2/payments/{B1}")          # a read before it is due must not expire it
    check("E2: retrieve before the time is up: CREATED", b1["status"] == "CREATED")
    expires_at = b1["expires_at"]
    exp_dt = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
    sleep_until(t0 + 135, "until after expires_at (no job, no reads)")
    check("E2: still no job running", pending_jobs() == 0)
    r = (db_status(B1), db_status(B2))
    check("E2: in the DB both payments are still CREATED (nothing has looked at them)", r[0]["status"] == "CREATED" and r[1]["status"] == "CREATED", str(r))
    check("E2: no webhook yet", len(deliveries(B1)) == 0 and len(deliveries(B2)) == 0)
    read_time = now()
    st, b1 = call("GET", f"/api/v2/payments/{B1}")
    check("E2: retrieve now returns EXPIRED (expired on read)", st == 200 and b1["status"] == "EXPIRED", b1["status"])
    r1 = db_status(B1)
    check("E2: ...and it is stored as EXPIRED, is_expired 1", r1["status"] == "EXPIRED" and r1["is_expired"] == "1", str(r1))
    time.sleep(6)
    d1 = deliveries(B1)
    check("E2: the expired webhook is sent now (late, because it was read late): one row, success", len(d1) == 1 and d1[0]["event"] == "expire" and d1[0]["state"] == "success", str(d1))
    check("E2: ...its expired_at is the stored expires_at, NOT the time of the read", d1 and d1[0]["body"]["expired_at"] == expires_at and (read_time - exp_dt).total_seconds() > 10, f"{d1[0]['body'].get('expired_at') if d1 else None} vs {expires_at}")
    # list expires the other one
    st, _ = call("GET", "/api/v2/payments?limit=1")
    check("E2: the LIST endpoint expires the other due payment too", db_status(B2)["status"] == "EXPIRED", str(db_status(B2)))
    time.sleep(6)
    check("E2: ...with its own single webhook", len(deliveries(B2)) == 1 and deliveries(B2)[0]["body"]["status"] == "expired", str(deliveries(B2)))
    call("GET", f"/api/v2/payments/{B1}"); call("GET", f"/api/v2/payments/{B2}"); call("GET", "/api/v2/payments?limit=20")
    time.sleep(4)
    check("E2: more reads add no second webhook", len(deliveries(B1)) == 1 and len(deliveries(B2)) == 1)
    st, a = call("POST", f"/api/v2/payments/{B1}/captures", body={"amount": "100.00", "reference_id": "c"})
    check("E2: capture on it -> 400", err400((st, a), "could not capture not authorized payments"))
    # a fresh checkout restarts the job
    sid3, B3 = checkout(m)
    check("E2: a new checkout restarts the job (1 pending)", pending_jobs() == 1, f"pending={pending_jobs()}")
    return {}


if __name__ == "__main__":
    which = sys.argv[1] if len(sys.argv) > 1 else "all"
    if which in ("1", "all"):
        batch1()
    if which in ("2", "all"):
        batch2()
    print(f"\n{sum(results)}/{len(results)} passed", flush=True)