#!/usr/bin/env python3
"""Run the first-slice contract against the real local Go API and PostgreSQL."""

from __future__ import annotations

import argparse
import http.cookiejar
import json
import time
import urllib.error
import urllib.request
import uuid


class Client:
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
        )

    def request(
        self,
        method: str,
        path: str,
        body: dict | None = None,
        *,
        key: str | None = None,
        expected: int = 200,
    ) -> dict:
        data = json.dumps(body).encode() if body is not None else None
        headers = {"Accept": "application/json"}
        if data is not None:
            headers["Content-Type"] = "application/json"
        if key:
            headers["Idempotency-Key"] = key
        request = urllib.request.Request(
            self.base_url + path, data=data, headers=headers, method=method
        )
        try:
            response = self.opener.open(request, timeout=10)
            status = response.status
            payload = json.loads(response.read() or b"{}")
        except urllib.error.HTTPError as error:
            status = error.code
            payload = json.loads(error.read() or b"{}")
        if status != expected:
            raise AssertionError(
                f"{method} {path}: expected {expected}, got {status}: {payload}"
            )
        return payload


def error_code(payload: dict) -> str | None:
    error = payload.get("error")
    return error.get("code") if isinstance(error, dict) else None


def mapping_keys(value: object) -> set[str]:
    """Collect every JSON object key recursively for privacy-boundary checks."""
    if isinstance(value, dict):
        keys = set(value)
        for child in value.values():
            keys.update(mapping_keys(child))
        return keys
    if isinstance(value, list):
        keys: set[str] = set()
        for child in value:
            keys.update(mapping_keys(child))
        return keys
    return set()


def ticket_with_number(
    tickets: object,
    ticket_number: str,
    source: str,
    project_id: str | None = None,
) -> dict:
    assert isinstance(tickets, list), f"{source} must be a ticket list"
    matches = [
        ticket
        for ticket in tickets
        if isinstance(ticket, dict)
        and ticket.get("ticket_number") == ticket_number
        and (project_id is None or ticket.get("project_id") == project_id)
    ]
    assert len(matches) == 1, (
        f"{source} must contain exactly one ticket {ticket_number}, got {len(matches)}"
    )
    return matches[0]


def assert_personal_continuity(
    ticket: object,
    *,
    phone: str,
    last_name: str,
    honorific: str,
    source: str,
) -> dict:
    assert isinstance(ticket, dict), f"{source} must be a ticket object"
    assert ticket.get("phone") == phone, f"{source} changed the ticket phone"
    assert ticket.get("last_name") == last_name, (
        f"{source} changed the ticket last name"
    )
    assert ticket.get("honorific") == honorific, (
        f"{source} changed the ticket honorific"
    )
    return ticket


