"""Strict Native Messaging wire validation.

This module is stdlib-only and contains no downloader imports.
"""

from __future__ import annotations

import json
import re
import struct
import time
import unicodedata
from typing import Any, BinaryIO, Iterable

from .constants import (
    CANONICAL_URL,
    DURATION_TOLERANCE_MS,
    EXPECTED_DURATION_MS,
    EXTENSION_BUILD,
    HOST_BUILD,
    MAX_COOKIE_COUNT,
    MAX_INPUT_FRAME,
    MAX_OUTPUT_FRAME,
    MAX_SAFE_INTEGER,
    SCHEMA_VERSION,
    TARGET_BVID,
    TARGET_PATH,
)

_NONCE_RE = re.compile(r"[0-9a-f]{32}\Z")
_PREPARE_ID_RE = re.compile(r"[0-9a-f]{32}\Z")
_STORE_RE = re.compile(r"[0-9]{1,8}\Z")
_PARENT_RE = re.compile(r"--parent-window=[0-9]+\Z")
_COOKIE_SAME_SITE = {"no_restriction", "lax", "strict", "unspecified"}


class ProtocolError(Exception):
    """A fixed-code input rejection that is safe to expose."""

    def __init__(self, code: str = "E_PROTOCOL") -> None:
        super().__init__(code)
        self.code = code


def _reject_constant(_: str) -> None:
    raise ProtocolError()


def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ProtocolError()
        result[key] = value
    return result


def strict_json_loads(payload: bytes) -> dict[str, Any]:
    try:
        text = payload.decode("utf-8", errors="strict")
        value = json.loads(
            text,
            object_pairs_hook=_unique_object,
            parse_constant=_reject_constant,
        )
    except ProtocolError:
        raise
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise ProtocolError() from exc
    if not isinstance(value, dict):
        raise ProtocolError()
    return value


def encode_json(value: dict[str, Any], limit: int = MAX_OUTPUT_FRAME) -> bytes:
    payload = json.dumps(
        value,
        ensure_ascii=True,
        allow_nan=False,
        separators=(",", ":"),
        sort_keys=True,
    ).encode("utf-8")
    if len(payload) > limit:
        raise ProtocolError()
    return payload


def read_frame(stream: BinaryIO, limit: int = MAX_INPUT_FRAME) -> bytes | None:
    header = stream.read(4)
    if header == b"":
        return None
    if len(header) != 4:
        raise ProtocolError()
    (length,) = struct.unpack("<I", header)
    if length == 0 or length > limit:
        raise ProtocolError()
    payload = stream.read(length)
    if len(payload) != length:
        raise ProtocolError()
    return payload


def write_frame(stream: BinaryIO, value: dict[str, Any]) -> None:
    payload = encode_json(value)
    stream.write(struct.pack("<I", len(payload)))
    stream.write(payload)
    stream.flush()


def _exact_keys(value: dict[str, Any], expected: Iterable[str]) -> None:
    if set(value) != set(expected):
        raise ProtocolError()


