# A complete webhook receiver: verifies the signature, records each event once, answers 2xx
# quickly and processes events in a background thread. Python 3.9+ standard library, plus
# requests for the API call.
#
#   RELATIONSAI_WEBHOOK_SECRET=whsec_... RELATIONSAI_API_KEY=rai_test_... python receiver.py
#
# Events are recorded in a SQLite file (WEBHOOK_DB) whose primary key is the event id.
import hashlib
import hmac
import json
import os
import queue
import re
import sqlite3
import sys
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import requests

BASE = os.environ.get("RELATIONSAI_API_BASE", "https://voice-api.relationsai.com/v1")
SECRETS = [s for s in os.environ.get("RELATIONSAI_WEBHOOK_SECRET", "").split(",") if s]
TOLERANCE_SECONDS = 300
if not SECRETS:
    sys.exit("Set RELATIONSAI_WEBHOOK_SECRET to the endpoint's signing secret (whsec_...)")

db = sqlite3.connect(os.environ.get("WEBHOOK_DB", "webhook-events.sqlite3"), check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY, body BLOB, received_at TEXT, processed_at TEXT)")
db.commit()
db_lock = threading.Lock()
work = queue.Queue()


# 1. The signature covers "<t>.<raw body>". Verify the bytes exactly as received.
def verify_signature(raw_body: bytes, header: str) -> bool:
    t = None
    signatures = []
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            t = int(value)
        elif key == "v1" and re.fullmatch(r"[0-9a-f]{64}", value):
            signatures.append(value)
    if t is None or not signatures:
        return False
    if abs(time.time() - t) > TOLERANCE_SECONDS:
        return False
    signed = f"{t}.".encode() + raw_body
    # During a secret rotation the header carries one v1 per secret: accept any match.
    for secret in SECRETS:
        expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, s) for s in signatures):
            return True
    return False


# 3. Business logic, run after the 2xx. Deliveries can arrive in any order.
def handle(event):
    obj = event["data"]["object"]
    if event["type"] == "call.ended":
        # Outcome is known; transcript, recording, analysis and cost may still be pending.
        print(f"call {obj['id']} ended: {obj['outcome']}", flush=True)
    elif event["type"] == "call.settled":
        # Everything is final now: fetch the analysis and write it to your system.
        r = requests.get(
            f"{BASE}/calls/{obj['id']}/analysis",
            headers={"Authorization": f"Bearer {os.environ['RELATIONSAI_API_KEY']}"},
            timeout=30,
        )
        analysis = r.json()
        if not r.ok:
            raise RuntimeError(f"analysis: {analysis['error']['code']} (request {analysis['error']['request_id']})")
        record = obj["metadata"].get("crm_record_id", "(no record id)")
        print(f"call {obj['id']} settled for {record}: {analysis['summary']['text'] or analysis['state']}", flush=True)
    elif event["type"] in ("campaign.paused", "campaign.stopped"):
        print(f"campaign {obj['id']} {obj['status']} (reason: {event['data']['reason']})", flush=True)
    else:
        # Event types are an open list: log and ignore the ones you do not use.
        print(f"{event['type']} for {obj['id']}", flush=True)


def worker():
    while True:
        event_id = work.get()
        with db_lock:
            row = db.execute("SELECT body FROM events WHERE id = ? AND processed_at IS NULL", (event_id,)).fetchone()
        if row is None:
            continue  # already processed
        event = json.loads(row[0])
        try:
            handle(event)
        except Exception as err:  # Already acknowledged: the sweep below retries it.
            print(f"processing {event_id} failed, will retry: {err}", file=sys.stderr, flush=True)
            continue
        with db_lock:
            db.execute("UPDATE events SET processed_at = ? WHERE id = ?", (datetime.now(timezone.utc).isoformat(), event_id))
            db.commit()
        print(f"processed {event_id} {event['type']}", flush=True)


def sweep():
    # Pick up events received but not processed (after a crash or a failed attempt).
    while True:
        with db_lock:
            pending = [r[0] for r in db.execute("SELECT id FROM events WHERE processed_at IS NULL")]
        for event_id in pending:
            work.put(event_id)
        time.sleep(60)


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != "/webhooks/relationsai":
            return self.reply(404, b"")
        length = int(self.headers.get("Content-Length") or 0)
        if length > 1_000_000:
            return self.reply(413, b"too large")
        raw = self.rfile.read(length)
        if not verify_signature(raw, self.headers.get("RelationsAI-Signature")):
            return self.reply(400, b"invalid signature")
        try:
            event = json.loads(raw)
        except ValueError:
            return self.reply(400, b"invalid JSON")
        if not re.fullmatch(r"evt_[A-Za-z0-9_-]{1,60}", str(event.get("id", ""))):
            return self.reply(400, b"invalid event id")
        # 2. Record the event once: the primary key refuses a second copy of the same event id.
        try:
            with db_lock:
                db.execute(
                    "INSERT INTO events (id, body, received_at) VALUES (?, ?, ?)",
                    (event["id"], raw, datetime.now(timezone.utc).isoformat()),
                )
                db.commit()
        except sqlite3.IntegrityError:
            return self.reply(200, b"duplicate")
        self.reply(200, b"ok")
        work.put(event["id"])

    def reply(self, status, body):
        self.send_response(status)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *args):
        pass  # keep the output to one line per event


if __name__ == "__main__":
    server = ThreadingHTTPServer(("", int(os.environ.get("PORT", "8000"))), Handler)
    threading.Thread(target=worker, daemon=True).start()
    threading.Thread(target=sweep, daemon=True).start()
    print(f"Listening on http://localhost:{server.server_address[1]}/webhooks/relationsai", flush=True)
    server.serve_forever()