#!/usr/bin/env python3
"""
Twitch EventSub → Mock Discord Webhook Bridge
==============================================
Receives signed Twitch-style webhook events on POST /webhook,
verifies HMAC signatures, deduplicates by event_id, rate-limits
outbound Discord forwards, and retries on 429 backoff.

Run: python webhook_bridge.py [--secret <key>] [--port <port>]
"""

from __future__ import annotations

import asyncio
import hashlib
import hmac
import http.server
import json
import logging
import os
import queue
import secrets
import socketserver
import sys
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

SECRET = os.environ.get("WEBHOOK_SECRET", secrets.token_hex(16))
PORT = int(os.environ.get("WEBHOOK_PORT", "8080"))
RATE_LIMIT = 5  # Discord forwards per second
IDEMPOTENCY_TTL = 300  # seconds

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("bridge")


# ---------------------------------------------------------------------------
# Mock Discord Webhook Client
# ---------------------------------------------------------------------------

class DiscordMock:
    """Simulates a downstream Discord webhook endpoint.

    Succeeds ~95 % of the time.  Occasionally returns HTTP 429 with a
    ``Retry-After`` header to exercise the backoff path.
    """

    def __init__(self, fail_rate: float = 0.05) -> None:
        self.fail_rate = fail_rate
        self.forward_count: int = 0
        self._lock = asyncio.Lock()

    async def send(self, payload: dict[str, Any]) -> tuple[int, dict[str, str]]:
        """Return (status_code, headers) simulating Discord's response."""
        async with self._lock:
            self.forward_count += 1

        import random

        if random.random() < self.fail_rate:
            retry_after = random.choice([1, 2, 3])
            log.warning("Discord returned 429 (Retry-After=%ds)", retry_after)
            return 429, {"Retry-After": str(retry_after)}

        log.info("Discord forward #%d succeeded", self.forward_count)
        return 200, {}


# ---------------------------------------------------------------------------
# Idempotency Store
# ---------------------------------------------------------------------------

class IdempotencyStore:
    """In-memory deduplication store keyed by ``event_id``.

    Entries older than ``IDEMPOTENCY_TTL`` seconds are evicted on access.
    """

    def __init__(self, ttl: int = IDEMPOTENCY_TTL) -> None:
        self._ttl = ttl
        self._store: dict[str, float] = {}  # event_id -> timestamp

    def is_duplicate(self, event_id: str) -> bool:
        """Return True if *event_id* was seen within the TTL window."""
        now = time.monotonic()
        # Evict stale entries
        self._store = {k: v for k, v in self._store.items() if now - v < self._ttl}
        if event_id in self._store:
            return True
        self._store[event_id] = now
        return False


# ---------------------------------------------------------------------------
# Token-Bucket Rate Limiter
# ---------------------------------------------------------------------------

class TokenBucket:
    """Non-blocking token-bucket rate limiter.

    Capacity and refill rate are both *rate* tokens/second.
    ``try_acquire()`` returns True immediately if a token is available,
    otherwise False.  The caller should ``await self.wait()`` to block
    until a token is available when the bucket is empty.
    """

    def __init__(self, rate: float) -> None:
        self._rate = rate
        self._tokens = float(rate)
        self._last = time.monotonic()
        self._lock = asyncio.Lock()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._last
        self._tokens = min(float(self._rate), self._tokens + elapsed * self._rate)
        self._last = now

    async def try_acquire(self) -> bool:
        """Try to consume a token without blocking. Returns True on success."""
        async with self._lock:
            self._refill()
            if self._tokens >= 1.0:
                self._tokens -= 1.0
                return True
            return False

    async def wait(self) -> None:
        """Block until a token becomes available, then consume it."""
        while True:
            if await self.try_acquire():
                return
            # Sleep just enough for one token to refill
            await asyncio.sleep(1.0 / self._rate)


# ---------------------------------------------------------------------------
# Webhook Bridge Logic
# ---------------------------------------------------------------------------

