#!/usr/bin/env python3
"""Wibber backend integration tests. Needs: server on 127.0.0.1:8080 (multi-worker), MariaDB db wibber_test."""
import json, subprocess, threading, time, urllib.request, urllib.error, sys, concurrent.futures as cf
BASE = "http://127.0.0.1:8080/api/v1/"
def sql(q): return subprocess.run(["mysql", "wibber_test", "-N", "-B", "-e", q], capture_output=True, text=True).stdout.strip()
def call(method, path, body=None, tok=None):
    req = urllib.request.Request(BASE + path, method=method, data=json.dumps(body).encode() if body is not None else None)
    if tok: req.add_header("Authorization", "Bearer " + tok)
    try:
        with urllib.request.urlopen(req, timeout=30) as r: return r.status, json.load(r)
    except urllib.error.HTTPError as e: return e.code, json.load(e)
RESULTS = []
def check(name, cond, info=""):
    RESULTS.append((name, bool(cond))); print(("PASS " if cond else "FAIL ") + name + ("" if cond else "  <-- " + str(info))); sys.stdout.flush()
_n = [0]
def reg(coins=None):
    _n[0] += 1; sql("DELETE FROM rate_limits")
    s, d = call("POST", "auth/register.php", {"email": f"u{_n[0]}_{int(time.time()*1000)}@t.com", "password": "secret1", "username": f"User{_n[0]}"})
    assert s == 201, d
    u = {"id": d["data"]["user"]["id"], "tok": d["data"]["token"]}
    if coins is not None: sql(f"UPDATE coin_wallets SET balance={coins} WHERE user_id={u['id']}")
    return u
def reset():
    sql("DELETE FROM matchmaking_queue; DELETE FROM active_sessions; DELETE FROM video_rooms; DELETE FROM text_chat_sessions; DELETE FROM user_blocks; DELETE FROM user_reports; DELETE FROM rate_limits")
def bal(u): return int(sql(f"SELECT balance FROM coin_wallets WHERE user_id={u['id']}"))
def settle(users, mode, tries=25):
    """join everyone concurrently, then poll until all matched (or tries exhausted). returns {uid: room}"""
    qs = {}
    with cf.ThreadPoolExecutor(len(users)) as ex:
        for u, (s, d) in zip(users, ex.map(lambda u: call("POST", "matchmaking/join.php", {"mode": mode}, u["tok"]), users)): qs[u["id"]] = (u, d["data"] if s == 200 else d)
    rooms = {}
    for _ in range(tries):
        for uid, (u, d) in list(qs.items()):
            if uid in rooms: continue
            s, d = call("GET", f"matchmaking/status.php?queue_id={d['queue_id']}", None, u["tok"])
            qs[uid] = (u, d["data"])
            if d["data"]["state"] == "matched": rooms[uid] = d["data"]["room"]
        if len(rooms) == len(users): break
        time.sleep(0.15)
    return rooms

print("== Stage 3: users & coins")
a = reg()
s, d = call("GET", "users/profile.php", None, a["tok"]); check("profile returns 20 starting coins", s == 200 and d["data"]["user"]["coins"] == 20, d)
s, d = call("POST", "users/update-profile.php", {"username": "Neo", "bio": "hi"}, a["tok"]); check("update profile", s == 200 and d["data"]["user"]["username"] == "Neo" and d["data"]["profile"]["bio"] == "hi", d)
s, d = call("POST", "users/update-profile.php", {"username": "x"}, a["tok"]); check("bad username rejected", s == 422, d)
s, d = call("POST", "users/update-profile.php", {}, a["tok"]); check("empty update rejected", s == 422, d)
s, d = call("POST", "users/heartbeat.php", {}, a["tok"]); check("heartbeat returns counts", s == 200 and d["data"]["online_count"] >= 1 and "searching_video" in d["data"], d)
s, d = call("GET", "coins/history.php", None, a["tok"]); check("history shows signup bonus", s == 200 and d["data"]["transactions"][0]["type"] == "signup_bonus" and d["data"]["transactions"][0]["amount"] == 20, d)
s, d = call("GET", "app/config.php"); check("public app config", s == 200 and d["data"]["search_cost"] == 5, d)
s, d = call("GET", "coins/balance.php"); check("balance needs auth", s == 401)

