from __future__ import annotations

import asyncio
import json
import logging
import random
import threading
import time
from dataclasses import asdict, dataclass, field, is_dataclass
from typing import (
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Optional,
    Tuple,
    TypedDict,
    Union,
)

import httpx

from .cooldown import BlockFoundCooldown, get_block_found_cooldown


Json = Union[dict, list, str, int, float, None]


class RpcError(Exception):
    def __init__(self, code: int, message: str, data: Any = None):
        super().__init__(f"RPC error {code}: {message}")
        self.code = code
        self.data = data


class TransportError(Exception):
    pass


class SubmitRejected(Exception):
    """Raised when the node explicitly rejects a share/block (semantic failure)."""


@dataclass
class SubmitterConfig:
    rpc_url: str = "http://127.0.0.1:8545/rpc"
    timeout_s: float = 5.0
    connect_timeout_s: float = 2.0
    read_timeout_s: float = 5.0
    write_timeout_s: float = 5.0
    pool_timeout_s: float = 5.0
    submit_timeout_s: float = 5.0
    max_retries: int = 5
    initial_backoff_s: float = 0.25
    max_backoff_s: float = 5.0
    jitter: float = 0.25  # 0..1 proportion of backoff to add/sub as jitter
    batch_size: int = 256
    http_headers: Dict[str, str] = field(
        default_factory=lambda: {"Content-Type": "application/json"}
    )
    method_submit_share: str = "miner.submitShare"
    method_submit_share_batch: str = "miner.submitShareBatch"
    method_submit_block: str = "miner.submitBlock"


class ShareResult(TypedDict, total=False):
    accepted: bool
    reason: str
    hash: str
    d_ratio: float
    height: int


@dataclass
class SubmitStats:
    shares_accepted: int = 0
    shares_rejected: int = 0
    shares_errors: int = 0
    blocks_accepted: int = 0
    blocks_rejected: int = 0
    last_error: Optional[str] = None

    def snapshot(self) -> "SubmitStats":
        return SubmitStats(
            shares_accepted=self.shares_accepted,
            shares_rejected=self.shares_rejected,
            shares_errors=self.shares_errors,
            blocks_accepted=self.blocks_accepted,
            blocks_rejected=self.blocks_rejected,
            last_error=self.last_error,
        )


def json_sanitize(obj: Any) -> Json:
    if obj is None or isinstance(obj, (str, int, float, bool)):
        return obj
    if isinstance(obj, bytes):
        return f"0x{obj.hex()}"
    if isinstance(obj, (bytearray, memoryview)):
        return f"0x{bytes(obj).hex()}"
    if is_dataclass(obj):
        return json_sanitize(asdict(obj))
    if hasattr(obj, "model_dump"):
        return json_sanitize(obj.model_dump())  # type: ignore[call-arg]
    if hasattr(obj, "dict") and callable(obj.dict):
        return json_sanitize(obj.dict())  # type: ignore[call-arg]
    if isinstance(obj, dict):
        return {str(k): json_sanitize(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple, set)):
        return [json_sanitize(v) for v in obj]
    if hasattr(obj, "to_dict"):
        return json_sanitize(obj.to_dict())  # type: ignore[attr-defined]
    if hasattr(obj, "__dict__"):
        return json_sanitize(vars(obj))
    return str(obj)


def json_validate(obj: Any) -> Json:
    sanitized = json_sanitize(obj)
    try:
        json.dumps(sanitized)
    except TypeError as exc:
        raise ValueError("payload must be JSON-serializable") from exc
    return sanitized


# Constants for auto-generated jobId
TIMESTAMP_SUFFIX_MODULO = 100000  # Last 5 digits of timestamp for uniqueness


def _extract_nonce_value(m: Dict[str, Any]) -> Any:
    """Extract nonce value from share dict using common key variants."""
    return m.get("nonce") or m.get("nonce64") or m.get("n") or 0


def _format_nonce_for_jobid(nonce_val: Any) -> str:
    """Format nonce value as hex string prefix for jobId generation."""
    if isinstance(nonce_val, int):
        return f"{nonce_val:016x}"[:8]  # First 8 hex chars
    elif isinstance(nonce_val, (bytes, bytearray)):
        return nonce_val.hex()[:8]
    elif isinstance(nonce_val, str):
        nonce_str = nonce_val.replace("0x", "")
        return nonce_str[:8]
    else:
        return "00000000"  # Fallback for invalid nonce