def assert_public_phone_projection(payload: dict, phone: str, source: str) -> None:
    assert payload.get("phone_last4") == phone[-4:]
    phone_keys = {
        key for key in mapping_keys(payload) if "phone" in key.lower()
    }
    assert phone_keys == {"phone_last4"}, (
        f"{source} may expose only phone_last4, got {phone_keys}"
    )
    assert phone not in json.dumps(payload, ensure_ascii=False), (
        f"{source} leaked the complete phone"
    )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", default="http://127.0.0.1:8080")
    parser.add_argument("--username", default="staff")
    parser.add_argument("--admin-username", default="admin")
    parser.add_argument("--password", default="ChangeMe123!")
    parser.add_argument("--display-token", required=True)
    args = parser.parse_args()

    client = Client(args.base_url)
    health = client.request("GET", "/healthz")
    assert health["status"] == "ok"

    client.request(
        "POST",
        "/api/staff/auth/login",
        {"username": args.username, "password": args.password},
    )
    me = client.request("GET", "/api/staff/auth/me")
    project_id = me["projects"][0]["id"]

    snapshot = client.request("GET", f"/api/staff/projects/{project_id}/queue")
    assert snapshot.get("current_batch") is None, (
        "smoke requires a project with no active call batch; finish or miss the "
        "current batch before running it"
    )
    assert snapshot.get("waiting") == [], (
        "smoke requires a freshly rebuilt, dedicated database with no waiting "
        "tickets; rebuild and seed the smoke database before running it"
    )
    phone = "139" + str(int(time.time() * 1000))[-8:]
    last_name = "烟测"
    honorific = "游客"
    ticket_body = {
        "phone": phone,
        "last_name": last_name,
        "honorific": honorific,
        "allow_duplicate": False,
    }
    first_key = str(uuid.uuid4())
    first = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/tickets",
        ticket_body,
        key=first_key,
        expected=201,
    )
    first_ticket = first["ticket"]
    assert first_ticket["ticket_number"]
    assert first_ticket["public_token"]
    assert_personal_continuity(
        first_ticket,
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="create response",
    )

    replay = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/tickets",
        ticket_body,
        key=first_key,
        expected=201,
    )
    assert replay["ticket"]["id"] == first_ticket["id"]
    assert_personal_continuity(
        replay["ticket"],
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="create idempotency replay",
    )

    # Staff and admin are authenticated operational surfaces, so they must
    # expose the complete phone beside the exact ticket number.
    staff_snapshot = client.request(
        "GET", f"/api/staff/projects/{project_id}/queue"
    )
    staff_ticket = ticket_with_number(
        staff_snapshot["waiting"], first_ticket["ticket_number"], "staff queue"
    )
    assert_personal_continuity(
        staff_ticket,
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="staff waiting queue",
    )

    public = client.request(
        "GET", f"/api/public/status/{first_ticket['public_token']}"
    )
    assert public["ticket_number"] == first_ticket["ticket_number"]
    assert_public_phone_projection(public, phone, "public status")

    phone_lookup = client.request(
        "POST", "/api/public/status/search", {"phone": phone}
    )
    assert len(phone_lookup.get("tickets", [])) == 1
    phone_lookup_ticket = phone_lookup["tickets"][0]
    assert phone_lookup_ticket["ticket_number"] == first_ticket["ticket_number"]
    assert_public_phone_projection(phone_lookup_ticket, phone, "phone lookup")

    client.request(
        "POST",
        "/api/admin/auth/login",
        {"username": args.admin_username, "password": args.password},
    )
    overview = client.request("GET", "/api/admin/overview")
    admin_ticket = ticket_with_number(
        overview.get("active_tickets"),
        first_ticket["ticket_number"],
        "admin active_tickets",
        project_id,
    )
    assert_personal_continuity(
        admin_ticket,
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="admin active_tickets",
    )

    duplicate = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/tickets",
        ticket_body,
        key=str(uuid.uuid4()),
        expected=409,
    )
    assert error_code(duplicate) == "DUPLICATE_PHONE"

    confirmed_body = {**ticket_body, "allow_duplicate": True}
    confirmed = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/tickets",
        confirmed_body,
        key=str(uuid.uuid4()),
        expected=201,
    )
    confirmed_ticket = confirmed["ticket"]
    assert confirmed_ticket["id"] != first_ticket["id"]
    assert_personal_continuity(
        confirmed_ticket,
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="confirmed duplicate create response",
    )

    phone_lookup = client.request(
        "POST", "/api/public/status/search", {"phone": phone}
    )
    phone_lookup_numbers = {
        ticket["ticket_number"] for ticket in phone_lookup.get("tickets", [])
    }
    assert phone_lookup_numbers == {
        first_ticket["ticket_number"], confirmed_ticket["ticket_number"]
    }
    for ticket in phone_lookup["tickets"]:
        assert_public_phone_projection(ticket, phone, "phone lookup with duplicate tickets")

    snapshot = client.request("GET", f"/api/staff/projects/{project_id}/queue")
    project = snapshot["project"]
    batch_size = int(
        project.get("call_batch_size") or project.get("batch_size") or 0
    )
    assert batch_size > 0, "project call batch size must be positive"
    created_tickets = [first_ticket, confirmed_ticket]
    while len(created_tickets) < batch_size + 1:
        extra = client.request(
            "POST",
            f"/api/staff/projects/{project_id}/tickets",
            confirmed_body,
            key=str(uuid.uuid4()),
            expected=201,
        )
        created_tickets.append(extra["ticket"])
    snapshot = client.request("GET", f"/api/staff/projects/{project_id}/queue")
    waiting_before_call = snapshot["waiting"]
    waiting_ids = {ticket["id"] for ticket in waiting_before_call}
    assert waiting_ids == {ticket["id"] for ticket in created_tickets}, (
        "dedicated smoke queue contains unexpected tickets"
    )
    for ticket in waiting_before_call:
        assert_personal_continuity(
            ticket,
            phone=phone,
            last_name=last_name,
            honorific=honorific,
            source="staff waiting queue before call-next",
        )
    stale_call = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/call-next",
        {"expected_revision": max(0, int(snapshot["revision"]) - 1), "count": batch_size},
        key=str(uuid.uuid4()),
        expected=409,
    )
    assert error_code(stale_call) == "REVISION_CONFLICT"

    expected_called_ids = [
        ticket["id"] for ticket in waiting_before_call[:batch_size]
    ]
    call_key = str(uuid.uuid4())
    call_body = {"expected_revision": int(snapshot["revision"]), "count": batch_size}
    called = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/call-next",
        call_body,
        key=call_key,
    )
    called_tickets = called["batch"]["tickets"]
    assert [ticket["id"] for ticket in called_tickets] == expected_called_ids
    assert len(called_tickets) == min(batch_size, len(waiting_before_call))
    assert called_tickets, "call-next must call at least one waiting ticket"
    for ticket in called_tickets:
        assert_personal_continuity(
            ticket,
            phone=phone,
            last_name=last_name,
            honorific=honorific,
            source="call-next response",
        )
    assert called["device_results"][0]["status"] == "SUCCESS"

    call_replay = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/call-next",
        call_body,
        key=call_key,
    )
    assert call_replay["batch"]["id"] == called["batch"]["id"]
    for ticket in call_replay["batch"]["tickets"]:
        assert_personal_continuity(
            ticket,
            phone=phone,
            last_name=last_name,
            honorific=honorific,
            source="call-next idempotency replay",
        )

    active_snapshot = client.request(
        "GET", f"/api/staff/projects/{project_id}/queue"
    )
    current_batch = active_snapshot.get("current_batch")
    assert isinstance(current_batch, dict)
    assert current_batch["id"] == called["batch"]["id"]
    assert [ticket["id"] for ticket in current_batch["tickets"]] == (
        expected_called_ids
    )
    for ticket in current_batch["tickets"]:
        assert_personal_continuity(
            ticket,
            phone=phone,
            last_name=last_name,
            honorific=honorific,
            source="staff current_batch",
        )

    second_called = client.request(
        "POST",
        f"/api/staff/projects/{project_id}/call-next",
        {"expected_revision": int(called["revision"]), "count": 1},
        key=str(uuid.uuid4()),
    )
    assert second_called["batch"]["id"] != called["batch"]["id"]
    assert len(second_called["batch"]["tickets"]) == 1
    final_staff_snapshot = client.request(
        "GET", f"/api/staff/projects/{project_id}/queue"
    )
    assert final_staff_snapshot["current_batch"]["id"] == second_called["batch"]["id"]
    assert final_staff_snapshot["metrics"]["waiting_count"] == 0
    completed_public = client.request(
        "GET", f"/api/public/status/{first_ticket['public_token']}"
    )
    assert completed_public["status"] == "COMPLETED"
    assert_public_phone_projection(completed_public, phone, "auto-completed public status")
    display = client.request(
        "GET", f"/api/display/{args.display_token}/snapshot"
    )
    serialized_display = json.dumps(display, ensure_ascii=False)
    display_keys = mapping_keys(display)
    assert "honorific" not in display_keys
    display_phone_keys = {
        key for key in display_keys if "phone" in key.lower()
    }
    forbidden_display_phone_keys = {"phone", "phone_last4", "phone_masked"}
    assert display_phone_keys.isdisjoint(forbidden_display_phone_keys)
    assert not display_phone_keys, (
        f"display leaked phone fields: {sorted(display_phone_keys)}"
    )
    assert phone not in serialized_display
    current_numbers = {
        ticket["ticket_number"]
        for ticket in (display.get("current_batch") or {}).get("tickets", [])
    }
    assert first_ticket["ticket_number"] not in current_numbers

    overview = client.request("GET", "/api/admin/overview")
    project = next(item for item in overview["projects"] if item["id"] == project_id)
    assert project["waiting_count"] == 0
    current_admin_ticket = ticket_with_number(
        overview.get("active_tickets"),
        second_called["batch"]["tickets"][0]["ticket_number"],
        "admin active_tickets after automatic rotation",
        project_id,
    )
    assert_personal_continuity(
        current_admin_ticket,
        phone=phone,
        last_name=last_name,
        honorific=honorific,
        source="admin active_tickets after automatic rotation",
    )

    print(
        json.dumps(
            {
                "status": "ok",
                "project_id": project_id,
                "first_ticket": first_ticket["ticket_number"],
                "first_batch_id": called["batch"]["id"],
                "second_batch_id": second_called["batch"]["id"],
            },
            ensure_ascii=False,
            indent=2,
        )
    )


if __name__ == "__main__":
    main()