"""Installed NEOedgeX market worker: NEOFE1 polling, local signing, deal readback.

This is the existing ``neoedges`` inbox contract brought into the product that already
contains the marketplace adapter.  No inbound edge port exists.  A task is signed only when
it matches a proposal in the local config byte-for-byte.  The completed handshake is fetched
from the register read surface and then handed to the local core as a CONTRACT.
"""

from __future__ import annotations

import hashlib
import json
import os
import random
import re
import threading
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

from .canonical import canonical_bytes
from .identity import DeviceIdentity

CONFIG_SCHEMA = "neoedgex-market-worker.v1"
CONSENT_SCHEMA = "NEOEDGES_PROCESS_CONSENT_V1"
VOTE_DOMAIN = b"NEOEDGES:CONSENT_VOTE:V1"
TXN_DOMAIN = b"NEOEDGES:HANDSHAKE_TXN:V1"


class MarketWorkerError(ValueError):
    pass


def grammar_member_id(participant_id: str) -> str:
    """The existing explicit crosswalk: register/Chamber id -> NEOedge grammar id."""
    if participant_id.startswith("urn:neoedge:member:"):
        return participant_id
    if re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,62}", participant_id) is None:
        raise MarketWorkerError("PARTICIPANT_ID_CANNOT_ENTER_EDGE_GRAMMAR")
    return "urn:neoedge:member:" + participant_id


@dataclass(frozen=True)
class Proposal:
    offer_ref: str
    process_version: str
    parties: tuple[str, str]
    terms: dict
    decision: str = "YES"

    def body(self) -> dict:
        return {"schema": CONSENT_SCHEMA, "handshake_ref": self.offer_ref,
                "process_version": self.process_version, "parties": sorted(self.parties),
                "terms": self.terms}

    def digest(self) -> str:
        return hashlib.sha256(canonical_bytes(self.body())).hexdigest()

    def txn_id(self) -> str:
        return hashlib.sha256(TXN_DOMAIN + canonical_bytes({
            "handshake_ref": self.offer_ref, "terms_hash": self.digest(),
            "parties": sorted(self.parties)})).hexdigest()


@dataclass(frozen=True)
class WorkerConfig:
    member_id: str
    connector_id: str
    door_url: str
    token_file: str
    poll_seconds: int
    jitter_seconds: int
    proposals: tuple[Proposal, ...]


def load_config(path: str | os.PathLike) -> WorkerConfig:
    try:
        document = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, ValueError) as error:
        raise MarketWorkerError(f"CONFIG_UNREADABLE:{error}") from error
    expected = {"schema_v", "member_id", "connector_id", "door_url", "token_file",
                "poll_seconds", "jitter_seconds", "proposals"}
    if not isinstance(document, dict) or set(document) != expected \
            or document.get("schema_v") != CONFIG_SCHEMA:
        raise MarketWorkerError("CONFIG_FIELD_SET_INVALID")
    if not isinstance(document["poll_seconds"], int) or document["poll_seconds"] < 1 \
            or not isinstance(document["jitter_seconds"], int) or document["jitter_seconds"] < 0:
        raise MarketWorkerError("POLL_INTERVAL_INVALID")
    proposals, seen = [], set()
    for row in document["proposals"]:
        if not isinstance(row, dict) or set(row) != {
                "offer_ref", "process_version", "parties", "terms", "decision"}:
            raise MarketWorkerError("PROPOSAL_FIELD_SET_INVALID")
        parties = row["parties"]
        if not isinstance(parties, list) or len(parties) != 2 or len(set(parties)) != 2:
            raise MarketWorkerError("PROPOSAL_PARTIES_INVALID")
        proposal = Proposal(str(row["offer_ref"]), str(row["process_version"]),
                            (str(parties[0]), str(parties[1])), row["terms"],
                            str(row["decision"]))
        if proposal.offer_ref in seen or proposal.decision not in ("YES", "NO") \
                or document["member_id"] not in proposal.parties:
            raise MarketWorkerError("PROPOSAL_INVALID_OR_DUPLICATE")
        seen.add(proposal.offer_ref)
        proposals.append(proposal)
    return WorkerConfig(str(document["member_id"]), str(document["connector_id"]),
                        str(document["door_url"]).rstrip("/"), str(document["token_file"]),
                        document["poll_seconds"], document["jitter_seconds"], tuple(proposals))