@dataclass
class EventContext:
    """Holds the parsed event and its raw body for signing."""

    event_id: str
    event_type: str
    data: Any
    raw_body: bytes


class WebhookBridge:
    """Core bridge: verify → deduplicate → rate-limit → forward."""

    def __init__(
        self,
        secret: str,
        discord: DiscordMock,
        store: IdempotencyStore | None = None,
        limiter: TokenBucket | None = None,
    ) -> None:
        self.secret = secret
        self.discord = discord
        self.store = store or IdempotencyStore()
        self.limiter = limiter or TokenBucket(RATE_LIMIT)
        # Track successful forwards for test assertions
        self.successful_forwards: list[dict] = []

    # -- HMAC verification --------------------------------------------------

    def verify_signature(self, body: bytes, signature: str) -> bool:
        """Constant-time HMAC-SHA256 verification."""
        expected = hmac.new(
            self.secret.encode(), body, hashlib.sha256
        ).hexdigest()
        return hmac.compare_digest(expected, signature)

    # -- Event parsing ------------------------------------------------------

    @staticmethod
    def parse_event(body: bytes) -> EventContext | None:
        """Parse and validate the JSON payload. Returns None on failure."""
        try:
            data = json.loads(body)
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            log.error("Malformed JSON body: %s", exc)
            return None

        if not isinstance(data, dict):
            log.error("JSON body is not an object")
            return None

        event_id = data.get("event_id")
        event_type = data.get("type")
        if not event_id or not isinstance(event_id, str):
            log.error("Missing or invalid 'event_id'")
            return None
        if not event_type or not isinstance(event_type, str):
            log.error("Missing or invalid 'type'")
            return None

        return EventContext(
            event_id=event_id,
            event_type=event_type,
            data=data.get("data", {}),
            raw_body=body,
        )

    # -- Forward pipeline ---------------------------------------------------

    async def forward(self, ctx: EventContext) -> int:
        """Run the full pipeline. Returns HTTP status code to send back."""

        # 1. Idempotency check
        if self.store.is_duplicate(ctx.event_id):
            log.info("Duplicate event_id=%s — skipping", ctx.event_id)
            return 200

        # 2. Rate-limit: acquire a token (may block briefly)
        await self.limiter.wait()

        # 3. Build Discord payload
        discord_payload = {
            "content": (
                f"📨 Twitch Event [{ctx.event_type}]\n"
                f"ID: {ctx.event_id}\n"
                f"Data: {json.dumps(ctx.data, default=str)}"
            ),
        }

        # 4. Send to Discord with single-shot 429 backoff retry
        status, headers = await self.discord.send(discord_payload)

        if status == 429:
            retry_after = int(headers.get("Retry-After", "1"))
            log.warning(
                "Discord 429 on event_id=%s — backing off %ds",
                ctx.event_id,
                retry_after,
            )
            await asyncio.sleep(retry_after)
            # Retry once
            status, _ = await self.discord.send(discord_payload)

        if status == 200:
            self.successful_forwards.append(discord_payload)
            log.info("Forwarded event_id=%s ✓", ctx.event_id)
        else:
            log.error("Discord returned %d for event_id=%s", status, ctx.event_id)

        return 200 if status == 200 else 502


# ---------------------------------------------------------------------------
# Async HTTP Server (pure stdlib, no third-party deps)
# ---------------------------------------------------------------------------