def _default_share_encoder(share: Any) -> Dict[str, Any]:
    """
    Try hard to convert a FoundShare-like object into a JSON-RPC payload.

    Expected keys (best-effort):
      - header: dict or bytes(hex)
      - nonce: int or hex string
      - mixSeed: bytes(hex) or omitted
      - proof: dict (HashShare envelope) or fields under "hashshare"
      - d_ratio: float (share difficulty ratio)
      - height: int (template height)

    We support:
      - dataclass with asdict()
      - object with .to_dict()
      - mapping
    """
    # Mapping already?
    if isinstance(share, dict):
        m = dict(share)
    else:
        # dataclass?
        try:
            m = asdict(share)  # type: ignore[arg-type]
        except Exception:
            # generic "to_dict"
            if hasattr(share, "to_dict"):
                m = share.to_dict()  # type: ignore[attr-defined]
            else:
                # As a last resort, introspect attributes
                m = {k: getattr(share, k) for k in dir(share) if not k.startswith("_")}

    # Normalize common keys/casing
    payload: Dict[str, Any] = {}
    # Job id (for stale rejection)
    job_id = m.get("jobId") or m.get("job_id") or m.get("job")
    if job_id is None:
        # Generate a fallback jobId from nonce, height, and timestamp
        # This ensures shares always have a jobId for tracking and deduplication
        nonce_val = _extract_nonce_value(m)
        height_val = m.get("height") or 0
        timestamp_val = int(time.time() * 1000)
        
        # Format nonce as hex prefix
        nonce_prefix = _format_nonce_for_jobid(nonce_val)
        
        # Generate deterministic jobId: "auto-{height}-{nonce_prefix}-{timestamp_suffix}"
        # This format helps identify auto-generated IDs and provides uniqueness
        job_id = f"auto-{height_val}-{nonce_prefix}-{timestamp_val % TIMESTAMP_SUFFIX_MODULO}"
    
    payload["jobId"] = job_id

    # Header/template
    header = m.get("header") or m.get("header_template") or m.get("candidate_header")
    if header is None:
        raise ValueError("share encoder: missing 'header'")
    payload["header"] = header

    # Nonce / mixSeed
    nonce = _extract_nonce_value(m)
    if nonce is None or nonce == 0:
        raise ValueError("share encoder: missing 'nonce'")
    payload["nonce"] = nonce

    mix = m.get("mix_seed") or m.get("mixSeed") or m.get("mix")
    if mix is not None:
        payload["mixSeed"] = mix

    # Proof (HashShare)
    proof = (
        m.get("proof")
        or m.get("hashshare")
        or m.get("hash_share")
        or m.get("hashShare")
        or m.get("proof_envelope")
    )
    if proof is None:
        raise ValueError("share encoder: missing 'proof' (HashShare envelope)")
    payload["proof"] = proof

    # Optional hints
    if "d_ratio" in m:
        payload["d_ratio"] = m["d_ratio"]
    if "height" in m:
        payload["height"] = m["height"]
    if "shareTarget" in m:
        payload["shareTarget"] = m["shareTarget"]
    # Forward any UWP envelopes attached by the scanner so the node-side
    # UWP verifier (rpc.methods.miner._process_attached_proofs) can credit
    # bonus AICF credits to the miner.
    attached = m.get("attachedProofs") or m.get("attached_proofs")
    if isinstance(attached, list) and attached:
        payload["attachedProofs"] = attached

    return payload


def _default_block_encoder(candidate_block: Any) -> Dict[str, Any]:
    """
    Convert a candidate Block object/dict to the RPC payload expected by miner.submitBlock.
    Minimal contract: {'header': {...}, 'txs': [...], 'proofs': [...]}.
    """
    if isinstance(candidate_block, dict):
        b = dict(candidate_block)
    else:
        try:
            b = asdict(candidate_block)  # type: ignore[arg-type]
        except Exception:
            if hasattr(candidate_block, "to_dict"):
                b = candidate_block.to_dict()  # type: ignore[attr-defined]
            else:
                b = {
                    k: getattr(candidate_block, k)
                    for k in dir(candidate_block)
                    if not k.startswith("_")
                }

    for key in ("header", "txs", "proofs"):
        if key not in b:
            raise ValueError(f"block encoder: missing '{key}'")
    return {"header": b["header"], "txs": b["txs"], "proofs": b["proofs"]}