class HttpWire:
    def __init__(self, base_url: str, token: str, timeout_s: float = 10.0) -> None:
        self.base = base_url.rstrip("/")
        self.token = token
        self.timeout_s = timeout_s

    def call(self, method: str, path: str, body: dict | None = None) -> tuple[int, dict | None]:
        request = urllib.request.Request(
            self.base + path,
            data=None if body is None else canonical_bytes(body),
            headers={"Content-Type": "application/json", "Authorization": "Bearer " + self.token},
            method=method,
        )
        try:
            with urllib.request.urlopen(request, timeout=self.timeout_s) as response:
                return response.status, json.loads(response.read())
        except urllib.error.HTTPError as error:
            try:
                return error.code, json.loads(error.read())
            except ValueError:
                return error.code, None
        except (urllib.error.URLError, OSError, ValueError):
            return 0, None


class GrossoSettlementSink:
    """POST one verified ingress and require exact H2 readback before success."""

    def __init__(self, base_url: str, token_file: str | os.PathLike,
                 *, wire: HttpWire | None = None) -> None:
        token_path = Path(token_file)
        try:
            if token_path.stat().st_mode & 0o077:
                raise MarketWorkerError("GROSSO_TOKEN_FILE_NOT_0600")
            token = token_path.read_text(encoding="utf-8").strip()
        except OSError as error:
            raise MarketWorkerError(f"GROSSO_TOKEN_FILE_UNREADABLE:{error}") from error
        if not token:
            raise MarketWorkerError("GROSSO_TOKEN_EMPTY")
        self.wire = wire or HttpWire(base_url, token)

    def __call__(self, document: dict) -> None:
        txn_id = document.get("handshake", {}).get("ref")
        commitment = hashlib.sha256(canonical_bytes(document)).hexdigest()
        status, answer = self.wire.call("POST", "/settlements/neoedgex", document)
        if status not in (200, 201) or not isinstance(answer, dict) \
                or answer.get("txn_id") != txn_id \
                or answer.get("commitment_hex") != commitment:
            raise MarketWorkerError("GROSSO_SETTLEMENT_REFUSED_OR_MISMATCH")
        read_status, readback = self.wire.call("GET", f"/settlements/{txn_id}")
        fields = ("txn_id", "settlement", "outcome_recorded", "reason", "commitment_hex")
        if read_status != 200 or not isinstance(readback, dict) \
                or any(readback.get(field) != answer.get(field) for field in fields):
            raise MarketWorkerError("GROSSO_SETTLEMENT_READBACK_MISMATCH")