print("== spin")
sql("DELETE FROM rate_limits"); b = reg(); start = bal(b)
with cf.ThreadPoolExecutor(6) as ex: rs = list(ex.map(lambda _: call("POST", "coins/spin.php", {}, b["tok"]), range(6)))
ok = [d for s, d in rs if s == 200]; cd = [d for s, d in rs if s == 429]
check("6 simultaneous spins with 25s cooldown: exactly 1 succeeds", len(ok) == 1 and len(cd) == 5, [s for s, _ in rs])
check("wallet = start + reward; 1 spin txn", bal(b) == start + ok[0]["data"]["reward"] and sql(f"SELECT COUNT(*) FROM coin_transactions WHERE user_id={b['id']} AND type='spin'") == "1")
check("segment_index matches reward", [5,10,30,20,5,10,30,20][ok[0]["data"]["segment_index"]] == ok[0]["data"]["reward"])
sql("UPDATE app_settings SET v='0' WHERE k='spin_cooldown_seconds'"); sql("DELETE FROM rate_limits"); base = bal(b)
with cf.ThreadPoolExecutor(8) as ex: rs = list(ex.map(lambda _: call("POST", "coins/spin.php", {}, b["tok"]), range(8)))
tot = sum(d["data"]["reward"] for s, d in rs if s == 200)
check("8 concurrent spins (cooldown off): no lost updates", all(s == 200 for s, _ in rs) and bal(b) == base + tot, (bal(b), base + tot))
check("ledger balance_after is consistent", sql(f"SELECT MAX(balance_after) FROM coin_transactions WHERE user_id={b['id']}") == str(bal(b)))
rewards = []
for i in range(30):
    if i % 8 == 0: sql("DELETE FROM rate_limits")
    s, d = call("POST", "coins/spin.php", {}, b["tok"]); rewards.append((s, d.get("error", {}).get("code")))
check("daily cap (20/24h) enforced", ("429", "spin_limit") in [(str(s), c) for s, c in rewards] and sql(f"SELECT COUNT(*) FROM spin_rewards WHERE user_id={b['id']}") == "20", sql(f"SELECT COUNT(*) FROM spin_rewards WHERE user_id={b['id']}"))
sql("UPDATE app_settings SET v='25' WHERE k='spin_cooldown_seconds'")