class JsonRpcClient:
    def __init__(
        self, url: str, headers: Optional[Dict[str, str]] = None, timeout_s: float = 5.0
    ):
        self._url = url
        self._headers = headers or {"Content-Type": "application/json"}
        self._timeout_s = timeout_s
        self._id = 0
        self._log = logging.getLogger("mining.share_submitter.rpc")
        self._client = httpx.Client(timeout=self._timeout_s, headers=self._headers)

    def _next_id(self) -> int:
        self._id += 1
        return self._id

    def call(self, method: str, params: Any) -> Any:
        req = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": method,
            "params": params,
        }
        try:
            resp = self._client.post(self._url, json=json_sanitize(req))
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise TransportError(
                f"HTTP {e.response.status_code}: {e.response.text}"
            ) from e
        except httpx.TimeoutException as e:
            raise TransportError(f"timeout: {e}") from e
        except httpx.TransportError as e:
            raise TransportError(str(e)) from e
        except Exception as e:
            raise TransportError(repr(e)) from e

        try:
            obj = resp.json()
        except Exception as e:
            raise TransportError(f"Invalid JSON from RPC: {e}") from e

        if "error" in obj and obj["error"]:
            err = obj["error"]
            raise RpcError(
                err.get("code", -32000),
                err.get("message", "Unknown error"),
                err.get("data"),
            )
        return obj.get("result")

    def batch(self, calls: List[Tuple[str, Any]]) -> List[Any]:
        batch_req = [
            {"jsonrpc": "2.0", "id": self._next_id(), "method": m, "params": p}
            for (m, p) in calls
        ]
        try:
            resp = self._client.post(self._url, json=json_sanitize(batch_req))
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise TransportError(
                f"HTTP {e.response.status_code}: {e.response.text}"
            ) from e
        except httpx.TimeoutException as e:
            raise TransportError(f"timeout: {e}") from e
        except httpx.TransportError as e:
            raise TransportError(str(e)) from e
        except Exception as e:
            raise TransportError(repr(e)) from e

        try:
            arr = resp.json()
        except Exception as e:
            raise TransportError(f"Invalid JSON from RPC: {e}") from e

        # Map id -> result/error; then reorder to match input order
        by_id = {item["id"]: item for item in arr}
        out: List[Any] = []
        for req in batch_req:
            item = by_id.get(req["id"])
            if item is None:
                out.append(RpcError(-32000, "Missing response item"))
                continue
            if "error" in item and item["error"]:
                err = item["error"]
                out.append(
                    RpcError(
                        err.get("code", -32000),
                        err.get("message", "Unknown error"),
                        err.get("data"),
                    )
                )
            else:
                out.append(item.get("result"))
        return out

    def close(self) -> None:
        self._client.close()


class AsyncJsonRpcClient:
    def __init__(
        self,
        url: str,
        headers: Optional[Dict[str, str]] = None,
        timeout: Optional[httpx.Timeout] = None,
        client: Optional[httpx.AsyncClient] = None,
    ) -> None:
        self._url = url
        self._headers = headers or {"Content-Type": "application/json"}
        self._timeout = timeout or httpx.Timeout(5.0)
        self._id = 0
        self._log = logging.getLogger("mining.share_submitter.rpc")
        self._owns_client = client is None
        self._client = client or httpx.AsyncClient(
            timeout=self._timeout, headers=self._headers
        )

    def _next_id(self) -> int:
        self._id += 1
        return self._id

    async def call(
        self, method: str, params: Any, *, timeout_s: Optional[float] = None
    ) -> Any:
        req = {
            "jsonrpc": "2.0",
            "id": self._next_id(),
            "method": method,
            "params": params,
        }
        req = json_sanitize(req)
        timeout = self._timeout if timeout_s is None else httpx.Timeout(timeout_s)
        try:
            resp = await self._client.post(self._url, json=req, timeout=timeout)
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise TransportError(
                f"HTTP {e.response.status_code}: {e.response.text}"
            ) from e
        except httpx.TimeoutException as e:
            raise TransportError(f"timeout: {e}") from e
        except httpx.TransportError as e:
            raise TransportError(str(e)) from e
        except Exception as e:
            raise TransportError(repr(e)) from e

        try:
            obj = resp.json()
        except Exception as e:
            raise TransportError(f"Invalid JSON from RPC: {e}") from e

        if "error" in obj and obj["error"]:
            err = obj["error"]
            raise RpcError(
                err.get("code", -32000),
                err.get("message", "Unknown error"),
                err.get("data"),
            )
        return obj.get("result")

    async def aclose(self) -> None:
        if self._owns_client:
            await self._client.aclose()