class _AsyncHandler:
    """Minimal async HTTP request handler using asyncio streams."""

    def __init__(self, bridge: WebhookBridge) -> None:
        self.bridge = bridge

    async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        try:
            # Read request line
            request_line = await asyncio.wait_for(reader.readline(), timeout=10.0)
            if not request_line:
                return
            parts = request_line.decode().strip().split()
            if len(parts) < 2:
                self._send(writer, 400, "Bad Request")
                return

            method, path = parts[0], parts[1]

            # Read headers
            headers: dict[str, str] = {}
            while True:
                line = await asyncio.wait_for(reader.readline(), timeout=10.0)
                if line in (b"\r\n", b"\n", b""):
                    break
                decoded = line.decode().strip()
                if ":" in decoded:
                    key, val = decoded.split(":", 1)
                    headers[key.strip().lower()] = val.strip()

            # Read body
            body = b""
            content_length = int(headers.get("content-length", "0"))
            if content_length > 0:
                body = await asyncio.wait_for(reader.readexactly(content_length), timeout=10.0)

            if path == "/webhook" and method == "POST":
                await self._handle_webhook(reader, writer, body, headers)
            else:
                self._send(writer, 404, "Not Found")

        except asyncio.TimeoutError:
            self._send(writer, 408, "Request Timeout")
        except Exception as exc:
            log.exception("Unhandled error in request handler")
            self._send(writer, 500, "Internal Server Error")
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            except Exception:
                pass

    async def _handle_webhook(
        self, reader, writer, body: bytes, headers: dict[str, str]
    ) -> None:
        # Signature verification
        signature = headers.get("x-signature")
        if not signature:
            log.warning("Missing X-Signature header")
            self._send(writer, 401, "Unauthorized")
            return

        if not self.bridge.verify_signature(body, signature):
            log.warning("HMAC signature mismatch")
            self._send(writer, 401, "Unauthorized")
            return

        # Parse event
        ctx = self.bridge.parse_event(body)
        if ctx is None:
            self._send(writer, 400, "Bad Request — invalid payload")
            return

        log.info("Received event_id=%s type=%s", ctx.event_id, ctx.event_type)

        # Forward (async pipeline with rate limiting & retry)
        status = await self.bridge.forward(ctx)
        self._send(writer, status, "OK" if status == 200 else "Error")

    @staticmethod
    def _send(writer: asyncio.StreamWriter, status: int, reason: str) -> None:
        body = f"{status} {reason}\n".encode()
        response = (
            b"HTTP/1.1 " + f"{status} {reason}".encode() + b"\r\n"
            b"Content-Type: text/plain\r\n"
            b"Content-Length: " + str(len(body)).encode() + b"\r\n"
            b"Connection: close\r\n"
            b"\r\n"
        )
        writer.write(response + body)


async def run_server(bridge: WebhookBridge, host: str = "127.0.0.1", port: int = PORT) -> None:
    """Start the async HTTP server."""
    server = await asyncio.start_server(
        lambda r, w: _AsyncHandler(bridge).handle(r, w), host, port
    )
    addr = server.sockets[0].getsockname()
    log.info("Bridge listening on http://%s:%d", addr[0], addr[1])
    async with server:
        await server.serve_forever()


# ---------------------------------------------------------------------------
# Threaded HTTP Server (for test compatibility with urllib/http.client)
# ---------------------------------------------------------------------------

class _ThreadedServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    daemon_threads = True
    allow_reuse_address = True


class _BridgeHandler(http.server.BaseHTTPRequestHandler):
    """Synchronous wrapper around the async bridge for use with http.server."""

    bridge: WebhookBridge = None  # type: ignore[assignment]

    def do_POST(self) -> None:  # noqa: N802
        if self.path != "/webhook":
            self.send_response(404)
            self.end_headers()
            self.wfile.write(b"Not Found")
            return

        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        headers_dict = {k.lower(): v for k, v in self.headers.items()}

        # Signature verification
        signature = headers_dict.get("x-signature")
        if not signature:
            self.send_response(401)
            self.end_headers()
            self.wfile.write(b"Unauthorized")
            return

        if not self.bridge.verify_signature(body, signature):
            self.send_response(401)
            self.end_headers()
            self.wfile.write(b"Unauthorized")
            return

        ctx = self.bridge.parse_event(body)
        if ctx is None:
            self.send_response(400)
            self.end_headers()
            self.wfile.write(b"Bad Request")
            return

        # Run async forward pipeline in a thread-safe way
        loop = asyncio.new_event_loop()
        try:
            task = loop.run_until_complete(self.bridge.forward(ctx))
            self.send_response(task)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            self.wfile.write(b"OK" if task == 200 else b"Error")
        finally:
            loop.close()

    def log_message(self, fmt: str, *args: Any) -> None:  # noqa: ARG002
        log.info(fmt, *args)