print("== Stage 4: matchmaking")
reset(); lone = reg(); s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, lone["tok"])
check("lone user waits, charged 5", s == 200 and d["data"]["state"] == "waiting" and bal(lone) == 15, d)
q = d["data"]["queue_id"]
s, d = call("POST", "matchmaking/cancel.php", {"queue_id": q}, lone["tok"]); check("cancel refunds", d["data"]["state"] == "cancelled" and bal(lone) == 20, d)
call("POST", "matchmaking/cancel.php", {"queue_id": q}, lone["tok"]); call("POST", "matchmaking/cancel.php", {"queue_id": q}, lone["tok"])
check("repeated cancel does not refund twice", bal(lone) == 20 and sql(f"SELECT COUNT(*) FROM coin_transactions WHERE user_id={lone['id']} AND type='search_refund'") == "1")
poor = reg(coins=3); s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, poor["tok"])
check("insufficient coins -> 402, no queue row, balance untouched", s == 402 and sql(f"SELECT COUNT(*) FROM matchmaking_queue WHERE user_id={poor['id']}") == "0" and bal(poor) == 3, d)
s, d = call("POST", "matchmaking/join.php", {"mode": "audio"}, lone["tok"]); check("bad mode rejected", s == 422, d)
reset(); dup = reg()
with cf.ThreadPoolExecutor(6) as ex: rs = list(ex.map(lambda _: call("POST", "matchmaking/join.php", {"mode": "video"}, dup["tok"]), range(6)))
check("6 simultaneous joins by one user: 1 queue row, charged once", sql(f"SELECT COUNT(*) FROM matchmaking_queue WHERE user_id={dup['id']}") == "1" and bal(dup) == 15 and len({d['data']['queue_id'] for s, d in rs if s == 200}) == 1, [s for s, _ in rs])
reset(); x, y, z = reg(), reg(), reg()
sql(f"INSERT INTO user_blocks VALUES ({x['id']},{y['id']},NOW())")
rooms = settle([x, y], "video", 6); check("blocked pair never matches", len(rooms) == 0, rooms)
rooms = settle([z], "video", 12); check("third user matches one of them, not both", len(rooms) == 1, rooms)
print("== 20 users join at the same instant")
reset(); us = [reg() for _ in range(20)]; rooms = settle(us, "video")
check("all 20 matched", len(rooms) == 20, len(rooms))
check("10 rooms, 20 active_sessions, no user in two rooms", sql("SELECT COUNT(*) FROM video_rooms") == "10" and sql("SELECT COUNT(*) FROM active_sessions") == "20" and sql("SELECT COUNT(DISTINCT user_id) FROM active_sessions") == "20")
sym = all(rooms[r["partner"]["id"]]["partner"]["id"] == uid for uid, r in rooms.items())
check("pairings symmetric; each pair has exactly 1 creator", sym and all(rooms[uid]["is_creator"] != rooms[rooms[uid]["partner"]["id"]]["is_creator"] for uid in rooms))
check("each user charged exactly 5", all(bal(u) == 15 for u in us))
check("peer ids are server-issued, unique, cross-consistent", len({r["my_peer_id"] for r in rooms.values()}) == 20 and all(rooms[rooms[uid]["partner"]["id"]]["my_peer_id"] == r["partner_peer_id"] for uid, r in rooms.items()))
print("== video room lifecycle")
reset(); p, q2 = reg(), reg(); rooms = settle([p, q2], "video"); rp, rq = rooms[p["id"]], rooms[q2["id"]]; rid = rp["room_id"]
s, d = call("POST", "matchmaking/join.php", {"mode": "text"}, p["tok"]); check("already in session -> 409", s == 409 and d["error"]["code"] == "already_in_session", d)
s, d = call("GET", f"video/room-status.php?room_id={rid}", None, q2["tok"]); check("partner_ready false initially", s == 200 and d["data"]["partner_ready"] is False and d["data"]["status"] == "waiting", d)
call("POST", "video/peer-ready.php", {"room_id": rid}, p["tok"])
s, d = call("GET", f"video/room-status.php?room_id={rid}", None, q2["tok"]); check("partner sees ready after peer-ready", d["data"]["partner_ready"] is True)
s, d = call("POST", "video/peer-ready.php", {"room_id": rid}, q2["tok"]); check("both ready -> active", d["data"]["status"] == "active", d)
out = reg(); s, d = call("GET", f"video/room-status.php?room_id={rid}", None, out["tok"]); check("outsider cannot read room (404)", s == 404, d)
s, d = call("POST", "video/end-room.php", {"room_id": rid}, out["tok"]); check("outsider cannot end room", s == 404, d)
s, d = call("POST", "video/end-room.php", {"room_id": rid}, p["tok"]); check("end-room ok", s == 200)
s, d = call("GET", f"video/room-status.php?room_id={rid}", None, q2["tok"]); check("partner gets clean ended state", d["data"]["status"] == "ended" and d["data"]["end_reason"] == "left", d)
check("active_sessions cleared for both", sql("SELECT COUNT(*) FROM active_sessions") == "0")
s, d = call("POST", "video/end-room.php", {"room_id": rid}, p["tok"]); check("end-room idempotent", s == 200)
s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, p["tok"]); check("can search again after room ended", s == 200 and d["data"]["state"] == "waiting", d)
print("== liveness / cleanup")
reset(); p, q2 = reg(), reg(); rooms = settle([p, q2], "video"); rid = rooms[p["id"]]["room_id"]
sql(f"UPDATE video_rooms SET user1_seen_at=NOW()-INTERVAL 3 MINUTE, user2_seen_at=NOW()-INTERVAL 3 MINUTE WHERE id={rid}")
sql(f"UPDATE video_rooms SET user2_seen_at=NOW() WHERE id={rid}")   # u2 alive, u1 silent
alive = p if sql(f"SELECT user2_id FROM video_rooms WHERE id={rid}") == str(p["id"]) else q2
s, d = call("GET", f"video/room-status.php?room_id={rid}", None, alive["tok"])
check("silent partner => room ends (partner_timeout or lazily)", d["data"]["status"] == "ended" and d["data"]["end_reason"] == "partner_timeout", d)
reset(); p, q2 = reg(), reg(); rooms = settle([p, q2], "video"); rid = rooms[p["id"]]["room_id"]
sql(f"UPDATE video_rooms SET user1_seen_at=NOW()-INTERVAL 5 MINUTE, user2_seen_at=NOW()-INTERVAL 5 MINUTE WHERE id={rid}")
o = subprocess.run(["php", "/home/claude/wibber-backend/app/cron/cleanup.php"], capture_output=True, text=True); j = json.loads(o.stdout or "{}")
check("cron ends abandoned room & frees both users", j.get("rooms_ended") == 1 and sql("SELECT COUNT(*) FROM active_sessions") == "0" and sql(f"SELECT status FROM video_rooms WHERE id={rid}") == "ended", (o.stdout, o.stderr))
reset(); e = reg(); s, d = call("POST", "matchmaking/join.php", {"mode": "text"}, e["tok"]); qq = d["data"]["queue_id"]
sql(f"UPDATE matchmaking_queue SET expires_at=NOW()-INTERVAL 1 MINUTE WHERE id={qq}")
s, d = call("GET", f"matchmaking/status.php?queue_id={qq}", None, e["tok"]); call("GET", f"matchmaking/status.php?queue_id={qq}", None, e["tok"])
check("expired search -> 'expired' and refunded once", d["data"]["state"] == "expired" and bal(e) == 20 and sql(f"SELECT COUNT(*) FROM coin_transactions WHERE user_id={e['id']} AND type='search_refund'") == "1", d)
reset(); e = reg(); call("POST", "matchmaking/join.php", {"mode": "text"}, e["tok"]); sql("UPDATE matchmaking_queue SET expires_at=NOW()-INTERVAL 1 MINUTE")
o = subprocess.run(["php", "/home/claude/wibber-backend/app/cron/cleanup.php"], capture_output=True, text=True)
check("cron expires stale queue rows and refunds", json.loads(o.stdout)["queue_expired"] == 1 and bal(e) == 20, o.stdout)