class ShareSubmitter:
    """
    Consumes FoundShare objects, submits them to the local node via JSON-RPC with
    retries/backoff, and records basic stats.

    It first attempts "miner.submitShareBatch" for efficiency; if the node returns
    -32601 (Method not found), it falls back to one-by-one "miner.submitShare".
    """

    def __init__(
        self,
        cfg: SubmitterConfig,
        *,
        share_encoder: Callable[[Any], Dict[str, Any]] = _default_share_encoder,
        block_encoder: Callable[[Any], Dict[str, Any]] = _default_block_encoder,
        logger: Optional[logging.Logger] = None,
        http_client: Optional[httpx.AsyncClient] = None,
        cooldown: BlockFoundCooldown | None = None,
    ) -> None:
        self.cfg = cfg
        self.rpc = JsonRpcClient(cfg.rpc_url, cfg.http_headers, cfg.timeout_s)
        async_timeout = httpx.Timeout(
            timeout=cfg.submit_timeout_s,
            connect=cfg.connect_timeout_s,
            read=cfg.read_timeout_s,
            write=cfg.write_timeout_s,
            pool=cfg.pool_timeout_s,
        )
        self._async_rpc = AsyncJsonRpcClient(
            cfg.rpc_url,
            cfg.http_headers,
            timeout=async_timeout,
            client=http_client,
        )
        self._share_encoder = share_encoder
        self._block_encoder = block_encoder
        self._log = logger or logging.getLogger("mining.share_submitter")
        self._stats = SubmitStats()
        self._closed = False
        self._thread: Optional[threading.Thread] = None
        self._stop_evt = threading.Event()
        self._cooldown = cooldown or get_block_found_cooldown()

    # ──────────────────────────────────────────────────────────────────────
    # Public stats/lifecycle
    # ──────────────────────────────────────────────────────────────────────

    def stats(self) -> SubmitStats:
        return self._stats.snapshot()

    def close(self) -> None:
        self._closed = True
        self._stop_evt.set()
        if hasattr(self.rpc, "close"):
            self.rpc.close()

    async def aclose(self) -> None:
        self.close()
        await self._async_rpc.aclose()

    # ──────────────────────────────────────────────────────────────────────
    # Single-shot APIs (usable without the background consumer)
    # ──────────────────────────────────────────────────────────────────────

    async def submit(self, params: Dict[str, Any]) -> ShareResult:
        payload = json_validate(self._share_encoder(params))
        if not isinstance(payload, dict):
            raise ValueError("share payload must be an object")
        # Note: jobId should always be present due to fallback generation in encoder
        # This check is a safety net for custom encoders
        if not payload.get("jobId"):
            raise ValueError(
                "share payload missing jobId (encoder should have generated fallback)"
            )
        backoff = self.cfg.initial_backoff_s
        tries = 0
        while True:
            tries += 1
            t0 = time.perf_counter()
            try:
                res = await self._async_rpc.call(
                    self.cfg.method_submit_share, [payload]
                )
                dt = time.perf_counter() - t0
                if isinstance(res, dict):
                    accepted = bool(res.get("accepted", False))
                    if accepted:
                        self._stats.shares_accepted += 1
                        if res.get("isBlock"):
                            self._stats.blocks_accepted += 1
                            self._cooldown.notify_block_accepted(
                                height=res.get("height"), block_hash=res.get("hash")
                            )
                    else:
                        self._stats.shares_rejected += 1
                        if res.get("isBlock"):
                            self._stats.blocks_rejected += 1
                    self._log.info(
                        "submitShare result accepted=%s reason=%s latency=%.3fs jobId=%s",
                        accepted,
                        res.get("reason"),
                        dt,
                        payload.get("jobId"),
                    )
                    return ShareResult(
                        accepted=accepted,
                        reason=str(res.get("reason", "")),
                        hash=str(res.get("hash", "")),
                        d_ratio=float(res.get("d_ratio", payload.get("d_ratio") or 0.0)),
                        height=int(res.get("height", payload.get("height") or 0)),
                    )
                if res is True:
                    self._stats.shares_accepted += 1
                    self._log.info(
                        "submitShare result accepted=True latency=%.3fs jobId=%s",
                        dt,
                        payload.get("jobId"),
                    )
                    return ShareResult(accepted=True)
                self._stats.shares_rejected += 1
                self._log.info(
                    "submitShare result accepted=False reason=unexpected-result latency=%.3fs jobId=%s",
                    dt,
                    payload.get("jobId"),
                )
                return ShareResult(accepted=False, reason="unexpected-result")
            except RpcError as e:
                if e.code == -32601 and tries == 1:
                    try:
                        br = await self._async_rpc.call(
                            self.cfg.method_submit_share_batch, [[payload]]
                        )
                        if isinstance(br, list) and br:
                            out = br[0]
                            if isinstance(out, dict):
                                accepted = bool(out.get("accepted", False))
                                if accepted:
                                    self._stats.shares_accepted += 1
                                    if out.get("isBlock"):
                                        self._cooldown.notify_block_accepted(
                                            height=out.get("height"),
                                            block_hash=out.get("hash"),
                                        )
                                else:
                                    self._stats.shares_rejected += 1
                                return ShareResult(
                                    accepted=accepted,
                                    reason=str(out.get("reason", "")),
                                    hash=str(out.get("hash", "")),
                                    d_ratio=float(
                                        out.get("d_ratio", payload.get("d_ratio") or 0.0)
                                    ),
                                    height=int(
                                        out.get("height", payload.get("height") or 0)
                                    ),
                                )
                            if out is True:
                                self._stats.shares_accepted += 1
                                return ShareResult(accepted=True)
                            self._stats.shares_rejected += 1
                            return ShareResult(
                                accepted=False, reason="unexpected-batch-result"
                            )
                    except RpcError as e2:
                        self._stats.last_error = f"{e2.code}:{e2}"
                if e.code in (-32010, -32011, -32012):
                    self._stats.shares_rejected += 1
                    return ShareResult(accepted=False, reason=f"rpc:{e.code}:{e}")
                self._stats.shares_errors += 1
                self._stats.last_error = f"{e.code}:{e}"
            except TransportError as e:
                self._stats.shares_errors += 1
                self._stats.last_error = str(e)

            if tries >= self.cfg.max_retries:
                return ShareResult(
                    accepted=False,
                    reason=f"retries-exhausted:{self._stats.last_error or ''}",
                )

            sleep = backoff * (1.0 + (random.random() * 2 - 1) * self.cfg.jitter)
            sleep = max(0.0, min(sleep, self.cfg.max_backoff_s))
            self._log.warning(
                "submitShare retry in %.2fs (try %s/%s): %s",
                sleep,
                tries,
                self.cfg.max_retries,
                self._stats.last_error,
            )
            await asyncio.sleep(sleep)
            backoff = min(backoff * 2.0, self.cfg.max_backoff_s)

    def submit_share_once(self, share: Any) -> ShareResult:
        payload = json_validate(self._share_encoder(share))
        if not isinstance(payload, dict):
            raise ValueError("share payload must be an object")
        # Note: jobId should always be present due to fallback generation in encoder
        # This check is a safety net for custom encoders
        if not payload.get("jobId"):
            raise ValueError(
                "share payload missing jobId (encoder should have generated fallback)"
            )
        backoff = self.cfg.initial_backoff_s
        tries = 0
        while True:
            tries += 1
            try:
                res = self.rpc.call(self.cfg.method_submit_share, [payload])
                # Result contract (recommended): {'accepted': bool, 'reason': str, 'hash': '0x…', ...}
                if isinstance(res, dict):
                    accepted = bool(res.get("accepted", False))
                    if accepted:
                        self._stats.shares_accepted += 1
                        if res.get("isBlock"):
                            self._stats.blocks_accepted += 1
                            self._cooldown.notify_block_accepted(
                                height=res.get("height"), block_hash=res.get("hash")
                            )
                    else:
                        self._stats.shares_rejected += 1
                        if res.get("isBlock"):
                            self._stats.blocks_rejected += 1
                    return ShareResult(
                        accepted=accepted,
                        reason=str(res.get("reason", "")),
                        hash=str(res.get("hash", "")),
                        d_ratio=float(
                            res.get("d_ratio", payload.get("d_ratio") or 0.0)
                        ),
                        height=int(res.get("height", payload.get("height") or 0)),
                    )
                # Fallback: some nodes may return True/False
                if res is True:
                    self._stats.shares_accepted += 1
                    return ShareResult(accepted=True)
                self._stats.shares_rejected += 1
                return ShareResult(accepted=False, reason="unexpected-result")
            except RpcError as e:
                # If method not found, caller might support batch only or a different mount.
                if e.code == -32601 and tries == 1:
                    # Try batch form with a single element
                    try:
                        br = self.rpc.call(
                            self.cfg.method_submit_share_batch, [[payload]]
                        )
                        # Expect a list of ShareResult
                        if isinstance(br, list) and br:
                            out = br[0]
                            if isinstance(out, dict):
                                accepted = bool(out.get("accepted", False))
                                if accepted:
                                    self._stats.shares_accepted += 1
                                    if out.get("isBlock"):
                                        self._cooldown.notify_block_accepted(
                                            height=out.get("height"),
                                            block_hash=out.get("hash"),
                                        )
                                else:
                                    self._stats.shares_rejected += 1
                                return ShareResult(
                                    accepted=accepted,
                                    reason=str(out.get("reason", "")),
                                    hash=str(out.get("hash", "")),
                                    d_ratio=float(
                                        out.get(
                                            "d_ratio", payload.get("d_ratio") or 0.0
                                        )
                                    ),
                                    height=int(
                                        out.get("height", payload.get("height") or 0)
                                    ),
                                )
                            if out is True:
                                self._stats.shares_accepted += 1
                                return ShareResult(accepted=True)
                            self._stats.shares_rejected += 1
                            return ShareResult(
                                accepted=False, reason="unexpected-batch-result"
                            )
                    except RpcError as e2:
                        # keep falling through to retry logic
                        self._stats.last_error = f"{e2.code}:{e2}"
                # Semantic rejections should not be retried aggressively
                if e.code in (
                    -32010,
                    -32011,
                    -32012,
                ):  # examples: WorkExpired, LowDifficulty, Duplicate
                    self._stats.shares_rejected += 1
                    return ShareResult(accepted=False, reason=f"rpc:{e.code}:{e}")
                # Transport-level or transient server errors: backoff
                self._stats.shares_errors += 1
                self._stats.last_error = f"{e.code}:{e}"
            except TransportError as e:
                self._stats.shares_errors += 1
                self._stats.last_error = str(e)

            if tries >= self.cfg.max_retries:
                return ShareResult(
                    accepted=False,
                    reason=f"retries-exhausted:{self._stats.last_error or ''}",
                )

            # Exponential backoff with jitter
            sleep = backoff * (1.0 + (random.random() * 2 - 1) * self.cfg.jitter)
            sleep = max(0.0, min(sleep, self.cfg.max_backoff_s))
            self._log.debug(
                "share submit retry",
                tries=tries,
                sleep=sleep,
                last_error=self._stats.last_error,
            )
            time.sleep(sleep)
            backoff = min(backoff * 2.0, self.cfg.max_backoff_s)

    def submit_block_once(self, candidate_block: Any) -> Dict[str, Any]:
        payload = json_validate(self._block_encoder(candidate_block))
        if not isinstance(payload, dict):
            raise ValueError("block payload must be an object")
        backoff = self.cfg.initial_backoff_s
        tries = 0
        while True:
            tries += 1
            try:
                res = self.rpc.call(self.cfg.method_submit_block, [payload])
                # Contract: {'accepted': bool, 'reason': str, 'hash': '0x…', 'height': N, 'duplicate': bool}
                if isinstance(res, dict):
                    accepted = bool(res.get("accepted", False))
                    is_duplicate = bool(res.get("duplicate", False))
                    if accepted and not is_duplicate:
                        # Only count blocks that are accepted AND not duplicates
                        self._stats.blocks_accepted += 1
                        self._cooldown.notify_block_accepted(
                            height=res.get("height"), block_hash=res.get("hash")
                        )
                    elif not accepted:
                        self._stats.blocks_rejected += 1
                    # Note: duplicate blocks (accepted but already found) are not counted
                    return res
                # Fallback
                if res is True:
                    self._stats.blocks_accepted += 1
                    self._cooldown.notify_block_accepted()
                    return {"accepted": True}
                self._stats.blocks_rejected += 1
                return {"accepted": False, "reason": "unexpected-result"}
            except RpcError as e:
                if e.code in (
                    -32020,
                    -32021,
                    -32022,
                ):  # e.g., InvalidBlock, BadProofs, Stale
                    self._stats.blocks_rejected += 1
                    return {"accepted": False, "reason": f"rpc:{e.code}:{e}"}
                self._stats.last_error = f"{e.code}:{e}"
            except TransportError as e:
                self._stats.last_error = str(e)

            if tries >= self.cfg.max_retries:
                return {
                    "accepted": False,
                    "reason": f"retries-exhausted:{self._stats.last_error or ''}",
                }

            sleep = backoff * (1.0 + (random.random() * 2 - 1) * self.cfg.jitter)
            sleep = max(0.0, min(sleep, self.cfg.max_backoff_s))
            self._log.debug(
                "block submit retry",
                tries=tries,
                sleep=sleep,
                last_error=self._stats.last_error,
            )
            time.sleep(sleep)
            backoff = min(backoff * 2.0, self.cfg.max_backoff_s)

    async def submit_block(self, candidate_block: Any) -> Dict[str, Any]:
        payload = json_validate(self._block_encoder(candidate_block))
        if not isinstance(payload, dict):
            raise ValueError("block payload must be an object")
        backoff = self.cfg.initial_backoff_s
        tries = 0
        while True:
            tries += 1
            t0 = time.perf_counter()
            try:
                res = await self._async_rpc.call(
                    self.cfg.method_submit_block, [payload]
                )
                dt = time.perf_counter() - t0
                if isinstance(res, dict):
                    accepted = bool(res.get("accepted", False))
                    is_duplicate = bool(res.get("duplicate", False))
                    if accepted and not is_duplicate:
                        # Only count blocks that are accepted AND not duplicates
                        self._stats.blocks_accepted += 1
                        self._cooldown.notify_block_accepted(
                            height=res.get("height"), block_hash=res.get("hash")
                        )
                    elif not accepted:
                        self._stats.blocks_rejected += 1
                    # Note: duplicate blocks (accepted but already found) are not counted
                    self._log.info(
                        "submitBlock result accepted=%s duplicate=%s reason=%s latency=%.3fs",
                        accepted,
                        is_duplicate,
                        res.get("reason"),
                        dt,
                    )
                    return res
                if res is True:
                    self._stats.blocks_accepted += 1
                    self._cooldown.notify_block_accepted()
                    self._log.info(
                        "submitBlock result accepted=True latency=%.3fs", dt
                    )
                    return {"accepted": True}
                self._stats.blocks_rejected += 1
                self._log.info(
                    "submitBlock result accepted=False reason=unexpected-result latency=%.3fs",
                    dt,
                )
                return {"accepted": False, "reason": "unexpected-result"}
            except RpcError as e:
                if e.code in (
                    -32020,
                    -32021,
                    -32022,
                ):  # InvalidBlock, BadProofs, Stale
                    self._stats.blocks_rejected += 1
                    return {"accepted": False, "reason": f"rpc:{e.code}:{e}"}
                self._stats.last_error = f"{e.code}:{e}"
            except TransportError as e:
                self._stats.last_error = str(e)

            if tries >= self.cfg.max_retries:
                return {
                    "accepted": False,
                    "reason": f"retries-exhausted:{self._stats.last_error or ''}",
                }

            sleep = backoff * (1.0 + (random.random() * 2 - 1) * self.cfg.jitter)
            sleep = max(0.0, min(sleep, self.cfg.max_backoff_s))
            self._log.warning(
                "submitBlock retry in %.2fs (try %s/%s): %s",
                sleep,
                tries,
                self.cfg.max_retries,
                self._stats.last_error,
            )
            await asyncio.sleep(sleep)
            backoff = min(backoff * 2.0, self.cfg.max_backoff_s)

    # ──────────────────────────────────────────────────────────────────────
    # Background consumer from a ShareBuffer
    # ──────────────────────────────────────────────────────────────────────

    def start_consumer(
        self,
        buffer: "ShareBufferLike",
        *,
        name: str = "ShareSubmitter",
        max_items_per_batch: Optional[int] = None,
        poll_timeout_s: float = 0.05,
    ) -> None:
        """
        Start a background thread that continuously drains from `buffer` and submits shares.

        `buffer` must expose:
            - pop_batch(max_items: int, timeout: float) -> list[Any]
            - stats() -> object (optional, only for debug logs)
        """
        if self._thread and self._thread.is_alive():
            return
        self._stop_evt.clear()
        batch = max_items_per_batch or self.cfg.batch_size

        def _run() -> None:
            self._log.info("share submitter started", url=self.cfg.rpc_url, batch=batch)
            while not self._stop_evt.is_set():
                try:
                    shares = buffer.pop_batch(max_items=batch, timeout=poll_timeout_s)
                except Exception as e:  # defensive; buffer should not raise
                    self._log.exception("buffer.pop_batch failed: %r", e)
                    time.sleep(0.1)
                    continue
                if not shares:
                    continue
                # Try batch submit first; fall back to one-by-one if unsupported
                encoded = []
                try:
                    for s in shares:
                        encoded.append(json_validate(self._share_encoder(s)))
                except Exception as e:
                    # If any share can't be encoded, fall back to per-share processing
                    self._log.warning(
                        "share encoder error; falling back per-share: %r", e
                    )
                    encoded = []

                if encoded:
                    if not all(isinstance(item, dict) for item in encoded):
                        self._log.warning(
                            "share encoder produced non-object payload; falling back per-share"
                        )
                        encoded = []
                    try:
                        results = self._submit_batch(encoded)
                        # Update stats from results
                        for r in results:
                            if isinstance(r, dict) and r.get("accepted") is True:
                                self._stats.shares_accepted += 1
                            elif r is True:
                                self._stats.shares_accepted += 1
                            else:
                                self._stats.shares_rejected += 1
                        continue
                    except RpcError as e:
                        if e.code != -32601:
                            # Batch supported but failed; treat as transient and retry individually
                            self._log.debug(
                                "batch submit failed; falling back per-share: %s", e
                            )
                        # else: method not found → permanent fallback below
                    except TransportError as e:
                        self._log.debug(
                            "transport error on batch; falling back per-share: %s", e
                        )
                        # fall through to per-share loop

                # Per-share submissions with individual retries
                for s in shares:
                    try:
                        res = self.submit_share_once(s)
                        # (stats already updated in submit_share_once)
                        if not res.get("accepted", False):
                            self._log.debug(
                                "share rejected", reason=res.get("reason", "")
                            )
                    except Exception as e:
                        self._stats.shares_errors += 1
                        self._stats.last_error = repr(e)
                        self._log.debug("share submit error: %s", e)

            self._log.info("share submitter stopped")

        self._thread = threading.Thread(target=_run, name=name, daemon=True)
        self._thread.start()

    def stop_consumer(self, join: bool = True, timeout: Optional[float] = 5.0) -> None:
        self._stop_evt.set()
        if self._thread and join:
            self._thread.join(timeout=timeout)

    # ──────────────────────────────────────────────────────────────────────
    # Internals
    # ──────────────────────────────────────────────────────────────────────

    def _submit_batch(self, encoded_shares: List[Dict[str, Any]]) -> List[Any]:
        """
        Attempt batch submit via miner.submitShareBatch. Raises RpcError/TransportError on failure.
        """
        return self.rpc.call(self.cfg.method_submit_share_batch, [encoded_shares])


