"""Explicit HTTP operations with durable recovery and read acknowledgement."""

import base64
import copy
import json
import math
import re
import secrets
import time
from html.parser import HTMLParser
from urllib.parse import urlsplit

import httpx

from .errors import (AmbiguousWrite, ApiError, FarsideError, PaymentPolicyError,
                     RetryLater, StateError, TransportError)
from .state import StateStore

DEFAULT_ORIGIN = "https://farside.science"
DEFAULT_RECEIVER = "0x7520faE4b7Bb289BCed657BA51b427B0D9a6ed87"
USDC_BASE = "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913"
PROTOCOL = {"X-Farside-Protocol": "preview-1"}
DEFAULT_LIMITS = {"poll_interval_seconds": 60, "read_interval_seconds": 1,
                  "write_interval_seconds": 2, "max_page_size": 50,
                  "max_acknowledgements": 50}
MAX_INTEGER = 2**63 - 1


class _ServiceData(HTMLParser):
    """Extract one inert JSON document without executing or fetching HTML."""

    def __init__(self):
        super().__init__(convert_charrefs=False)
        self.count = 0
        self.parts = []
        self.reading = False
        self.complete = False
        self.invalid = False

    def handle_starttag(self, tag, attrs):
        if tag != "script":
            return
        values = dict(attrs)
        if values.get("id") == "farside-service":
            self.count += 1
            self.reading = True
            if (values.get("type", "").lower() != "application/json"
                    or "src" in values or len(values) != len(attrs)):
                self.invalid = True

    def handle_data(self, data):
        if self.reading:
            self.parts.append(data)

    def handle_endtag(self, tag):
        if tag == "script" and self.reading:
            self.reading = False
            self.complete = True


def _integer(value, minimum=1, maximum=MAX_INTEGER):
    if type(value) is not int or not minimum <= value <= maximum:
        raise FarsideError("Invalid integer parameter")
    return value


def _text(value, maximum=2000):
    if not isinstance(value, str) or not value.strip() or len(value) > maximum:
        raise FarsideError("Text must be nonblank and within the character limit")
    try:
        value.encode("utf-8")
    except UnicodeError:
        raise FarsideError("Text must contain valid Unicode") from None
    return value


def _encode(value):
    try:
        encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    except (ValueError, UnicodeError, TypeError):
        raise FarsideError("Invalid JSON data") from None
    if len(encoded) > 32768:
        raise FarsideError("JSON body exceeds the service limit")
    return encoded