def start_threaded_server(bridge: WebhookBridge, port: int = PORT) -> _ThreadedServer:
    """Start a threaded HTTP server in a background thread. Returns the server."""
    _BridgeHandler.bridge = bridge
    srv = _ThreadedServer(("127.0.0.1", port), _BridgeHandler)
    thread = threading.Thread(target=srv.serve_forever, daemon=True)
    thread.start()
    log.info("Threaded test server on http://%s:%d", "127.0.0.1", port)
    return srv


# ---------------------------------------------------------------------------
# Helpers for tests
# ---------------------------------------------------------------------------

def make_payload(event_id: str, event_type: str = "chat", data: dict | None = None) -> bytes:
    """Create a signed webhook payload."""
    body_dict = {"event_id": event_id, "type": event_type, "data": data or {}}
    body = json.dumps(body_dict).encode()
    sig = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
    return body, sig


def post_webhook(url: str, body: bytes, signature: str) -> tuple[int, str]:
    """Send a POST request and return (status_code, body_text)."""
    import urllib.request
    import urllib.error

    req = urllib.request.Request(
        url,
        data=body,
        headers={
            "Content-Type": "application/json",
            "X-Signature": signature,
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status, resp.read().decode()
    except urllib.error.HTTPError as exc:
        return exc.code, exc.read().decode()


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------

def run_tests(port: int = 9099) -> int:
    """Run all tests. Returns number of failures."""
    import urllib.request
    import urllib.error

    url = f"http://127.0.0.1:{port}/webhook"
    failures = 0

    def assert_eq(actual, expected, msg: str) -> None:
        nonlocal failures
        if actual != expected:
            print(f"  ✗ FAIL: {msg} — expected {expected!r}, got {actual!r}")
            failures += 1
        else:
            print(f"  ✓ PASS: {msg}")

    def assert_true(condition: bool, msg: str) -> None:
        nonlocal failures
        if not condition:
            print(f"  ✗ FAIL: {msg}")
            failures += 1
        else:
            print(f"  ✓ PASS: {msg}")

    # -----------------------------------------------------------------------
    # Setup fresh bridge for tests
    # -----------------------------------------------------------------------
    discord = DiscordMock(fail_rate=0.0)  # deterministic: no random failures
    bridge = WebhookBridge(secret=SECRET, discord=discord)
    server = start_threaded_server(bridge, port)
    time.sleep(0.2)  # let server start

    try:
        # -------------------------------------------------------------------
        print("\n=== Test (a): Valid signed request → forwarded, returns 200 ===")
        # -------------------------------------------------------------------
        before = discord.forward_count
        body, sig = make_payload("evt-a1", "chat", {"message": "hello"})
        status, text = post_webhook(url, body, sig)
        assert_eq(status, 200, "HTTP status")
        assert_true(discord.forward_count == before + 1, "Discord forward count increased by 1")
        assert_true(len(bridge.successful_forwards) == 1, "Bridge recorded 1 successful forward")

        # -------------------------------------------------------------------
        print("\n=== Test (b): Tampered signature → 401, nothing forwarded ===")
        # -------------------------------------------------------------------
        before = discord.forward_count
        bad_body = body + b"tampered"
        bad_sig = hmac.new(SECRET.encode(), bad_body, hashlib.sha256).hexdigest()
        # Actually tamper: use wrong signature for correct body
        status, text = post_webhook(url, body, "deadbeef" * 8)
        assert_eq(status, 401, "HTTP status")
        assert_true(discord.forward_count == before, "No additional Discord forwards")

        # -------------------------------------------------------------------
        print("\n=== Test (c): Replay same event_id → skipped, count unchanged ===")
        # -------------------------------------------------------------------
        before = discord.forward_count
        status, text = post_webhook(url, body, sig)
        assert_eq(status, 200, "HTTP status (still 200 for replay)")
        assert_true(discord.forward_count == before, "Discord forward count did NOT increase")
        assert_true(len(bridge.successful_forwards) == 1, "Bridge still has only 1 forward")

        # -------------------------------------------------------------------
        print("\n=== Test (d): Burst >5 events/sec → rate limiter delays excess ===")
        # -------------------------------------------------------------------
        # We send 10 events rapidly. The limiter allows 5/sec, so the last
        # 5 should be delayed. We measure wall-clock time to verify.
        burst_ids = [f"evt-burst-{i}" for i in range(10)]
        payloads_sigs = [make_payload(eid, "chat", {"n": i}) for i, eid in enumerate(burst_ids)]

        start_time = time.monotonic()
        results: list[tuple[int, float]] = []  # (status, timestamp)
        for i, (b, s) in enumerate(payloads_sigs):
            st, _ = post_webhook(url, b, s)
            results.append((st, time.monotonic() - start_time))

        elapsed = time.monotonic() - start_time
        assert_true(all(r[0] == 200 for r in results), "All burst requests returned 200")

        # With a 5/sec limiter and 10 events, total time should be > 1s
        # (first 5 immediate, next 5 wait ~1s for tokens)
        assert_true(
            elapsed >= 0.8,
            f"Burst took {elapsed:.2f}s (expected ≥0.8s due to rate limiting)",
        )

        # At most 5 forwards should have succeeded (the rest may have hit 429
        # in the mock, but with fail_rate=0.0 they should all succeed eventually)
        # Actually with the token bucket + retry, all 10 should forward but
        # spread over time. Let's just verify the timing constraint above.
        print(f"    Burst completed in {elapsed:.2f}s with {discord.forward_count} forwards")

        # -------------------------------------------------------------------
        print("\n=== Test (e): Missing X-Signature header → 401 ===")
        # -------------------------------------------------------------------
        body_only = json.dumps({"event_id": "no-sig", "type": "chat"}).encode()
        req = urllib.request.Request(
            url, data=body_only, headers={"Content-Type": "application/json"}, method="POST"
        )
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                assert_true(False, "Expected 401 for missing signature")
        except urllib.error.HTTPError as exc:
            assert_eq(exc.code, 401, "Missing header → 401")

        # -------------------------------------------------------------------
        print("\n=== Test (f): Malformed JSON → 400 ===")
        # -------------------------------------------------------------------
        req = urllib.request.Request(
            url,
            data=b"{not valid json",
            headers={
                "Content-Type": "application/json",
                "X-Signature": hmac.new(SECRET.encode(), b"{not valid json", hashlib.sha256).hexdigest(),
            },
            method="POST",
        )
        try:
            with urllib.request.urlopen(req, timeout=5) as resp:
                assert_true(False, "Expected 400 for malformed JSON")
        except urllib.error.HTTPError as exc:
            assert_eq(exc.code, 400, "Malformed JSON → 400")

    finally:
        server.shutdown()
        server.server_close()

    print(f"\n{'='*50}")
    if failures:
        print(f"FAILED: {failures} test(s) failed")
    else:
        print("ALL TESTS PASSED ✓")
    return failures


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="Twitch EventSub → Discord webhook bridge")
    parser.add_argument("--secret", default=SECRET, help="HMAC shared secret")
    parser.add_argument("--port", type=int, default=PORT, help="HTTP listen port")
    parser.add_argument("--test", action="store_true", help="Run tests and exit")
    args = parser.parse_args()

    global SECRET, PORT
    SECRET = args.secret
    PORT = args.port

    if args.test:
        sys.exit(run_tests(PORT))

    discord = DiscordMock()
    bridge = WebhookBridge(secret=SECRET, discord=discord)

    print(f"Starting bridge on http://127.0.0.1:{PORT}")
    print(f"Secret: {SECRET}")
    print("Press Ctrl+C to stop")

    try:
        asyncio.run(run_server(bridge, "127.0.0.1", PORT))
    except KeyboardInterrupt:
        print("\nShutting down.")


if __name__ == "__main__":
    main()