# Protocol (duck type) for ShareBuffer-like objects
class ShareBufferLike:
    def pop_batch(
        self, max_items: int = 1024, timeout: float = 0.0
    ) -> List[Any]:  # pragma: no cover - interface
        raise NotImplementedError

    def stats(self) -> Any:  # pragma: no cover - interface
        raise NotImplementedError


# ─────────────────────────────────────────────────────────────────────────────
# CLI for quick manual testing
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":  # pragma: no cover
    import argparse
    import sys

    ap = argparse.ArgumentParser(
        description="Submit a single share or block to the local node RPC."
    )
    ap.add_argument(
        "--rpc", default="http://127.0.0.1:8545/rpc", help="JSON-RPC endpoint"
    )
    ap.add_argument("--timeout", type=float, default=5.0, help="HTTP timeout (s)")
    ap.add_argument(
        "--headers",
        type=str,
        default="",
        help='Extra headers JSON, e.g. \'{"Authorization":"Bearer …"}\'',
    )
    sub = ap.add_subparsers(dest="cmd", required=True)

    sp1 = sub.add_parser("share", help="Submit a single share JSON file")
    sp1.add_argument(
        "--file",
        required=True,
        help="Path to share JSON (payload with header/nonce/proof)",
    )

    sp2 = sub.add_parser("block", help="Submit a candidate block JSON file")
    sp2.add_argument(
        "--file", required=True, help="Path to block JSON (header/txs/proofs)"
    )

    args = ap.parse_args()
    headers = {}
    if args.headers:
        try:
            headers = json.loads(args.headers)
        except Exception as e:
            print(f"Invalid headers JSON: {e}", file=sys.stderr)
            sys.exit(2)

    cfg = SubmitterConfig(
        rpc_url=args.rpc, timeout_s=args.timeout, http_headers=headers
    )
    subm = ShareSubmitter(cfg)

    try:
        with open(args.file, "rb") as fh:
            obj = json.loads(fh.read())
    except Exception as e:
        print(f"Failed to read JSON: {e}", file=sys.stderr)
        sys.exit(2)

    if args.cmd == "share":
        res = subm.submit_share_once(obj)
        print(json.dumps(res, indent=2))
    else:
        res = subm.submit_block_once(obj)
        print(json.dumps(res, indent=2))
