#!/usr/bin/env python3
"""
UCall L-08: Redis Failure & Resilience Evaluation
──────────────────────────────────────────────────
Verifies:
  1. Redis healthy baseline (active presence + signaling)
  2. Redis failure induced (observe existing connections, heartbeats, routing fallback, REST auth)
  3. Redis recovery (reconnection, presence restoration, subsequent signaling)
"""

import asyncio
import json
import os
import signal
import socket
import subprocess
import sys
import time
import urllib.request
import uuid
from typing import Dict, List, Optional, Tuple

try:
    import psutil
except ImportError:
    subprocess.check_call([sys.executable, "-m", "pip", "install", "psutil", "-q"])
    import psutil

try:
    import websockets
except ImportError:
    subprocess.check_call([sys.executable, "-m", "pip", "install", "websockets", "-q"])
    import websockets

BASE = "http://127.0.0.1:8080"
WS_BASE = "ws://127.0.0.1:8080/api/v1/ws"


def get_redis_pid() -> Optional[int]:
    for p in psutil.process_iter(["pid", "name", "cmdline"]):
        try:
            cmd = " ".join(p.info["cmdline"] or [])
            if "redis-server" in cmd and "grep" not in cmd:
                return p.pid
        except Exception:
            continue
    return None


def get_server_pid() -> Optional[int]:
    for p in psutil.process_iter(["pid", "name", "cmdline"]):
        try:
            cmd = " ".join(p.info["cmdline"] or [])
            if "ucall-server" in cmd and "grep" not in cmd:
                return p.pid
        except Exception:
            continue
    return None


def ping_redis(timeout: float = 0.5) -> bool:
    try:
        s = socket.create_connection(("127.0.0.1", 6379), timeout=timeout)
        s.sendall(b"PING\r\n")
        resp = s.recv(1024)
        s.close()
        return b"PONG" in resp
    except Exception:
        return False


def rand_phone() -> str:
    import random
    return f"+1{random.randint(2000000000, 9999999999)}"


def register_user(phone: str, name: str = "RedisTestUser") -> Optional[dict]:
    body = json.dumps({"phone": phone, "display_name": name}).encode()
    req = urllib.request.Request(
        f"{BASE}/api/v1/auth/register",
        data=body,
        headers={"Content-Type": "application/json"}
    )
    try:
        with urllib.request.urlopen(req, timeout=5) as res:
            data = json.loads(res.read().decode())
            return data.get("data")
    except Exception:
        return None


def get_identity(uid: str, token: str, timeout: float = 3.0) -> Tuple[int, dict]:
    req = urllib.request.Request(f"{BASE}/api/v1/identity/{uid}")
    req.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as res:
            return res.status, json.loads(res.read().decode())
    except urllib.error.HTTPError as e:
        return e.code, {}
    except Exception:
        return 0, {}


def ws_url(token: str, endpoint_id: Optional[uuid.UUID] = None) -> str:
    ep = endpoint_id or uuid.uuid4()
    return f"{WS_BASE}?token={token}&endpoint_id={ep}"