class MarketWorker:
    def __init__(self, config: WorkerConfig, identity: DeviceIdentity, *,
                 on_handshake: Callable[[dict], None], state_dir: str | os.PathLike,
                 wire: HttpWire | None = None,
                 on_settlement: Callable[[dict], None] | None = None) -> None:
        self.config = config
        self.identity = identity
        self.on_handshake = on_handshake
        self.on_settlement = on_settlement
        self.state_dir = Path(state_dir)
        token_path = Path(config.token_file)
        try:
            if token_path.stat().st_mode & 0o077:
                raise MarketWorkerError("TOKEN_FILE_NOT_0600")
            token = token_path.read_text(encoding="utf-8").strip()
        except OSError as error:
            raise MarketWorkerError(f"TOKEN_FILE_UNREADABLE:{error}") from error
        if not token:
            raise MarketWorkerError("TOKEN_EMPTY")
        self.wire = wire or HttpWire(config.door_url, token)
        self.proposals = {proposal.offer_ref: proposal for proposal in config.proposals}
        self.pending_path = self.state_dir / "market-pending.jsonl"
        self.receipts_path = self.state_dir / "market-receipts.jsonl"
        self.owner_inbox_path = self.state_dir / "market-owner-inbox.jsonl"

    def _append(self, path: Path, row: dict) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a", encoding="utf-8") as handle:
            handle.write(json.dumps(row, sort_keys=True, separators=(",", ":")) + "\n")
            handle.flush()
            os.fsync(handle.fileno())

    def _counterparty(self, task: dict) -> str:
        connectors = task.get("connectors")
        if isinstance(connectors, list):
            others = [str(c) for c in connectors if str(c) != self.config.connector_id]
            if others:
                return others[0]
        return ""

    def _owner_inbox_states(self) -> dict[str, str]:
        # Fold exactly as the TRU steward does: last row per intent wins, absent
        # state reads as AWAITING_OWNER.
        states: dict[str, str] = {}
        if self.owner_inbox_path.exists():
            for line in self.owner_inbox_path.read_text(encoding="utf-8").splitlines():
                try:
                    row = json.loads(line)
                except ValueError:
                    continue
                intent_id = row.get("intent_id")
                if isinstance(intent_id, str) and intent_id:
                    states[intent_id] = str(row.get("state", "AWAITING_OWNER"))
        return states

    def _owner_note(self, intent_id: str, offer_ref: str, counterparty: str,
                    state: str, at: str) -> None:
        """H1 pre-signing owner inbox (contract agreed with Codex, 2026-08-18):
        append-only, the TRU steward folds the LAST state per intent and rings only
        AWAITING_OWNER.  The inbox is enrichment, never the market's critical path —
        an unwritable inbox must not stop voting, so failures are swallowed."""
        try:
            if self._owner_inbox_states().get(intent_id) == state:
                return
            self._append(self.owner_inbox_path, {
                "intent_id": intent_id, "offer_ref": offer_ref,
                "counterparty": counterparty, "created_at": at, "state": state})
        except OSError:
            pass

    def _vote(self, task: dict, proposal: Proposal, at: str) -> dict:
        required = {"intent_id", "offer_ref", "terms_hash", "connectors", "process_version"}
        if not isinstance(task, dict) or required - set(task) \
                or self.config.connector_id not in task["connectors"] \
                or task["offer_ref"] != proposal.offer_ref \
                or task["terms_hash"] != proposal.digest() \
                or task["process_version"] != proposal.process_version:
            # The task names our offer but does not match it byte-for-byte: only the
            # owner's word (a config correction) can unblock this intent.
            intent_id = task.get("intent_id") if isinstance(task, dict) else None
            if isinstance(intent_id, str) and intent_id:
                self._owner_note(intent_id, proposal.offer_ref,
                                 self._counterparty(task), "AWAITING_OWNER", at)
            return {"state": "REFUSED_LOCAL", "intent_id": intent_id}
        unsigned = {"schema": CONSENT_SCHEMA, "handshake_ref": proposal.offer_ref,
                    "party_id": self.config.member_id, "decision": proposal.decision,
                    "proposal_digest": proposal.digest(),
                    "process_version": proposal.process_version, "at": at}
        vote_digest = hashlib.sha256(VOTE_DOMAIN + canonical_bytes(unsigned)).hexdigest()
        body = {"intent_id": task["intent_id"], "connector_id": self.config.connector_id,
                "decision": proposal.decision, "vote_digest": vote_digest, "at": at,
                "signature": self.identity.signer(VOTE_DOMAIN + canonical_bytes(unsigned)).hex(),
                "signature_suite": "ed25519-v1", "txn_id": proposal.txn_id()}
        status, answer = self.wire.call("POST", f"/intents/{task['intent_id']}/votes", body)
        # NEOFE1 may wrap the register's verdict.
        if status == 202 and isinstance(answer, dict) and "book" in answer:
            held = answer["book"]
            status, answer = ((0, None) if held is None else
                              (held["status_code"], held["body"]))
        state = "RECORDED" if status == 201 else "DUPLICATE" if status == 200 else \
            "UNDELIVERED" if status == 0 else "REFUSED_REMOTE"
        receipt = {"state": state, "intent_id": task["intent_id"],
                   "txn_id": proposal.txn_id(), "http_status": status}
        self._append(self.receipts_path, receipt)
        # Owner-inbox translation: a recorded/duplicate vote silences the intent
        # (SIGNED); a remote refusal is terminal for it (RESOLVED); UNDELIVERED is
        # transient and writes nothing (the next sweep retries).
        if state in ("RECORDED", "DUPLICATE"):
            self._owner_note(task["intent_id"], proposal.offer_ref,
                             self._counterparty(task), "SIGNED", at)
        elif state == "REFUSED_REMOTE":
            self._owner_note(task["intent_id"], proposal.offer_ref,
                             self._counterparty(task), "RESOLVED", at)
        if state in ("RECORDED", "DUPLICATE"):
            self._append(self.pending_path, {"txn_id": proposal.txn_id(),
                                             "intent_id": task["intent_id"],
                                             "offer_ref": proposal.offer_ref,
                                             "resolved": False})
        return receipt

    def _pending(self) -> dict[str, dict]:
        state = {}
        if self.pending_path.exists():
            for line in self.pending_path.read_text(encoding="utf-8").splitlines():
                row = json.loads(line)
                state[row["txn_id"]] = row
        return state

    def readback(self) -> int:
        resolved = 0
        for txn_id, pending in self._pending().items():
            if bool(pending["resolved"]):
                continue
            status, answer = self.wire.call("GET", f"/deal/{txn_id}")
            if status != 200 or not isinstance(answer, dict):
                continue
            events = answer.get("chain", ())
            handshake = next((event for event in events
                              if isinstance(event, dict) and event.get("event_type") == "HANDSHAKE_MADE"), None)
            if handshake is None:
                continue
            if not bool(pending.get("contract_recorded", False)):
                self.on_handshake(handshake)
                pending = {**pending, "contract_recorded": True}
                self._append(self.pending_path, pending)
            if self.on_settlement is not None:
                intent_id, offer_ref = pending.get("intent_id"), pending.get("offer_ref")
                proposal = self.proposals.get(offer_ref)
                if not isinstance(intent_id, str) or proposal is None:
                    # A legacy pending row may still complete the local CONTRACT, but cannot
                    # cross the GROSSO gate without its exact public intent identity.
                    continue
                intent_status, intent_view = self.wire.call("GET", f"/intents/{intent_id}")
                if intent_status != 200:
                    continue
                from .grosso_ingress import build_grosso_ingress
                self.on_settlement(build_grosso_ingress(handshake, intent_view, proposal))
            self._append(self.pending_path, {"txn_id": txn_id,
                                             "intent_id": pending.get("intent_id"),
                                             "offer_ref": pending.get("offer_ref"),
                                             "contract_recorded": True,
                                             "resolved": True})
            resolved += 1
        return resolved

    def sweep(self, at: str) -> tuple[dict, ...]:
        status, answer = self.wire.call("GET", f"/intents/waiting/{self.config.connector_id}")
        receipts = []
        if status == 200 and isinstance(answer, dict):
            for task in answer.get("waiting", ()):
                proposal = self.proposals.get(task.get("offer_ref")) if isinstance(task, dict) else None
                if proposal is not None:
                    receipts.append(self._vote(task, proposal, at))
                elif isinstance(task, dict):
                    # No pre-approved proposal for this offer: the intent waits for the
                    # owner's word, so it goes to the pre-signing owner inbox (deduped —
                    # repeated sweeps do not re-ring).
                    intent_id, offer_ref = task.get("intent_id"), task.get("offer_ref")
                    if isinstance(intent_id, str) and intent_id:
                        self._owner_note(intent_id, str(offer_ref or ""),
                                         self._counterparty(task), "AWAITING_OWNER", at)
        self.readback()
        return tuple(receipts)

    def run(self, stop: threading.Event, *, clock=time.time,
            rng: Callable[[int], int] | None = None) -> None:
        jitter = rng or (lambda upper: random.SystemRandom().randrange(upper))
        while not stop.is_set():
            at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(clock()))
            self.sweep(at)
            extra = jitter(self.config.jitter_seconds + 1) if self.config.jitter_seconds else 0
            stop.wait(self.config.poll_seconds + extra)