def _integer(value: Any, minimum: int, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise ProtocolError()
    if not minimum <= value <= maximum:
        raise ProtocolError()
    return value


def _boolean(value: Any) -> bool:
    if not isinstance(value, bool):
        raise ProtocolError()
    return value


def _safe_string(value: Any, minimum: int, maximum_utf8: int) -> str:
    if not isinstance(value, str):
        raise ProtocolError()
    encoded = value.encode("utf-8")
    if len(encoded) < minimum or len(encoded) > maximum_utf8:
        raise ProtocolError()
    if any(ch in "\x00\t\r\n" or unicodedata.category(ch).startswith("C") for ch in value):
        raise ProtocolError()
    return value


def validate_origin_argv(arguments: list[str], expected_origin: str) -> None:
    if len(arguments) not in (1, 2) or arguments[0] != expected_origin:
        raise ProtocolError("E_ORIGIN")
    if len(arguments) == 2 and not _PARENT_RE.fullmatch(arguments[1]):
        raise ProtocolError("E_ORIGIN")


def _validate_common(value: dict[str, Any], message_type: str, keys: Iterable[str]) -> None:
    _exact_keys(value, keys)
    if _integer(value.get("schema"), SCHEMA_VERSION, SCHEMA_VERSION) != SCHEMA_VERSION:
        raise ProtocolError()
    if value.get("type") != message_type or value.get("target") != TARGET_BVID:
        raise ProtocolError()


def validate_hello(value: dict[str, Any]) -> dict[str, Any]:
    _validate_common(value, "hello", {"schema", "type", "extension_build", "target"})
    if value["extension_build"] != EXTENSION_BUILD:
        raise ProtocolError("E_BUILD")
    return value


def validate_status(value: dict[str, Any]) -> dict[str, Any]:
    _validate_common(value, "status", {"schema", "type", "target"})
    return value


def validate_cancel(value: dict[str, Any]) -> dict[str, Any]:
    _validate_common(value, "cancel", {"schema", "type", "target", "task_nonce"})
    if not isinstance(value["task_nonce"], str) or not _NONCE_RE.fullmatch(value["task_nonce"]):
        raise ProtocolError()
    return value


def validate_prepare(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
    _validate_common(
        value,
        "prepare",
        {"schema", "type", "extension_build", "target", "prepare_id", "page_proof"},
    )
    if value["extension_build"] != EXTENSION_BUILD:
        raise ProtocolError("E_BUILD")
    if not isinstance(value["prepare_id"], str) or not _PREPARE_ID_RE.fullmatch(value["prepare_id"]):
        raise ProtocolError("E_PREPARE")
    validate_page_proof(value["page_proof"], now_ms=now_ms)
    return value


def validate_page_proof(value: Any, now_ms: int | None = None) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ProtocolError()
    _exact_keys(
        value,
        {
            "target",
            "canonical_url",
            "task_nonce",
            "observed_at_unix_ms",
            "observed_duration_ms",
            "video_width",
            "video_height",
            "ready_state",
            "eme_present",
        },
    )
    if value["target"] != TARGET_BVID or value["canonical_url"] != CANONICAL_URL:
        raise ProtocolError()
    if not isinstance(value["task_nonce"], str) or not _NONCE_RE.fullmatch(value["task_nonce"]):
        raise ProtocolError()
    observed_at = _integer(value["observed_at_unix_ms"], 1, MAX_SAFE_INTEGER)
    observed_duration = _integer(value["observed_duration_ms"], 1, MAX_SAFE_INTEGER)
    _integer(value["video_width"], 1, 7680)
    _integer(value["video_height"], 1, 4320)
    _integer(value["ready_state"], 1, 4)
    if _boolean(value["eme_present"]):
        raise ProtocolError("E_DRM")
    if abs(observed_duration - EXPECTED_DURATION_MS) > DURATION_TOLERANCE_MS:
        raise ProtocolError("E_DURATION")
    current = int(time.time() * 1000) if now_ms is None else now_ms
    if observed_at < current - 60_000 or observed_at > current + 5_000:
        raise ProtocolError("E_PAGE_PROOF")
    return value


def _cookie_path_matches(path: str) -> bool:
    if path == "/":
        return True
    if not TARGET_PATH.startswith(path):
        return False
    return path.endswith("/") or len(path) == len(TARGET_PATH) or TARGET_PATH[len(path)] == "/"


def validate_cookie(
    value: Any,
    store_id: str,
    observed_at_unix_ms: int,
) -> dict[str, Any]:
    if not isinstance(value, dict):
        raise ProtocolError("E_SECRET_INPUT")
    try:
        _exact_keys(
            value,
            {
                "name",
                "value",
                "domain",
                "host_only",
                "path",
                "secure",
                "http_only",
                "same_site",
                "session",
                "expiration_unix",
                "store_id",
                "partition_key",
            },
        )
        _safe_string(value["name"], 1, 256)
        _safe_string(value["value"], 1, 4096)
        host_only = _boolean(value["host_only"])
        expected_domain = "www.bilibili.com" if host_only else ".bilibili.com"
        if value["domain"] != expected_domain:
            raise ProtocolError()
        path = _safe_string(value["path"], 1, 1024)
        if not path.startswith("/") or not _cookie_path_matches(path):
            raise ProtocolError()
        _boolean(value["secure"])
        _boolean(value["http_only"])
        session = _boolean(value["session"])
        if value["same_site"] not in _COOKIE_SAME_SITE:
            raise ProtocolError()
        if value["store_id"] != store_id or value["partition_key"] is not None:
            raise ProtocolError()
        if session:
            if value["expiration_unix"] is not None:
                raise ProtocolError()
        else:
            expires = _integer(value["expiration_unix"], 1, MAX_SAFE_INTEGER)
            if expires <= observed_at_unix_ms // 1000:
                raise ProtocolError()
    except ProtocolError as exc:
        raise ProtocolError("E_SECRET_INPUT") from exc
    return value


def validate_start(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
    _validate_common(
        value,
        "start",
        {
            "schema",
            "type",
            "extension_build",
            "target",
            "canonical_url",
            "cookie_store_id",
            "prepare_id",
            "page_proof",
            "cookies",
        },
    )
    if value["extension_build"] != EXTENSION_BUILD:
        raise ProtocolError("E_BUILD")
    if value["canonical_url"] != CANONICAL_URL:
        raise ProtocolError()
    if not isinstance(value["prepare_id"], str) or not _PREPARE_ID_RE.fullmatch(value["prepare_id"]):
        raise ProtocolError("E_PREPARE")
    store_id = value["cookie_store_id"]
    if not isinstance(store_id, str) or not _STORE_RE.fullmatch(store_id):
        raise ProtocolError("E_SECRET_INPUT")
    proof = validate_page_proof(value["page_proof"], now_ms=now_ms)
    cookies = value["cookies"]
    if not isinstance(cookies, list) or not 1 <= len(cookies) <= MAX_COOKIE_COUNT:
        raise ProtocolError("E_SECRET_INPUT")
    for cookie in cookies:
        validate_cookie(cookie, store_id, proof["observed_at_unix_ms"])
    return value


def validate_message(value: dict[str, Any], now_ms: int | None = None) -> dict[str, Any]:
    message_type = value.get("type")
    if message_type == "hello":
        return validate_hello(value)
    if message_type == "status":
        return validate_status(value)
    if message_type == "cancel":
        return validate_cancel(value)
    if message_type == "prepare":
        return validate_prepare(value, now_ms=now_ms)
    if message_type == "start":
        return validate_start(value, now_ms=now_ms)
    raise ProtocolError()


def safe_response(
    message_type: str,
    phase: str,
    *,
    progress: int = 0,
    error_code: str | None = None,
    formal_filename: str | None = None,
    mapping_filename: str | None = None,
    prepare_id: str | None = None,
) -> dict[str, Any]:
    return {
        "schema": SCHEMA_VERSION,
        "type": message_type,
        "host_build": HOST_BUILD,
        "target": TARGET_BVID,
        "phase": phase,
        "progress": max(0, min(100, int(progress))),
        "error_code": error_code,
        "formal_filename": formal_filename,
        "mapping_filename": mapping_filename,
        "prepare_id": prepare_id,
    }