async def main():
    print("╔══════════════════════════════════════════════════════════════╗")
    print("║   UCall L-08: Redis Failure & Resilience Evaluation          ║")
    print("╚══════════════════════════════════════════════════════════════╝")
    print(f"  Target: {BASE}")

    redis_pid = get_redis_pid()
    if not redis_pid:
        print("  ❌ Redis server process not found!")
        return 1
    server_pid = get_server_pid()
    if not server_pid:
        print("  ❌ ucall-server process not found!")
        return 1

    print(f"  ucall-server PID: {server_pid} | redis-server PID: {redis_pid}")
    results = {}

    # ─────────────────────────────────────────────────────────────────────────
    # Phase 1: Baseline / Healthy State
    # ─────────────────────────────────────────────────────────────────────────
    print("\n━━  Phase 1: Baseline (Redis Healthy)  ━━")
    redis_ok = ping_redis()
    print(f"  Redis PING healthy: {redis_ok}")
    assert redis_ok, "Redis must be running at start"

    # Setup 2 test users for signaling & presence
    user_a = register_user(rand_phone(), "UserA")
    user_b = register_user(rand_phone(), "UserB")
    assert user_a and user_b, "Registration failed"

    tok_a, uid_a = user_a["access_token"], user_a["universal_id"]
    tok_b, uid_b = user_b["access_token"], user_b["universal_id"]

    # Connect WebSocket A and B
    ws_a = await websockets.connect(ws_url(tok_a))
    ws_b = await websockets.connect(ws_url(tok_b))
    print("  WebSocket A and B connected successfully")

    # Send heartbeat from B
    await ws_b.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
    await asyncio.sleep(0.1)

    # Check presence of B
    st, ident_b = get_identity(uid_b, tok_a)
    pres_b = ident_b.get("data", {}).get("presence", {})
    is_online_pre = pres_b.get("is_online", False)
    print(f"  Identity resolution for User B: status={st}, is_online={is_online_pre}")

    # Test baseline signaling call
    await ws_a.send(json.dumps({
        "method": "call.invite",
        "params": {
            "target_universal_id": uid_b,
            "call_type": "audio",
            "sdp_offer": "v=0\r\n"
        }
    }))
    msg_raw = await asyncio.wait_for(ws_b.recv(), timeout=3.0)
    msg = json.loads(msg_raw)
    call_id = msg.get("params", {}).get("call_id")
    print(f"  Pre-outage call signaling: received call_id={call_id}")
    await ws_b.send(json.dumps({
        "method": "call.accept",
        "params": {"call_id": call_id, "sdp_answer": "v=0\r\n"}
    }))
    conn_raw = await asyncio.wait_for(ws_a.recv(), timeout=3.0)
    conn_msg = json.loads(conn_raw)
    print(f"  Pre-outage call connected: method={conn_msg.get('method')}")
    await ws_a.send(json.dumps({"method": "call.hangup", "params": {"call_id": call_id}}))

    results["phase1_baseline"] = {
        "redis_healthy": True,
        "signaling_delivered": True,
        "presence_online": is_online_pre,
    }

    # ─────────────────────────────────────────────────────────────────────────
    # Phase 2: Redis Outage Induced
    # ─────────────────────────────────────────────────────────────────────────
    print("\n━━  Phase 2: Redis Outage Induced  ━━")
    print(f"  Suspending Redis (SIGSTOP {redis_pid})...")
    os.kill(redis_pid, signal.SIGSTOP)
    time.sleep(0.2)

    redis_down = not ping_redis(timeout=0.2)
    print(f"  Redis unreachability confirmed: {redis_down}")

    # 1. Check if ucall-server is still alive
    server_alive = psutil.pid_exists(server_pid)
    print(f"  ucall-server process alive (no crash): {server_alive}")

    # 2. Check existing WebSocket connections
    ws_a_open = (getattr(ws_a, "state", None) is not None and ws_a.state.name == "OPEN") or not getattr(ws_a, "closed", True)
    ws_b_open = (getattr(ws_b, "state", None) is not None and ws_b.state.name == "OPEN") or not getattr(ws_b, "closed", True)
    print(f"  Existing WebSocket A open: {ws_a_open}, WebSocket B open: {ws_b_open}")

    # 3. Test sending heartbeat during outage (must not crash server or disconnect socket)
    heartbeat_survived = False
    try:
        await ws_b.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
        heartbeat_survived = True
    except Exception as e:
        print(f"  Heartbeat send error: {e}")
    print(f"  WebSocket survived heartbeat send during outage: {heartbeat_survived}")

    # 4. REST Registration during outage (independent of Redis)
    user_c = register_user(rand_phone(), "UserC_DuringOutage")
    reg_during_outage = user_c is not None
    print(f"  REST User Registration during Redis outage: {reg_during_outage} (UID: {user_c.get('universal_id') if user_c else 'N/A'})")

    # 5. REST Identity lookup during outage (graceful presence fallback)
    st_outage, ident_outage = get_identity(uid_b, tok_a, timeout=2.0)
    pres_outage = ident_outage.get("data", {}).get("presence", {})
    # Should return 200 with presence defaulting to offline
    graceful_presence = (st_outage == 200) and (pres_outage.get("is_online") is False)
    print(f"  Identity resolution during outage: status={st_outage}, is_online={pres_outage.get('is_online')} (Graceful Fallback: {graceful_presence})")

    # 6. Signaling attempt during outage
    # The in-memory fallback routing should attempt to route or handle without panic
    signaling_survived = False
    try:
        await ws_a.send(json.dumps({
            "method": "call.invite",
            "params": {
                "target_universal_id": uid_b,
                "call_type": "audio",
                "sdp_offer": "v=0\r\n"
            }
        }))
        # Wait up to 1.5s for call.incoming
        raw_sig = await asyncio.wait_for(ws_b.recv(), timeout=1.5)
        sig_data = json.loads(raw_sig)
        if sig_data.get("method") == "call.incoming":
            signaling_survived = True
            c_id = sig_data.get("params", {}).get("call_id")
            await ws_b.send(json.dumps({"method": "call.accept", "params": {"call_id": c_id}}))
            raw_conn = await asyncio.wait_for(ws_a.recv(), timeout=1.5)
            await ws_a.send(json.dumps({"method": "call.hangup", "params": {"call_id": c_id}}))
    except Exception:
        # Fallback or graceful timeout without crash
        pass
    print(f"  Call signaling during Redis outage (in-memory delivery): {signaling_survived}")

    results["phase2_outage"] = {
        "server_alive": server_alive,
        "ws_connections_retained": ws_a_open and ws_b_open,
        "heartbeat_survived": heartbeat_survived,
        "registration_succeeded": reg_during_outage,
        "graceful_presence_fallback": graceful_presence,
        "in_memory_signaling": signaling_survived,
    }

    # ─────────────────────────────────────────────────────────────────────────
    # Phase 3: Redis Recovery
    # ─────────────────────────────────────────────────────────────────────────
    print("\n━━  Phase 3: Redis Recovery  ━━")
    print(f"  Resuming Redis (SIGCONT {redis_pid})...")
    os.kill(redis_pid, signal.SIGCONT)
    time.sleep(0.3)

    redis_recovered = ping_redis()
    print(f"  Redis PING recovered: {redis_recovered}")
    assert redis_recovered, "Redis must recover"

    # 1. Send heartbeat from B after recovery
    await ws_b.send(json.dumps({"method": "presence.heartbeat", "params": {}}))
    await asyncio.sleep(0.2)

    # 2. Verify presence in Redis restored
    st_rec, ident_rec = get_identity(uid_b, tok_a)
    pres_rec = ident_rec.get("data", {}).get("presence", {})
    presence_restored = (st_rec == 200) and (pres_rec.get("is_online") is True)
    print(f"  Identity resolution post-recovery: status={st_rec}, is_online={pres_rec.get('is_online')} (Restored: {presence_restored})")

    # 3. Test post-recovery call signaling
    t0 = time.perf_counter()
    await ws_a.send(json.dumps({
        "method": "call.invite",
        "params": {
            "target_universal_id": uid_b,
            "call_type": "audio",
            "sdp_offer": "v=0\r\n"
        }
    }))
    rec_call_raw = await asyncio.wait_for(ws_b.recv(), timeout=3.0)
    rec_call = json.loads(rec_call_raw)
    rec_call_id = rec_call.get("params", {}).get("call_id")
    await ws_b.send(json.dumps({
        "method": "call.accept",
        "params": {"call_id": rec_call_id, "sdp_answer": "v=0\r\n"}
    }))
    rec_conn_raw = await asyncio.wait_for(ws_a.recv(), timeout=3.0)
    rec_conn = json.loads(rec_conn_raw)
    rtt_ms = (time.perf_counter() - t0) * 1000
    signaling_restored = rec_conn.get("method") == "call.connected"
    print(f"  Post-recovery call connected: {signaling_restored} (RTT={rtt_ms:.2f}ms)")
    await ws_a.send(json.dumps({"method": "call.hangup", "params": {"call_id": rec_call_id}}))

    # 4. Connect fresh WebSocket client post-recovery
    user_d = register_user(rand_phone(), "UserD_PostRecovery")
    fresh_ws_ok = False
    if user_d:
        try:
            async with websockets.connect(ws_url(user_d["access_token"]), open_timeout=3.0) as ws_d:
                fresh_ws_ok = True
        except Exception:
            pass
    print(f"  Fresh WebSocket connection post-recovery: {fresh_ws_ok}")

    # Clean close
    await ws_a.close()
    await ws_b.close()

    results["phase3_recovery"] = {
        "redis_recovered": redis_recovered,
        "presence_restored": presence_restored,
        "signaling_restored": signaling_restored,
        "post_recovery_rtt_ms": rtt_ms,
        "fresh_ws_connected": fresh_ws_ok,
    }

    # Summary
    print("\n" + "=" * 65)
    all_pass = (
        results["phase1_baseline"]["signaling_delivered"]
        and results["phase2_outage"]["server_alive"]
        and results["phase2_outage"]["ws_connections_retained"]
        and results["phase2_outage"]["registration_succeeded"]
        and results["phase2_outage"]["graceful_presence_fallback"]
        and results["phase3_recovery"]["redis_recovered"]
        and results["phase3_recovery"]["presence_restored"]
        and results["phase3_recovery"]["signaling_restored"]
    )
    if all_pass:
        print("  L-08 Redis Resilience Evaluation: ✅  PASSED")
    else:
        print("  L-08 Redis Resilience Evaluation: ⚠️   DEVIATIONS OBSERVED")
    print("=" * 65)

    with open("tests/l08_redis_results.json", "w") as f:
        json.dump(results, f, indent=2)

    return 0 if all_pass else 1


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