class FarsideClient:
    def __init__(self, state_dir, origin=DEFAULT_ORIGIN, http_client=None,
                 timeout=100, clock=time.time):
        try:
            parts = urlsplit(origin)
            if (parts.scheme != "https" or not parts.hostname or parts.username is not None
                    or parts.password is not None or parts.path not in ("", "/")
                    or parts.query or parts.fragment):
                raise ValueError
            parts.port  # Reject malformed port values.
            self.origin = origin.rstrip("/")
        except (ValueError, TypeError):
            raise FarsideError("Origin must be a plain HTTPS origin") from None
        self.store = StateStore(state_dir)
        self.http = http_client if http_client is not None else httpx.Client(follow_redirects=False)
        self._owns_http = http_client is None
        self.timeout = timeout
        self.clock = clock
        self._service = None
        self._terms = None
        self._limits = dict(DEFAULT_LIMITS)

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.shutdown()

    def shutdown(self):
        if self._owns_http:
            self.http.close()

    def _json(self, response):
        try:
            if len(response.content) > 2_000_000:
                raise ValueError
            value = response.json()
            if not isinstance(value, dict):
                raise ValueError
            return value
        except (ValueError, UnicodeError):
            raise ApiError("invalid_response", response.status_code) from None

    def _error(self, response):
        try:
            code = response.json().get("error")
        except (ValueError, AttributeError):
            code = None
        if not isinstance(code, str) or not re.fullmatch(r"[a-z][a-z0-9_]{0,80}", code):
            code = "unexpected_response"
        raise ApiError(code, response.status_code)

    def _load(self, handle):
        state = self.store.load(handle)
        if state.get("origin") != self.origin:
            raise StateError("Local handle belongs to a different service origin")
        if (not re.fullmatch(r"[0-9a-f]{64}", str(state.get("secret", "")))
                or not isinstance(state.get("body"), str)
                or not isinstance(state.get("next_allowed"), dict)
                or not isinstance(state.get("limits"), dict)):
            raise StateError("Invalid local state")
        return state

    def _limits_from(self, value):
        if not isinstance(value, dict):
            raise ApiError("invalid_limits", 200)
        result = dict(self._limits)
        for key in DEFAULT_LIMITS:
            if key in value:
                item = value[key]
                if type(item) is not int or item < 0 or item > MAX_INTEGER:
                    raise ApiError("invalid_limits", 200)
                result[key] = item
        # The client never checks for updates more often than the documented minute.
        result["poll_interval_seconds"] = max(60, result["poll_interval_seconds"])
        return result

    def _request(self, method, path, *, handle=None, state=None, bucket=None,
                 body=None, headers=None, params=None, ambiguous=False):
        if bucket:
            now = self.clock()
            remaining = state["next_allowed"].get(bucket, 0) - now
            if remaining > 0:
                raise RetryLater(math.ceil(remaining))
            interval = state["limits"].get(bucket + "_interval_seconds", 0)
            state["next_allowed"][bucket] = now + interval
            self.store.save(handle, state)
        combined = {**PROTOCOL, **(headers or {})}
        if body is not None:
            combined["Content-Type"] = "application/json"
        try:
            response = self.http.request(method, self.origin + path, headers=combined,
                content=body, params=params, timeout=self.timeout, follow_redirects=False)
        except httpx.HTTPError:
            if ambiguous:
                raise AmbiguousWrite("Write outcome is unknown; reconcile before another write") from None
            raise TransportError("Request outcome is unknown; recover with the same local handle") from None
        if 300 <= response.status_code < 400:
            if ambiguous:
                raise AmbiguousWrite("Redirect refused; reconcile the write outcome")
            raise ApiError("redirect_refused", response.status_code)
        if response.status_code == 429:
            try:
                delay = max(1, int(response.headers.get("Retry-After", "10")))
            except ValueError:
                delay = 10
            if state is not None and bucket:
                now = self.clock()
                deadline = max(state["next_allowed"].get(bucket, 0), now + delay)
                state["next_allowed"][bucket] = deadline
                self.store.save(handle, state)
                delay = math.ceil(deadline - now)
            raise RetryLater(delay)
        return response

    def _read_service(self):
        self._service = None
        self._terms = None
        response = self._request("GET", "/docs")
        if response.status_code != 200:
            self._error(response)
        try:
            if (len(response.content) > 2_000_000
                    or response.headers.get("Content-Type", "").split(";", 1)[0].lower() != "text/html"):
                raise ValueError
            parser = _ServiceData()
            parser.feed(response.text)
            parser.close()
            if parser.count != 1 or not parser.complete or parser.reading or parser.invalid:
                raise ValueError
            service = json.loads("".join(parser.parts))
            if (not isinstance(service, dict) or service.get("document") != "farside-service"
                    or service.get("version") != "preview-1"
                    or not isinstance(service.get("published_terms"), dict)
                    or not isinstance(service.get("api"), dict)):
                raise ValueError
            published = service["published_terms"]
            version = published.get("version")
            if (type(version) is not int or not 1 <= version <= MAX_INTEGER
                    or response.headers.get("X-Farside-Terms-Version") != str(version)
                    or published.get("version_header") != "X-Farside-Terms-Version"):
                raise ValueError
        except (ValueError, UnicodeError, TypeError):
            raise ApiError("invalid_response", response.status_code) from None
        if published.get("url") != self.origin + "/docs":
            raise ApiError("invalid_terms_location", 200)
        self._limits = self._limits_from(service["api"].get("limits", {}))
        self._service = service
        self._terms = {"html": response.text, "version": version}

    def service(self):
        """Refresh current service data and the matching terms from one /docs GET."""
        self._read_service()
        return copy.deepcopy(self._service)

    def terms(self):
        """Read terms and their version, sharing the last service() response."""
        if self._terms is None:
            self._read_service()
        return dict(self._terms)

    def coupon(self):
        response = self._request("GET", "/agent/coupon")
        if response.status_code != 200:
            self._error(response)
        return self._json(response)

    def prepare(self, subject, text, active_days, terms_version, mode="research", coupon=None):
        _text(subject, 50)
        if re.search(r"[\t\n\r\v\f\u0085\u2028\u2029]", subject):
            raise FarsideError("Subject must be a single line")
        _text(text)
        _integer(active_days)
        _integer(terms_version)
        if active_days not in (3, 7, 30, 180) or mode not in ("research", "private"):
            raise FarsideError("Invalid dialogue lifetime or mode")
        if coupon is not None and (not isinstance(coupon, str) or not re.fullmatch(
                r"(?:[0-9a-f]{32}|owner_[0-9]+_[0-9a-f]{32})", coupon)):
            raise FarsideError("Invalid coupon code")
        body = _encode({"active_days": active_days, "terms_version": terms_version,
                        "mode": mode, "subject": subject, "text": text}).decode("utf-8")
        handle = secrets.token_hex(16)
        state = {"version": 1, "origin": self.origin, "secret": secrets.token_hex(32),
                 "body": body, "coupon": coupon, "cursor": 0, "pending_read": None,
                 "limits": dict(self._limits), "next_allowed": {}, "payment_attempted": False}
        with self.store.locked(handle):
            self.store.save(handle, state)
        return handle

    def _dialog_path(self, state):
        url = state.get("dialogue_url")
        if (not isinstance(url, str) or not url.startswith(self.origin + "/dialog/")
                or not re.fullmatch(r"/dialog/[A-Za-z0-9_-]{64}", url[len(self.origin):])):
            raise StateError("No valid dialogue link is saved for this handle")
        return url[len(self.origin):]

    def _opening(self, handle, state, signature=None):
        headers = {"X-Farside-Purchase": state["secret"]}
        if state.get("coupon") is not None:
            headers["X-Farside-Coupon"] = state["coupon"]
        if signature is not None:
            headers["PAYMENT-SIGNATURE"] = signature
        response = self._request("POST", "/agent/dialogues", handle=handle, state=state,
            bucket="purchase", body=state["body"].encode("utf-8"), headers=headers)
        if response.status_code == 201:
            result = self._json(response)
            candidate = dict(state, dialogue_url=result.get("dialogue_url"))
            self._dialog_path(candidate)
            state.update(dialogue_url=result["dialogue_url"], expires_at=result.get("expires_at"),
                         opening_method=result.get("opening_method", "paid"), status="issued")
            state.pop("payment_signature", None)
            state.pop("payment_offer", None)
            self.store.save(handle, state)
            return {**result, "status": "issued", "handle": handle}
        if response.status_code == 202:
            result = self._json(response)
            if result.get("purchase_status") != "pending":
                raise ApiError("invalid_response", 202)
            try:
                delay = max(60, int(response.headers.get("Retry-After", "60")))
            except ValueError:
                delay = 60
            state.update(status="pending", payment_attempted=True)
            # The server confirms it saved a submitted payment. Recovery is now
            # strictly unsigned; never recreate or replay its authorization.
            state.pop("payment_signature", None)
            state.pop("payment_offer", None)
            state["next_allowed"]["purchase"] = self.clock() + delay
            self.store.save(handle, state)
            return {"status": "pending", "handle": handle, "retry_after": delay}
        if response.status_code == 402:
            if state.get("coupon") is not None:
                raise ApiError("unexpected_payment_challenge", 402)
            offer = self._json(response)
            state["status"] = "payment_required"
            self.store.save(handle, state)
            return {"status": "payment_required", "handle": handle, "offer": offer,
                    "payment_attempted": state["payment_attempted"]}
        self._error(response)

    def open(self, handle):
        """Initial unsigned opening or recovery; never signs or pays automatically."""
        with self.store.locked(handle):
            state = self._load(handle)
            if state.get("status") == "issued":
                self._dialog_path(state)
                return {"status": "issued", "handle": handle,
                        "dialogue_url": state["dialogue_url"], "expires_at": state.get("expires_at"),
                        "opening_method": state.get("opening_method")}
            return self._opening(handle, state)

    recover = open

    def _validate_offer(self, offer, maximum, expected_pay_to):
        _integer(maximum)
        if expected_pay_to is None:
            if self.origin != DEFAULT_ORIGIN:
                raise PaymentPolicyError("Custom origins require an explicit expected recipient")
            expected_pay_to = DEFAULT_RECEIVER
        if not isinstance(expected_pay_to, str) or not re.fullmatch(r"0x[0-9a-fA-F]{40}", expected_pay_to):
            raise PaymentPolicyError("Invalid expected recipient")
        try:
            entries = offer["accepts"]
            if (offer.get("x402Version") != 2 or not isinstance(entries, list) or len(entries) != 1
                    or offer["resource"]["url"] != self.origin + "/agent/dialogues"):
                raise ValueError
            entry = entries[0]
            amount = entry["amount"]
            if (entry["scheme"] != "exact" or entry["network"] != "eip155:8453"
                    or entry["asset"].lower() != USDC_BASE.lower()
                    or entry["payTo"].lower() != expected_pay_to.lower()
                    or not isinstance(amount, str) or not re.fullmatch(r"[0-9]{1,20}", amount)
                    or not 0 < int(amount) <= maximum
                    or type(entry["maxTimeoutSeconds"]) is not int
                    or not 1 <= entry["maxTimeoutSeconds"] <= 300
                    or entry.get("extra", {}).get("assetTransferMethod") != "eip3009"
                    or entry.get("extra", {}).get("name") != "USD Coin"
                    or entry.get("extra", {}).get("version") != "2"):
                raise ValueError
        except (KeyError, ValueError, TypeError, AttributeError):
            raise PaymentPolicyError("Payment offer violates the configured spending policy") from None

    def pay(self, handle, signer, max_amount_atomic=2310, expected_pay_to=None):
        """Explicit one-time signing, after unsigned recovery and offer validation.

        Persist one bounded authorization before HTTP. Explicit pay() can replay
        that exact signature after unsigned recovery returns the same offer;
        it never signs again. Ordinary open()/recover() remain unsigned.
        """
        with self.store.locked(handle):
            state = self._load(handle)
            if state.get("coupon") is not None:
                raise PaymentPolicyError("Coupon attempts cannot be changed to paid attempts")
            if state.get("status") == "issued":
                self._dialog_path(state)
                return {"status": "issued", "handle": handle,
                        "dialogue_url": state["dialogue_url"], "expires_at": state.get("expires_at")}
            if state.get("payment_signature") is not None:
                # A reduced caller budget or changed expected recipient must
                # block reuse before any request, including recovery.
                self._validate_offer(state.get("payment_offer"), max_amount_atomic, expected_pay_to)
            result = self._opening(handle, state)
            if result["status"] != "payment_required":
                return result
            offer = result["offer"]
            self._validate_offer(offer, max_amount_atomic, expected_pay_to)
            if state["payment_attempted"]:
                signature = state.get("payment_signature")
                original = state.get("payment_offer")
                if signature is None or original is None:
                    raise PaymentPolicyError("Payment may have been submitted; reconcile before any new authorization")
                if offer["accepts"] != original.get("accepts") or offer["resource"]["url"] != original.get("resource", {}).get("url"):
                    raise PaymentPolicyError("Payment offer changed; reconcile the saved authorization")
                self._validate_signature(signature, original)
                return self._opening(handle, state, signature)
            try:
                signature = signer(copy.deepcopy(offer))
            except Exception:
                raise PaymentPolicyError("Payment signing failed") from None
            self._validate_signature(signature, offer)
            # The signature is a bounded spending credential, never a wallet key.
            # Persist it and the guard atomically before any signed HTTP request.
            state.update(payment_attempted=True, payment_signature=signature,
                         payment_offer=copy.deepcopy(offer))
            self.store.save(handle, state)
            return self._opening(handle, state, signature)

    def _validate_signature(self, signature, offer):
        """Check one x402 EIP-3009 payload and its remaining validity locally."""
        try:
            if not isinstance(signature, str) or not 1 <= len(signature) <= 12000:
                raise ValueError
            payload = json.loads(base64.b64decode(signature, validate=True))
            accepted = offer["accepts"][0]
            if (payload.get("x402Version") != 2 or payload["accepted"] != accepted
                    or payload["resource"]["url"] != offer["resource"]["url"]
                    or set(payload["payload"]) != {"signature", "authorization"}):
                raise ValueError
            inner = payload["payload"]
            authorization = inner["authorization"]
            if set(authorization) != {"from", "to", "value", "validAfter", "validBefore", "nonce"}:
                raise ValueError
            for name in ("from", "to"):
                if not isinstance(authorization[name], str) or not re.fullmatch(r"0x[0-9a-fA-F]{40}", authorization[name]):
                    raise ValueError
            for name in ("value", "validAfter", "validBefore"):
                if not isinstance(authorization[name], str) or not re.fullmatch(r"[0-9]{1,20}", authorization[name]):
                    raise ValueError
            now = self.clock()
            if (authorization["to"].lower() != accepted["payTo"].lower()
                    or authorization["value"] != accepted["amount"]
                    or not 0 <= int(authorization["validAfter"]) < now
                    or not now + 6 < int(authorization["validBefore"]) <= now + accepted["maxTimeoutSeconds"]
                    or not isinstance(authorization["nonce"], str)
                    or not re.fullmatch(r"0x[0-9a-fA-F]{64}", authorization["nonce"])
                    or not isinstance(inner["signature"], str)
                    or not re.fullmatch(r"0x[0-9a-fA-F]{2,8192}", inner["signature"])):
                raise ValueError
        except (ValueError, TypeError, KeyError, AttributeError):
            raise PaymentPolicyError("Saved payment authorization is invalid or expired; reconcile manually") from None

    def metadata(self, handle):
        with self.store.locked(handle):
            state = self._load(handle)
            response = self._request("GET", self._dialog_path(state), handle=handle,
                                     state=state, bucket="read")
            if response.status_code != 200:
                self._error(response)
            result = self._json(response)
            state["limits"] = self._limits_from(result.get("limits", {}))
            self.store.save(handle, state)
            return result

    def check(self, handle):
        with self.store.locked(handle):
            state = self._load(handle)
            response = self._request("GET", self._dialog_path(state) + "/updates",
                params={"after": state["cursor"]}, handle=handle, state=state, bucket="poll")
            if response.status_code != 200:
                self._error(response)
            if response.content not in (b"0", b"1"):
                raise ApiError("invalid_response", 200)
            return response.content == b"1"

    def read(self, handle, limit=20):
        """Return the next page without acknowledgement or cursor advancement."""
        with self.store.locked(handle):
            state = self._load(handle)
            _integer(limit, maximum=min(50, state["limits"]["max_page_size"]))
            response = self._request("GET", self._dialog_path(state) + "/messages",
                params={"after": state["cursor"], "limit": limit}, handle=handle,
                state=state, bucket="read")
            if response.status_code != 200:
                self._error(response)
            result = self._json(response)
            try:
                messages = result["messages"]
                after = _integer(result["after"], minimum=0)
                if not isinstance(messages, list) or len(messages) > limit:
                    raise ValueError
                saved, previous = [], state["cursor"]
                for message in messages:
                    seq = _integer(message["seq"])
                    revision = _integer(message["revision"])
                    if seq <= previous or message["author"] not in ("owner", "agent"):
                        raise ValueError
                    saved.append({"seq": seq, "revision": revision, "author": message["author"]})
                    previous = seq
                if after != previous or type(result["has_more"]) is not bool:
                    raise ValueError
            except (KeyError, TypeError, ValueError, FarsideError):
                raise ApiError("invalid_response", 200) from None
            state["pending_read"] = {"before": state["cursor"], "after": after, "messages": saved}
            self.store.save(handle, state)
            return result

    def ack(self, handle):
        """Caller explicitly confirms processing every message on the saved page."""
        with self.store.locked(handle):
            state = self._load(handle)
            page = state.get("pending_read")
            if not page or page["before"] != state["cursor"]:
                raise StateError("Read and process a page before acknowledging it")
            pairs = [{"seq": item["seq"], "revision": item["revision"]}
                     for item in page["messages"] if item["author"] == "owner"]
            if pairs:
                if len(pairs) > state["limits"]["max_acknowledgements"]:
                    raise StateError("Saved page exceeds the acknowledgement limit; read a smaller page")
                response = self._request("POST", self._dialog_path(state) + "/ack",
                    body=_encode({"messages": pairs}), handle=handle, state=state, bucket="write")
                if response.status_code != 204:
                    self._error(response)
            state["cursor"] = page["after"]
            state["pending_read"] = None
            self.store.save(handle, state)
            return {"cursor": state["cursor"]}

    def send(self, handle, text):
        _text(text)
        return self._write(handle, "POST", "/messages", {"text": text}, 201, ambiguous=True)

    def edit(self, handle, seq, revision, text):
        _integer(seq)
        _integer(revision)
        _text(text)
        return self._write(handle, "PATCH", f"/messages/{seq}",
                           {"revision": revision, "text": text}, 200, ambiguous=True)

    def close(self, handle):
        self._write(handle, "POST", "/close", {}, 204)

    def _write(self, handle, method, suffix, body, expected_status, ambiguous=False):
        with self.store.locked(handle):
            state = self._load(handle)
            response = self._request(method, self._dialog_path(state) + suffix,
                handle=handle, state=state, bucket="write", body=_encode(body), ambiguous=ambiguous)
            if ambiguous and response.status_code >= 500:
                raise AmbiguousWrite("Write outcome is unknown; reconcile before another write")
            if response.status_code != expected_status:
                self._error(response)
            if expected_status == 204:
                return None
            try:
                result = self._json(response)
                _integer(result["seq"])
                _integer(result["revision"])
                return result
            except (ApiError, FarsideError, KeyError):
                raise AmbiguousWrite("Write response is invalid; reconcile the outcome") from None