#!/usr/bin/env python3
"""PocketBase 0.36.8 integration checks for KQ auth/family invariants.

Requires KQ_POCKETBASE_BIN. Everything runs in a temporary directory with a
local Evolution mock; no live service is contacted.
"""

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


HERE = Path(__file__).resolve().parent
SCRIPTS = HERE.parent.parent / "scripts"
sys.path.insert(0, str(SCRIPTS))

from setup_auth_family import build_schema  # noqa: E402


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


class EvolutionHandler(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(payload)
            self.server.call_event.set()
        time.sleep(self.server.response_delay)
        self.send_response(self.server.response_status)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"ok":true}')

    def log_message(self, _format, *_args):
        return


class PocketBaseAuthIntegrationTests(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-auth-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"):
            shutil.copy2(HERE / filename, cls.hooks_dir / filename)

        cls.evolution = ThreadingHTTPServer(("127.0.0.1", free_port()), EvolutionHandler)
        cls.evolution.calls = []
        cls.evolution.calls_lock = threading.Lock()
        cls.evolution.call_event = threading.Event()
        cls.evolution.response_delay = 0
        cls.evolution.response_status = 200
        cls.evolution_thread = threading.Thread(target=cls.evolution.serve_forever, daemon=True)
        cls.evolution_thread.start()

        cls.pb_port = free_port()
        cls.base_url = f"http://127.0.0.1:{cls.pb_port}"
        cls.admin_email = "integration-admin@kq.local"
        cls.admin_password = "integration-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": f"http://127.0.0.1:{cls.evolution.server_port}",
                "EVOLUTION_INSTANCE": "Kings-and-Queens",
                "EVOLUTION_APIKEY": "integration-key",
                "KQ_OTP_PEPPER": "integration-pepper",
            }
        )
        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_schema()
        cls._create_fixtures()

    @classmethod
    def tearDownClass(cls):
        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, "evolution"):
            cls.evolution.shutdown()
            cls.evolution.server_close()
        if hasattr(cls, "temp"):
            cls.temp.cleanup()

    @classmethod
    def _wait_for_server(cls):
        deadline = time.time() + 15
        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, timeout=10):
        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)
        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_collection(cls, definition):
        status, body = cls.request("POST", "/api/collections", definition, cls.admin_token)
        if status not in (200, 201):
            raise RuntimeError(f"collection create failed ({status}): {body}")

    @classmethod
    def _create_schema(cls):
        schema = build_schema()
        users = dict(schema["existing_collections"]["users"])
        status, current_users = cls.request(
            "GET", "/api/collections/users", token=cls.admin_token
        )
        if status != 200:
            raise RuntimeError("default users collection missing")
        desired_user_fields = [
            *users.pop("fields"),
            {"name": "display_name", "type": "text", "required": False},
        ]
        merged_user_fields = list(current_users["fields"])
        existing_names = {field["name"] for field in merged_user_fields}
        merged_user_fields.extend(
            field for field in desired_user_fields if field["name"] not in existing_names
        )
        users["fields"] = merged_user_fields
        users["indexes"] = [
            *(current_users.get("indexes") or []),
            *[
                index
                for index in users.get("indexes", [])
                if index not in (current_users.get("indexes") or [])
            ],
        ]
        status, body = cls.request(
            "PATCH", f"/api/collections/{current_users['id']}", users, cls.admin_token
        )
        if status not in (200, 204):
            raise RuntimeError(f"users update failed ({status}): {body}")

        students = dict(schema["existing_collections"]["student_profiles"])
        students.update({"name": "student_profiles"})
        students["fields"] = [
            {"name": "user_id", "type": "relation", "required": True, "collectionId": "_pb_users_auth_", "maxSelect": 1, "cascadeDelete": True},
            {"name": "phone", "type": "text", "required": False},
            {"name": "is_active", "type": "bool", "required": False},
            *students["fields"],
        ]
        cls._create_collection(students)
        for name, definition in schema["collections"].items():
            cls._create_collection({"name": name, **definition})

    @classmethod
    def _record(cls, collection, payload):
        status, body = cls.request(
            "POST", f"/api/collections/{collection}/records", payload, cls.admin_token
        )
        if status not in (200, 201):
            raise RuntimeError(f"record create failed ({collection}, {status}): {body}")
        return body

    @classmethod
    def _user(cls, email, role, phone="", password="integration-user-password"):
        return cls._record(
            "users",
            {
                "email": email,
                "password": password,
                "passwordConfirm": password,
                "verified": True,
                "rol": role,
                "phone": phone,
                "otp_enabled": True,
                "display_name": role.title(),
            },
        )

    @classmethod
    def _student(cls, suffix, phone=""):
        user = cls._user(f"student-{suffix}@kq.local", "student", phone)
        profile = cls._record(
            "student_profiles",
            {"user_id": user["id"], "phone": phone, "is_active": True, "is_minor": True},
        )
        return user, profile

    @classmethod
    def _guardian(cls, suffix, phone):
        user = cls._user(f"guardian-{suffix}@kq.local", "guardian", phone)
        profile = cls._record(
            "guardian_profiles", {"user_id": user["id"], "phone": phone, "is_active": True}
        )
        return user, profile

    @classmethod
    def _create_fixtures(cls):
        cls.teacher_phone = "5492917777777"
        cls.teacher_password = "integration-teacher-password"
        cls.teacher = cls._user(
            "teacher@kq.local", "teacher", cls.teacher_phone, cls.teacher_password
        )
        status, auth = cls.request(
            "POST",
            "/api/collections/users/auth-with-password",
            {"identity": "teacher@kq.local", "password": cls.teacher_password},
        )
        if status != 200:
            raise RuntimeError("teacher auth failed")
        cls.teacher_token = auth["token"]
        cls.student_a, cls.student_profile_a = cls._student("a")
        cls.student_b, cls.student_profile_b = cls._student("b")
        cls.guardian_a, cls.guardian_profile_a = cls._guardian("a", "5492918100001")
        cls.guardian_b, cls.guardian_profile_b = cls._guardian("b", "5492918100002")

    @classmethod
    def _list(cls, collection):
        status, body = cls.request(
            "GET", f"/api/collections/{collection}/records?perPage=500", token=cls.admin_token
        )
        if status != 200:
            raise RuntimeError(f"record list failed ({collection}, {status})")
        return body["items"]

    @classmethod
    def _clear(cls, collection):
        try:
            records = cls._list(collection)
        except RuntimeError:
            return
        for record in records:
            status, _ = cls.request(
                "DELETE",
                f"/api/collections/{collection}/records/{record['id']}",
                token=cls.admin_token,
            )
            if status != 204:
                raise RuntimeError(f"record delete failed ({collection}, {status})")

    @classmethod
    def _reset_auth_state(cls):
        cls._clear("otp_codes")
        cls._clear("otp_identity_grants")
        with cls.evolution.calls_lock:
            cls.evolution.calls.clear()
        cls.evolution.call_event.clear()
        cls.evolution.response_delay = 0
        cls.evolution.response_status = 200

    @classmethod
    def _request_otp(cls, phone):
        started = time.monotonic()
        result = cls.request("POST", "/api/kq/auth/request-otp", {"phone": phone}, timeout=5)
        return result, time.monotonic() - started

    def test_auth_and_family_critical_invariants(self):
        self._reset_auth_state()
        unknown = "5492917999999"
        first, unknown_elapsed = self._request_otp(unknown)
        second, _ = self._request_otp(unknown)
        self.assertEqual(first, (200, {"ok": True}))
        self.assertEqual(second, (200, {"ok": True}))
        unknown_records = self._list("otp_codes")
        self.assertEqual(len(unknown_records), 1)
        self.assertFalse(unknown_records[0]["eligible"])
        self.assertEqual(len(self.evolution.calls), 0)

        # The unknown reservation throttles the same IP before identity lookup.
        blocked_known, _ = self._request_otp(self.teacher_phone)
        self.assertEqual(blocked_known, (200, {"ok": True}))
        self.assertEqual(len(self.evolution.calls), 0)

        self._reset_auth_state()
        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
            results = list(pool.map(lambda _: self._request_otp(self.teacher_phone)[0], range(8)))
        self.assertTrue(all(result == (200, {"ok": True}) for result in results))
        self.assertTrue(self.evolution.call_event.wait(5))
        deadline = time.time() + 5
        while time.time() < deadline and len(self.evolution.calls) < 1:
            time.sleep(0.05)
        self.assertEqual(len(self.evolution.calls), 1)
        eligible_codes = [record for record in self._list("otp_codes") if record["eligible"]]
        self.assertEqual(len(eligible_codes), 1)
        code = re.search(r"\b(\d{6})\b", self.evolution.calls[0]["text"]).group(1)

        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
            verified = list(
                pool.map(
                    lambda _: self.request(
                        "POST",
                        "/api/kq/auth/verify-otp",
                        {"phone": self.teacher_phone, "code": code},
                    ),
                    range(8),
                )
            )
        self.assertEqual(sum(status == 200 for status, _ in verified), 1)
        self.assertEqual(len(self._list("otp_identity_grants")), 1)
        self.assertTrue(self._list("otp_codes")[0]["consumed"])
        replay, _ = self.request(
            "POST", "/api/kq/auth/verify-otp", {"phone": self.teacher_phone, "code": code}
        )
        self.assertEqual(replay, 400)
        self.assertEqual(len(self._list("otp_identity_grants")), 1)

        # Evolution latency/failure never changes the response BODY (always a generic
        # 200) and the response stays bounded by the send timeout. The eligible path
        # waits for the synchronous WhatsApp send, so it is slower than the unknown
        # path: delivery is prioritised over timing uniformity for this low-stakes login.
        self._reset_auth_state()
        self.evolution.response_delay = 2
        self.evolution.response_status = 500
        failed_send, known_elapsed = self._request_otp(self.teacher_phone)
        self.assertEqual(failed_send, (200, {"ok": True}))
        self.assertLess(known_elapsed, 16)
        self.assertTrue(self.evolution.call_event.wait(5))
        self.assertEqual(len([r for r in self._list("otp_codes") if r["eligible"]]), 1)

        # A correct code rolls back consumption when grant creation fails.
        deadline = time.time() + 5
        while time.time() < deadline and not self.evolution.calls:
            time.sleep(0.05)
        rollback_code = re.search(r"\b(\d{6})\b", self.evolution.calls[0]["text"]).group(1)
        grants = build_schema()["collections"]["otp_identity_grants"]
        grant_collection = next(
            collection
            for collection in self.request("GET", "/api/collections?perPage=200", token=self.admin_token)[1]["items"]
            if collection["name"] == "otp_identity_grants"
        )
        self.assertEqual(
            self.request("DELETE", f"/api/collections/{grant_collection['id']}", token=self.admin_token)[0],
            204,
        )
        rollback_status, _ = self.request(
            "POST",
            "/api/kq/auth/verify-otp",
            {"phone": self.teacher_phone, "code": rollback_code},
        )
        self.assertEqual(rollback_status, 400)
        self.assertFalse(self._list("otp_codes")[0]["consumed"])
        self._create_collection({"name": "otp_identity_grants", **grants})

        # Explicit guardian_id cannot steal a phone already owned by another guardian.
        original_student_phone = self.student_profile_a["phone"]
        conflict_status, _ = self.request(
            "POST",
            "/api/kq/family/save",
            {
                "student_id": self.student_a["id"],
                "is_minor": True,
                "student_phone": "",
                "guardian_id": self.guardian_a["id"],
                "guardian_name": "Guardian A",
                "guardian_phone": self.guardian_b["phone"],
                "replace_primary": True,
            },
            self.teacher_token,
        )
        self.assertEqual(conflict_status, 400)
        guardian_a = self.request(
            "GET", f"/api/collections/users/records/{self.guardian_a['id']}", token=self.admin_token
        )[1]
        student_profile_a = self.request(
            "GET",
            f"/api/collections/student_profiles/records/{self.student_profile_a['id']}",
            token=self.admin_token,
        )[1]
        self.assertEqual(guardian_a["phone"], self.guardian_a["phone"])
        self.assertEqual(student_profile_a["phone"], original_student_phone)

        # Concurrent claims for a new guardian phone result in a single owner.
        shared_phone = "5492918100099"
        payloads = [
            {
                "student_id": self.student_a["id"],
                "is_minor": True,
                "student_phone": "",
                "guardian_id": self.guardian_a["id"],
                "guardian_name": "Guardian A",
                "guardian_phone": shared_phone,
                "replace_primary": True,
            },
            {
                "student_id": self.student_b["id"],
                "is_minor": True,
                "student_phone": "",
                "guardian_id": self.guardian_b["id"],
                "guardian_name": "Guardian B",
                "guardian_phone": shared_phone,
                "replace_primary": True,
            },
        ]
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            family_results = list(
                pool.map(
                    lambda payload: self.request(
                        "POST", "/api/kq/family/save", payload, self.teacher_token
                    ),
                    payloads,
                )
            )
        self.assertEqual(
            sum(status == 200 for status, _ in family_results),
            1,
            msg=[(status, body.get("message", "")) for status, body in family_results],
        )
        self.assertEqual(
            sum(record["phone"] == shared_phone for record in self._list("guardian_profiles")),
            1,
        )


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