print("== Stage 6: text chat")
reset(); p, q2 = reg(), reg(); rooms = settle([p, q2], "text"); sid = rooms[p["id"]]["room_id"]
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "hello", "client_msg_id": "m1"}, p["tok"]); m1 = d["data"]["id"]; check("send message", s == 201, d)
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "hello", "client_msg_id": "m1"}, p["tok"]); check("retry with same client_msg_id is deduped", d["data"]["id"] == m1 and sql(f"SELECT COUNT(*) FROM text_messages WHERE session_id={sid}") == "1", d)
call("POST", "chat/send.php", {"session_id": sid, "body": "hi back ✓ नमस्ते"}, q2["tok"])
s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id=0", None, p["tok"]); ids = [m["id"] for m in d["data"]["messages"]]
check("messages ordered, unicode intact, 'mine' flag", ids == sorted(ids) and len(ids) == 2 and d["data"]["messages"][1]["body"] == "hi back ✓ नमस्ते" and d["data"]["messages"][0]["mine"] is True, d)
s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id={ids[0]}", None, p["tok"]); check("after_id returns only new messages", [m["id"] for m in d["data"]["messages"]] == [ids[1]], d)
call("POST", "chat/typing.php", {"session_id": sid}, q2["tok"]); s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id=99999", None, p["tok"]); check("typing indicator visible to partner", d["data"]["partner_typing"] is True and d["data"]["messages"] == [], d)
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "   "}, p["tok"]); check("blank message rejected", s == 422, d)
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "x" * 1001}, p["tok"]); check("oversize message rejected", s == 422, d)
out = reg(); s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id=0", None, out["tok"]); check("outsider cannot read chat", s == 404, d)
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "intrude"}, out["tok"]); check("outsider cannot post", s == 404, d)
sql("DELETE FROM rate_limits"); n429 = 0
for i in range(35):
    s, d = call("POST", "chat/send.php", {"session_id": sid, "body": f"spam{i}"}, p["tok"]); n429 += s == 429
