#!/usr/bin/env python3
"""PocketBase integration checks for KQ support booking invariants.

Spins a temporary PocketBase (KQ_POCKETBASE_BIN) with the real hooks, applies the
auth/family and support schema through the actual setup scripts, and exercises the
support tool/teacher routes. A local HTTP server stands in for the n8n confirmation
webhook. No live service is contacted.
"""

import concurrent.futures
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

HERE = Path(__file__).resolve().parent
HOOKS = HERE / "pb_hooks"
SCRIPTS = HERE.parent / "scripts"

ART = timezone(timedelta(hours=-3))
TOOL_TOKEN = "integration-tool-token"


def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


class WebhookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length) or b"{}")
        with self.server.calls_lock:
            self.server.calls.append({
                "body": payload,
                "token": self.headers.get("X-KQ-Tool-Token", ""),
            })
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"ok":true}')

    def log_message(self, *_):
        return


def slot_at(hours_from_now, duration_minutes=60):
    start = datetime.now(ART) + timedelta(hours=hours_from_now)
    end = start + timedelta(minutes=duration_minutes)
    date = start.strftime("%Y-%m-%d")
    s = start.strftime("%H:%M")
    e = end.strftime("%H:%M") if end.strftime("%Y-%m-%d") == date else "23:59"
    if e <= s:
        e = "23:59"
    return date, s, e


class SupportBookingIntegrationTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        binary = os.environ.get("KQ_POCKETBASE_BIN", "")
        if not binary or not Path(binary).is_file():
            raise unittest.SkipTest("KQ_POCKETBASE_BIN is required")

        cls.temp = tempfile.TemporaryDirectory(prefix="kq-support-integration-")
        cls.root = Path(cls.temp.name)
        cls.data_dir = cls.root / "pb_data"
        cls.hooks_dir = cls.root / "pb_hooks"
        cls.hooks_dir.mkdir(parents=True)
        for filename in ("main.pb.js", "auth-service.js", "auth-utils.js", "support-utils.js"):
            shutil.copy2(HOOKS / filename, cls.hooks_dir / filename)

        cls.webhook = ThreadingHTTPServer(("127.0.0.1", free_port()), WebhookHandler)
        cls.webhook.calls = []
        cls.webhook.calls_lock = threading.Lock()
        cls.webhook_thread = threading.Thread(target=cls.webhook.serve_forever, daemon=True)
        cls.webhook_thread.start()

        cls.pb_port = free_port()
        cls.base_url = f"http://127.0.0.1:{cls.pb_port}"
        cls.admin_email = "support-admin@kq.local"
        cls.admin_password = "support-only-password-123"
        subprocess.run(
            [binary, "superuser", "upsert", cls.admin_email, cls.admin_password,
             "--dir", str(cls.data_dir), "--hooksDir", str(cls.hooks_dir)],
            check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
        )
        env = os.environ.copy()
        env.update({
            "EVOLUTION_URL": "http://127.0.0.1:9",
            "EVOLUTION_INSTANCE": "Kings-and-Queens",
            "EVOLUTION_APIKEY": "integration-key",
            "KQ_OTP_PEPPER": "integration-pepper",
            "KQ_TOOL_TOKEN": TOOL_TOKEN,
            "KQ_TEST_MODE": "true",
            "KQ_CHATWOOT_HOST": "chat.example.com",
            "KQ_CONFIRMATION_WEBHOOK_URL": f"http://127.0.0.1:{cls.webhook.server_port}/webhook/kq-support-confirmed",
        })
        cls.pb_log = tempfile.TemporaryFile(mode="w+")
        cls.pb = subprocess.Popen(
            [binary, "serve", "--http", f"127.0.0.1:{cls.pb_port}",
             "--dir", str(cls.data_dir), "--hooksDir", str(cls.hooks_dir),
             "--hooksWatch=false", "--dev=false"],
            env=env, stdout=cls.pb_log, stderr=subprocess.STDOUT,
        )
        cls._wait_for_server()
        _, auth = cls.request("POST", "/api/collections/_superusers/auth-with-password",
                              {"identity": cls.admin_email, "password": cls.admin_password})
        cls.admin_token = auth["token"]
        cls._create_prerequisite_collections()
        cls._apply_schema(binary)
        cls._create_teacher()

    @classmethod
    def tearDownClass(cls):
        if hasattr(cls, "pb_log") and os.environ.get("KQ_DUMP_PB_LOG"):
            cls.pb_log.seek(0)
            Path(os.environ["KQ_DUMP_PB_LOG"]).write_text(cls.pb_log.read())
        if hasattr(cls, "pb"):
            cls.pb.terminate()
            try:
                cls.pb.wait(timeout=5)
            except subprocess.TimeoutExpired:
                cls.pb.kill()
        if hasattr(cls, "pb_log"):
            cls.pb_log.close()
        if hasattr(cls, "webhook"):
            cls.webhook.shutdown()
            cls.webhook.server_close()
        if hasattr(cls, "temp"):
            cls.temp.cleanup()

    @classmethod
    def _wait_for_server(cls):
        deadline = time.time() + 20
        while time.time() < deadline:
            if cls.pb.poll() is not None:
                cls.pb_log.seek(0)
                raise RuntimeError(cls.pb_log.read())
            try:
                with urllib.request.urlopen(f"{cls.base_url}/api/health", timeout=0.5):
                    return
            except (urllib.error.URLError, TimeoutError):
                time.sleep(0.1)
        raise RuntimeError("temporary PocketBase did not start")

    @classmethod
    def request(cls, method, path, payload=None, token=None, headers=None, timeout=15):
        body = json.dumps(payload).encode() if payload is not None else None
        req = urllib.request.Request(cls.base_url + path, data=body, method=method)
        req.add_header("Content-Type", "application/json")
        if token:
            req.add_header("Authorization", token)
        for key, value in (headers or {}).items():
            req.add_header(key, value)
        try:
            with urllib.request.urlopen(req, timeout=timeout) as response:
                raw = response.read().decode()
                return response.status, json.loads(raw) if raw else {}
        except urllib.error.HTTPError as error:
            raw = error.read().decode()
            return error.code, (json.loads(raw) if raw else {})

    @classmethod
    def _create_prerequisite_collections(cls):
        # In production these already exist (setup_pocketbase.py). The family/support
        # setup scripts only extend them, so create minimal compatible versions here.
        classes = {
            "name": "classes", "type": "base", "fields": [
                {"name": "title", "type": "text", "required": False},
                {"name": "date", "type": "text", "required": True},
                {"name": "time_start", "type": "text", "required": False},
                {"name": "time_end", "type": "text", "required": False},
                {"name": "student_id", "type": "text", "required": False},
                {"name": "status", "type": "select", "required": True, "maxSelect": 1,
                 "values": ["scheduled", "active", "finished", "cancelled", "absent", "rescheduled"]},
                {"name": "notes", "type": "text", "required": False},
            ],
        }
        student_profiles = {
            "name": "student_profiles", "type": "base", "fields": [
                {"name": "user_id", "type": "text", "required": True},
                {"name": "level", "type": "text", "required": False},
                {"name": "phone", "type": "text", "required": False},
                {"name": "enrollment_date", "type": "text", "required": False},
                {"name": "is_active", "type": "bool", "required": False},
            ],
        }
        for definition in (classes, student_profiles):
            status, body = cls.request("POST", "/api/collections", definition, cls.admin_token)
            if status not in (200, 201):
                raise RuntimeError(f"prerequisite {definition['name']} failed ({status}): {body}")

    @classmethod
    def _apply_schema(cls, binary):
        env = os.environ.copy()
        env.update({
            "KQ_PB_URL": cls.base_url,
            "KQ_PB_ADMIN_EMAIL": cls.admin_email,
            "KQ_PB_ADMIN_PASSWORD": cls.admin_password,
        })
        for script in ("setup_auth_family.py", "setup_support_booking.py"):
            result = subprocess.run(
                [sys.executable, str(SCRIPTS / script)],
                env=env, capture_output=True, text=True)
            if result.returncode != 0:
                raise RuntimeError(f"{script} failed: {result.stdout}\n{result.stderr}")

    @classmethod
    def _create_teacher(cls):
        cls.teacher_email = "teacher@kq.local"
        cls.teacher_password = "teacher-password-123"
        status, body = cls.request("POST", "/api/collections/users/records", {
            "email": cls.teacher_email,
            "password": cls.teacher_password,
            "passwordConfirm": cls.teacher_password,
            "rol": "teacher",
            "name": "Ale",
            "verified": True,
            "emailVisibility": True,
        }, cls.admin_token)
        if status not in (200, 201):
            raise RuntimeError(f"teacher create failed ({status}): {body}")
        _, auth = cls.request("POST", "/api/collections/users/auth-with-password",
                              {"identity": cls.teacher_email, "password": cls.teacher_password})
        cls.teacher_token = auth["token"]

    # ── helpers ──────────────────────────────────────────────────────────────
    def tool(self, path, payload):
        return self.request("POST", path, payload, headers={"X-KQ-Tool-Token": TOOL_TOKEN})

    def teacher(self, path, payload):
        return self.request("POST", path, payload, token=self.teacher_token)

    def make_slot(self, hours, capacity=2, deposit=5000):
        date, start, end = slot_at(hours)
        status, body = self.teacher("/api/kq/support/slot", {
            "date": date, "time_start": start, "time_end": end,
            "capacity": capacity, "deposit_amount": deposit, "status": "open",
        })
        self.assertEqual(status, 200, body)
        return body["slot"]["id"], date, start, end

    def hold(self, slot_id, conversation_id, phone="2910000001"):
        return self.tool("/api/kq/support/hold", {
            "slot_id": slot_id, "student_name": "Test Student", "contact_phone": phone,
            "education_level": "primary", "school_year": "5to", "topic": "verbs",
            "conversation_id": conversation_id,
        })

    def admin_patch_booking(self, booking_id, patch):
        return self.request("PATCH", f"/api/collections/support_bookings/records/{booking_id}",
                            patch, self.admin_token)

    def seat_bookings(self, slot_id):
        status, body = self.request(
            "GET",
            f"/api/collections/support_bookings/records?perPage=200&filter=" +
            urllib.parse.quote(f"slot_id='{slot_id}'"),
            token=self.admin_token)
        self.assertEqual(status, 200, body)
        return body.get("items", [])

    # ── tests ────────────────────────────────────────────────────────────────
    def test_a_concurrent_last_seat(self):
        slot_id, _, _, _ = self.make_slot(120, capacity=1)
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            results = list(pool.map(lambda cid: self.hold(slot_id, cid), [900001, 900002]))
        statuses = sorted(status for status, _ in results)
        self.assertEqual(statuses, [200, 409], results)
        held = [b for b in self.seat_bookings(slot_id) if b["status"] == "held"]
        self.assertEqual(len(held), 1, held)

    def test_b_search_orders_and_caps_three(self):
        # Create four open future slots; search must return at most three, ordered.
        for h in (130, 131, 132, 133):
            self.make_slot(h, capacity=2)
        status, body = self.tool("/api/kq/support/search", {})
        self.assertEqual(status, 200, body)
        slots = body["slots"]
        self.assertLessEqual(len(slots), 3, slots)
        self.assertEqual(slots, sorted(slots, key=lambda s: (s["date"], s["time_start"])), slots)

    def test_c_hold_expiry(self):
        slot_id, _, _, _ = self.make_slot(140, capacity=2)
        status, body = self.hold(slot_id, 910001)
        self.assertEqual(status, 200, body)
        booking_id = body["booking_id"]
        past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
        s, b = self.admin_patch_booking(booking_id, {"hold_expires_at": past})
        self.assertIn(s, (200, 204), b)
        es, eb = self.teacher("/api/kq/support/_test/expire", {})
        self.assertEqual(es, 200, eb)
        s, b = self.request("GET", f"/api/collections/support_bookings/records/{booking_id}",
                            token=self.admin_token)
        self.assertEqual(b["status"], "expired", b)

    def test_d_payment_review_consumes_seat(self):
        slot_id, _, _, _ = self.make_slot(150, capacity=1)
        status, body = self.hold(slot_id, 920001)
        self.assertEqual(status, 200, body)
        past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
        self.admin_patch_booking(body["booking_id"], {"status": "payment_review", "hold_expires_at": past})
        # A second hold must be rejected: the payment_review seat still counts.
        s2, b2 = self.hold(slot_id, 920002)
        self.assertEqual(s2, 409, b2)

    def test_e_review_approve_fires_one_webhook(self):
        slot_id, _, _, _ = self.make_slot(160, capacity=2)
        status, body = self.hold(slot_id, 930001)
        self.assertEqual(status, 200, body)
        booking_id = body["booking_id"]
        self.admin_patch_booking(booking_id, {"status": "payment_review"})
        with self.webhook.calls_lock:
            before = len(self.webhook.calls)
        s, b = self.teacher("/api/kq/support/review", {"booking_id": booking_id, "decision": "approve"})
        self.assertEqual(s, 200, b)
        self.assertEqual(b["booking"]["status"], "confirmed", b)
        time.sleep(0.5)
        with self.webhook.calls_lock:
            new_calls = self.webhook.calls[before:]
        self.assertEqual(len(new_calls), 1, new_calls)
        self.assertEqual(new_calls[0]["token"], TOOL_TOKEN)
        self.assertEqual(new_calls[0]["body"]["booking_id"], booking_id)

    def test_f_cancellation_credit_and_forfeit(self):
        far_slot, _, _, _ = self.make_slot(30, capacity=2)   # >24h ahead → credit
        near_slot, _, _, _ = self.make_slot(12, capacity=2)  # <24h ahead → cancelled
        s1, b1 = self.hold(far_slot, 940001, phone="2915550001")
        s2, b2 = self.hold(near_slot, 940002, phone="2915550002")
        self.assertEqual(s1, 200, b1)
        self.assertEqual(s2, 200, b2)

        c1, cb1 = self.tool("/api/kq/support/cancel", {"booking_code": b1["booking_id"], "contact_phone": "2915550001"})
        c2, cb2 = self.tool("/api/kq/support/cancel", {"booking_code": b2["booking_id"], "contact_phone": "2915550002"})
        self.assertEqual(c1, 200, cb1)
        self.assertEqual(cb1["status"], "credit", cb1)
        self.assertEqual(cb1["credit_used"], False, cb1)
        self.assertEqual(c2, 200, cb2)
        self.assertEqual(cb2["status"], "cancelled", cb2)

        # Reprogram the credit once → 200; a second attempt to a different slot → 409.
        target_a, _, _, _ = self.make_slot(40, capacity=2)
        target_b, _, _, _ = self.make_slot(42, capacity=2)
        r1, rb1 = self.tool("/api/kq/support/reschedule",
                            {"booking_code": b1["booking_id"], "contact_phone": "2915550001", "slot_id": target_a})
        self.assertEqual(r1, 200, rb1)
        r2, rb2 = self.tool("/api/kq/support/reschedule",
                            {"booking_code": b1["booking_id"], "contact_phone": "2915550001", "slot_id": target_b})
        self.assertEqual(r2, 409, rb2)


if __name__ == "__main__":
    unittest.main(verbosity=2)