check("message flood rate-limited (30/min)", n429 >= 4, n429)
print("== report / block")
sql("DELETE FROM rate_limits")
s, d = call("POST", "reports/create.php", {"room_type": "text", "room_id": sid, "reason": "abuse"}, p["tok"]); call("POST", "reports/create.php", {"room_type": "text", "room_id": sid, "reason": "abuse"}, p["tok"])
check("report derives reported user server-side; duplicate collapsed", s == 201 and sql(f"SELECT reported_id FROM user_reports WHERE reporter_id={p['id']}") == str(q2["id"]) and sql("SELECT COUNT(*) FROM user_reports") == "1", d)
s, d = call("POST", "reports/create.php", {"room_type": "text", "room_id": sid, "reason": "nope"}, p["tok"]); check("unknown reason rejected", s == 422, d)
s, d = call("POST", "reports/create.php", {"room_type": "text", "room_id": sid, "reason": "spam"}, out["tok"]); check("outsider cannot report a room", s == 404, d)
s, d = call("POST", "blocks/create.php", {"user_id": out["id"]}, p["tok"]); check("cannot block someone never matched with", s == 404, d)
s, d = call("POST", "blocks/create.php", {"user_id": p["id"]}, p["tok"]); check("cannot block self", s == 422, d)
s, d = call("POST", "blocks/create.php", {"user_id": q2["id"]}, p["tok"]); check("block partner ends the live chat", s == 201 and sql(f"SELECT status FROM text_chat_sessions WHERE id={sid}") == "ended", d)
s, d = call("GET", "blocks/list.php", None, p["tok"]); check("block list", len(d["data"]["blocks"]) == 1 and d["data"]["blocks"][0]["user_id"] == q2["id"], d)
s, d = call("POST", "chat/send.php", {"session_id": sid, "body": "after end"}, q2["tok"]); check("send after end -> 410", s == 410, d)
s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id=0", None, q2["tok"]); check("partner reads ended state", d["data"]["status"] == "ended", d)
s, d = call("POST", "blocks/remove.php", {"user_id": q2["id"]}, p["tok"]); check("unblock", s == 200 and sql("SELECT COUNT(*) FROM user_blocks") == "0")
sql(f"UPDATE text_chat_sessions SET ended_at=NOW()-INTERVAL 48 HOUR WHERE id={sid}")
o = subprocess.run(["php", "/home/claude/wibber-backend/app/cron/cleanup.php"], capture_output=True, text=True)
check("retention: chat with OPEN report is kept past 24h", sql(f"SELECT COUNT(*) FROM text_chat_sessions WHERE id={sid}") == "1", o.stdout)
sql("UPDATE user_reports SET status='dismissed'"); o = subprocess.run(["php", "/home/claude/wibber-backend/app/cron/cleanup.php"], capture_output=True, text=True)
check("retention: after report closed, chat + messages purged", sql(f"SELECT COUNT(*) FROM text_chat_sessions WHERE id={sid}") == "0" and sql(f"SELECT COUNT(*) FROM text_messages WHERE session_id={sid}") == "0", o.stdout)
print("== account deletion")
dd = reg(); em = sql(f"SELECT email FROM users WHERE id={dd['id']}")
s, d = call("POST", "users/delete.php", {"password": "wrongpw1"}, dd["tok"]); check("delete needs correct password", s == 401, d)
s, d = call("POST", "users/delete.php", {"password": "secret1"}, dd["tok"]); check("delete account", s == 200, d)
s, d = call("GET", "auth/me.php", None, dd["tok"]); check("token dead after deletion", s == 401, d)
s, d = call("POST", "auth/login.php", {"email": em, "password": "secret1"}, None); check("cannot log in to deleted account", s == 401, d)
sql("DELETE FROM rate_limits"); s, d = call("POST", "auth/register.php", {"email": em, "password": "secret1", "username": "Again"}); check("email can be re-registered", s == 201, d)
print("== suspended account, existing token")
sus = reg(); sql(f"UPDATE users SET status='suspended' WHERE id={sus['id']}")
s, d = call("GET", "auth/me.php", None, sus["tok"]); check("suspended user's existing token is refused (403)", s == 403 and d["error"]["code"] == "account_suspended", d)
s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, sus["tok"]); check("suspended user cannot matchmake", s == 403, d)

print("== Android-integration additions")
reset(); sw = reg(); s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, sw["tok"]); q1 = d["data"]["queue_id"]
s, d = call("POST", "matchmaking/join.php", {"mode": "text"}, sw["tok"])
check("switching mode cancels+refunds old search, charges new once", s == 200 and d["data"]["queue_id"] != q1 and d["data"]["mode"] == "text" and bal(sw) == 15 and sql(f"SELECT status FROM matchmaking_queue WHERE id={q1}") == "cancelled" and sql(f"SELECT COUNT(*) FROM matchmaking_queue WHERE user_id={sw['id']} AND is_open=1") == "1", d)
reset(); old, fresh = reg(), reg(); call("POST", "matchmaking/join.php", {"mode": "video"}, old["tok"])
sql(f"UPDATE users SET last_seen_at=NOW()-INTERVAL 60 SECOND WHERE id={old['id']}")
s, d = call("POST", "matchmaking/join.php", {"mode": "video"}, fresh["tok"]); s, d = call("GET", f"matchmaking/status.php?queue_id={d['data']['queue_id']}", None, fresh["tok"])
check("candidate who stopped polling 60s ago is NOT matched", d["data"]["state"] == "waiting", d)
reset(); p, q2 = reg(), reg(); rooms = settle([p, q2], "text"); sid = rooms[p["id"]]["room_id"]
call("POST", "chat/send.php", {"session_id": sid, "body": "mine", "client_msg_id": "c-1"}, p["tok"]); call("POST", "chat/send.php", {"session_id": sid, "body": "theirs", "client_msg_id": "c-2"}, q2["tok"])
s, d = call("GET", f"chat/messages.php?session_id={sid}&after_id=0", None, p["tok"]); m = d["data"]["messages"]
check("messages expose client_msg_id only for own messages, plus unix ts", m[0]["client_msg_id"] == "c-1" and "client_msg_id" not in m[1] and isinstance(m[0]["ts"], int) and m[0]["ts"] > 1700000000, m)
f = [n for n, ok in RESULTS if not ok]; print(f"\n{len(RESULTS)-len(f)}/{len(RESULTS)} passed"); print("FAILED: " + "; ".join(f) if f else "ALL PASSED"); sys.exit(1 if f else 0)