"""Durable robot and camera time-series publishing."""

from __future__ import annotations

import base64
import hashlib
import math
import mimetypes
import json
import re
import threading
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, Mapping, Optional

from .http import HTTPResponse, TransportError, UrllibTransport, json_body
from .outbox import DurableOutbox, OutboxCapacityError, OutboxEvent

SCHEMA_VERSION = "labcobot.timeseries.v1"
STATUS_SCHEMA_VERSION = 1
CAPTURE_DATASET_SCHEMA = "labcobot.capture-dataset.v1"
SUPPORTED_IMAGE_TYPES = frozenset({"image/jpeg", "image/png", "image/webp"})
SAFE_CAMERA_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,255}$")


@dataclass(frozen=True)
class FlushResult:
    """Summary of one outbox flush."""

    attempted: int = 0
    delivered: int = 0
    deferred: int = 0
    dead_lettered: int = 0


class DataHubPublisher:
    """Queue and deliver authenticated device events to Supraworx.

    The server can deduplicate retries with the event id supplied both in the
    payload and the ``Idempotency-Key`` header.
    """

    def __init__(
        self,
        *,
        ingest_url: str,
        uid_device: str,
        api_secret: str,
        outbox: DurableOutbox,
        manufacturer_id: str = "",
        transport: Optional[UrllibTransport] = None,
        timeout: float = 15.0,
        max_image_bytes: int = 8 * 1024 * 1024,
        retry_base_seconds: float = 2.0,
        retry_max_seconds: float = 300.0,
        allow_insecure_http: bool = False,
        expected_dataset_asset_id: int = 0,
    ) -> None:
        if not ingest_url:
            raise ValueError("DataHub ingest URL is required")
        if not allow_insecure_http and not ingest_url.lower().startswith("https://"):
            raise ValueError("DataHub ingest URL must use HTTPS")
        if not uid_device:
            raise ValueError("uid_device is required")
        if not api_secret:
            raise ValueError("api_secret is required")
        if (
            isinstance(expected_dataset_asset_id, bool)
            or not isinstance(expected_dataset_asset_id, int)
            or expected_dataset_asset_id < 0
        ):
            raise ValueError("expected_dataset_asset_id must be a non-negative integer")
        self.ingest_url = ingest_url
        self.uid_device = uid_device
        self.api_secret = api_secret
        self.manufacturer_id = manufacturer_id
        self.outbox = outbox
        self.transport = transport or UrllibTransport()
        self.timeout = float(timeout)
        self.max_image_bytes = int(max_image_bytes)
        self.retry_base_seconds = float(retry_base_seconds)
        self.retry_max_seconds = float(retry_max_seconds)
        self.expected_dataset_asset_id = int(expected_dataset_asset_id)
        self._status_lock = threading.Lock()
        self._dataset_binding_verified = False
        self._dataset_binding_mismatch = False
        self._last_error_code = ""

    def enqueue_metrics(
        self,
        metrics: Mapping[str, Any],
        *,
        observed_at: Optional[datetime] = None,
        attributes: Optional[Mapping[str, Any]] = None,
        event_id: Optional[str] = None,
    ) -> str:
        """Persist a robot metric sample before attempting network delivery."""
        if not isinstance(metrics, Mapping) or not metrics:
            raise ValueError("metrics must be a non-empty mapping")
        values = _json_safe(dict(metrics))
        timestamp = _utc_timestamp(observed_at)
        identifier = event_id or str(uuid.uuid4())
        payload = self._base_payload(
            event_id=identifier,
            series="robot.metrics",
            observed_at=timestamp,
            attributes=attributes,
        )
        payload["values"] = values
        self.outbox.enqueue(
            identifier,
            "metrics",
            timestamp,
            payload,
        )
        return identifier

    def enqueue_image(
        self,
        image: bytes,
        *,
        camera_id: str,
        content_type: str = "image/jpeg",
        observed_at: Optional[datetime] = None,
        attributes: Optional[Mapping[str, Any]] = None,
        event_id: Optional[str] = None,
    ) -> str:
        """Persist a camera frame as a time-series event."""
        content_type = str(content_type).lower().split(";", 1)[0].strip()
        if content_type not in SUPPORTED_IMAGE_TYPES:
            raise ValueError("Unsupported camera image content type")
        image = bytes(image)
        if not image:
            raise ValueError("Camera image is empty")
        if len(image) > self.max_image_bytes:
            raise ValueError("Camera image exceeds configured maximum size")
        camera_id = str(camera_id).strip()
        if not SAFE_CAMERA_ID.fullmatch(camera_id):
            raise ValueError(
                "camera_id must contain only letters, numbers, '.', '_', ':', "
                "or '-' and be at most 255 characters"
            )

        timestamp = _utc_timestamp(observed_at)
        identifier = event_id or str(uuid.uuid4())
        payload = self._base_payload(
            event_id=identifier,
            series="camera.image",
            observed_at=timestamp,
            attributes=attributes,
        )
        payload["camera"] = {
            "camera_id": camera_id,
            "content_type": content_type,
            "byte_length": len(image),
            "sha256": hashlib.sha256(image).hexdigest(),
            "encoding": "base64",
        }
        self.outbox.enqueue(
            identifier,
            "image",
            timestamp,
            payload,
            blob=image,
        )
        return identifier

    def enqueue_image_file(
        self,
        path: Path,
        *,
        camera_id: str,
        content_type: str = "",
        observed_at: Optional[datetime] = None,
        attributes: Optional[Mapping[str, Any]] = None,
        event_id: Optional[str] = None,
    ) -> str:
        """Read and queue one bounded image file."""
        image_path = Path(path)
        size = image_path.stat().st_size
        if size > self.max_image_bytes:
            raise ValueError("Camera image exceeds configured maximum size")
        guessed_type = mimetypes.guess_type(image_path.name)[0] or "image/jpeg"
        with image_path.open("rb") as handle:
            image = handle.read(self.max_image_bytes + 1)
        return self.enqueue_image(
            image,
            camera_id=camera_id,
            content_type=content_type or guessed_type,
            observed_at=observed_at,
            attributes=attributes,
            event_id=event_id,
        )

    def flush(self, limit: int = 25) -> FlushResult:
        """Attempt delivery of currently due events."""
        events = self.outbox.claim(limit=limit)
        delivered = 0
        deferred = 0
        dead_lettered = 0
        for event in events:
            try:
                response = self._send(event)
            except TransportError as exc:
                self._set_delivery_error("transport_error")
                self.outbox.retry(
                    event.event_id,
                    str(exc),
                    self._retry_delay(event.attempts),
                )
                deferred += 1
                continue

            receipt = _json_mapping(response)
            successful = 200 <= response.status_code < 300
            duplicate = _is_duplicate_acknowledgement(
                response,
                event.event_id,
                payload=receipt,
            )
            if successful or duplicate:
                asset_id = _dataset_asset_id(receipt)
                if (
                    self.expected_dataset_asset_id
                    and asset_id != self.expected_dataset_asset_id
                ):
                    self._set_dataset_mismatch()
                    try:
                        self.outbox.dead_letter(
                            event.event_id,
                            "DataHub receipt did not confirm the expected dataset",
                        )
                    except OutboxCapacityError:
                        self._set_outbox_error("dead_letter_capacity")
                        self.outbox.retry(
                            event.event_id,
                            "dead_letter_capacity",
                            self._retry_delay(event.attempts),
                        )
                        deferred += 1
                    else:
                        dead_lettered += 1
                else:
                    try:
                        self.outbox.acknowledge(
                            event.event_id,
                            status_code=response.status_code,
                            dataset_asset_id=asset_id,
                            retain_receipt=_is_capture_frame_event(event),
                        )
                    except OutboxCapacityError:
                        self._set_outbox_error("receipt_capacity")
                        self.outbox.retry(
                            event.event_id,
                            "receipt_capacity",
                            self._retry_delay(event.attempts),
                        )
                        deferred += 1
                    else:
                        self._record_verified_receipt(asset_id)
                        delivered += 1
            elif _retryable_status(response.status_code):
                self._set_delivery_error("http_retryable")
                self.outbox.retry(
                    event.event_id,
                    "DataHub returned HTTP {}".format(response.status_code),
                    self._retry_delay(event.attempts),
                )
                deferred += 1
            else:
                self._set_delivery_error("http_rejected")
                try:
                    self.outbox.dead_letter(
                        event.event_id,
                        "DataHub rejected event with HTTP {}".format(
                            response.status_code
                        ),
                    )
                except OutboxCapacityError:
                    self._set_outbox_error("dead_letter_capacity")
                    self.outbox.retry(
                        event.event_id,
                        "dead_letter_capacity",
                        self._retry_delay(event.attempts),
                    )
                    deferred += 1
                else:
                    dead_lettered += 1
        return FlushResult(
            attempted=len(events),
            delivered=delivered,
            deferred=deferred,
            dead_lettered=dead_lettered,
        )

    def run_worker(
        self,
        stop_event: threading.Event,
        *,
        interval_seconds: float = 5.0,
        batch_size: int = 25,
        pause_predicate: Optional[Callable[[], bool]] = None,
        operation_lock: Any = None,
    ) -> None:
        """Flush until ``stop_event`` is set."""
        if pause_predicate is not None and not callable(pause_predicate):
            raise ValueError("pause_predicate must be callable")
        if operation_lock is not None and not all(
            callable(getattr(operation_lock, name, None))
            for name in ("acquire", "release")
        ):
            raise ValueError("operation_lock must provide acquire/release")
        while not stop_event.is_set():
            if pause_predicate is not None:
                try:
                    paused = pause_predicate()
                except Exception:
                    self._set_delivery_error("pause_predicate_error")
                    stop_event.wait(0.25)
                    continue
                if not isinstance(paused, bool):
                    self._set_delivery_error("pause_predicate_invalid")
                    stop_event.wait(0.25)
                    continue
                if paused:
                    stop_event.wait(0.25)
                    continue
            leased = False
            if operation_lock is not None:
                try:
                    leased = operation_lock.acquire(blocking=False)
                except Exception:
                    self._set_delivery_error("operation_lock_error")
                    stop_event.wait(0.25)
                    continue
                if type(leased) is not bool:
                    self._set_delivery_error("operation_lock_invalid")
                    stop_event.wait(0.25)
                    continue
                if not leased:
                    stop_event.wait(0.25)
                    continue
            try:
                result = self.flush(limit=batch_size)
            finally:
                if leased:
                    operation_lock.release()
            if result.attempted == 0:
                stop_event.wait(max(0.1, interval_seconds))
            elif operation_lock is not None:
                # Yield after each one-event lease so a local capture request
                # cannot be starved by a continuously non-empty outbox.
                stop_event.wait(0.01)

    def status(self) -> Mapping[str, Any]:
        """Return a credential- and URL-free DataHub binding status."""

        with self._status_lock:
            result = {
                "schema_version": STATUS_SCHEMA_VERSION,
                "configured": True,
                "expected_dataset_asset_id": self.expected_dataset_asset_id,
                "dataset_binding_required": bool(self.expected_dataset_asset_id),
                "dataset_binding_verified": self._dataset_binding_verified,
                "dataset_binding_mismatch": self._dataset_binding_mismatch,
                "error_code": self._last_error_code,
            }
        try:
            outbox_status = _bounded_outbox_status(self.outbox.stats())
        except Exception:
            outbox_status = _unavailable_outbox_status()
        result["outbox"] = dict(outbox_status)
        return result

    def _record_verified_receipt(self, asset_id: Optional[int]) -> None:
        with self._status_lock:
            if (
                self.expected_dataset_asset_id
                and asset_id == self.expected_dataset_asset_id
            ):
                self._dataset_binding_verified = True
            if not self._dataset_binding_mismatch:
                self._last_error_code = ""

    def _set_dataset_mismatch(self) -> None:
        with self._status_lock:
            self._dataset_binding_mismatch = True
            self._last_error_code = "dataset_asset_mismatch"

    def _set_delivery_error(self, code: str) -> None:
        with self._status_lock:
            if not self._dataset_binding_mismatch:
                self._last_error_code = str(code)[:64]

    def _set_outbox_error(self, code: str) -> None:
        with self._status_lock:
            self._last_error_code = str(code)[:64]

    def _base_payload(
        self,
        *,
        event_id: str,
        series: str,
        observed_at: str,
        attributes: Optional[Mapping[str, Any]],
    ) -> Dict[str, Any]:
        return {
            "schema_version": SCHEMA_VERSION,
            "event_id": event_id,
            "series": series,
            "observed_at": observed_at,
            "device": {
                "uid_device": self.uid_device,
                "manufacturer_id": self.manufacturer_id,
            },
            "attributes": _json_safe(dict(attributes or {})),
        }

    def _send(self, event: OutboxEvent) -> HTTPResponse:
        payload = dict(event.payload)
        if event.kind == "image":
            payload["image_data"] = base64.b64encode(event.blob).decode("ascii")

        # The existing registered-device log endpoint reads ``metrics`` while
        # the versioned envelope remains available for a dedicated DataHub
        # ingest endpoint.
        if event.kind == "metrics":
            payload["metrics"] = payload.get("values", {})
        else:
            payload["message"] = "camera image sample"

        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            "Idempotency-Key": event.event_id,
            "User-Agent": "LabCobot-DataHub/1",
            "X-Device-Id": self.uid_device,
            "X-Device-Secret": self.api_secret,
        }
        return self.transport.request(
            "POST",
            self.ingest_url,
            headers=headers,
            body=json_body(payload),
            timeout=self.timeout,
            max_response_bytes=1024 * 1024,
        )

    def _retry_delay(self, prior_attempts: int) -> float:
        exponent = min(max(0, int(prior_attempts)), 16)
        return min(
            self.retry_max_seconds,
            self.retry_base_seconds * (2**exponent),
        )


def _unavailable_outbox_status() -> Mapping[str, Any]:
    return {
        "schema_version": 1,
        "pending_count": 0,
        "queue_bytes": 0,
        "leased_count": 0,
        "oldest_lease_age_seconds": None,
        "retry_count": 0,
        "dead_letter_count": 0,
        "dead_letter_bytes": 0,
        "receipt_count": 0,
        "last_acknowledged_at": None,
        "last_acknowledgement_age_seconds": None,
        "capacity_ok": False,
    }


def _bounded_outbox_status(raw: Any) -> Mapping[str, Any]:
    if not isinstance(raw, Mapping) or raw.get("schema_version") != 1:
        return _unavailable_outbox_status()

    def count(name: str, maximum: int) -> Optional[int]:
        value = raw.get(name)
        if isinstance(value, bool) or not isinstance(value, int):
            return None
        if not 0 <= value <= maximum:
            return None
        return value

    def optional_number(name: str, maximum: float) -> Optional[float]:
        value = raw.get(name)
        if value is None:
            return None
        if isinstance(value, bool) or not isinstance(value, (int, float)):
            raise ValueError("invalid outbox number")
        parsed = float(value)
        if not math.isfinite(parsed) or not 0.0 <= parsed <= maximum:
            raise ValueError("invalid outbox number")
        return parsed

    pending = count("pending_count", 1_000_000_000)
    queue_bytes = count("queue_bytes", 10_000_000_000_000)
    leased = count("leased_count", 1_000_000_000)
    retries = count("retry_count", 1_000_000_000)
    dead = count("dead_letter_count", 1_000_000_000)
    dead_bytes = count("dead_letter_bytes", 10_000_000_000_000)
    receipts = count("receipt_count", 1_000_000_000)
    if (
        None
        in (
            pending,
            queue_bytes,
            leased,
            retries,
            dead,
            dead_bytes,
            receipts,
        )
        or type(raw.get("capacity_ok")) is not bool
    ):
        return _unavailable_outbox_status()
    return {
        "schema_version": 1,
        "pending_count": pending,
        "queue_bytes": queue_bytes,
        "leased_count": leased,
        "oldest_lease_age_seconds": optional_number(
            "oldest_lease_age_seconds", 315_576_000.0
        ),
        "retry_count": retries,
        "dead_letter_count": dead,
        "dead_letter_bytes": dead_bytes,
        "receipt_count": receipts,
        "last_acknowledged_at": optional_number(
            "last_acknowledged_at", 100_000_000_000.0
        ),
        "last_acknowledgement_age_seconds": optional_number(
            "last_acknowledgement_age_seconds", 315_576_000.0
        ),
        "capacity_ok": raw.get("capacity_ok") is True,
    }


def _utc_timestamp(value: Optional[datetime]) -> str:
    timestamp = value or datetime.now(timezone.utc)
    if timestamp.tzinfo is None:
        timestamp = timestamp.replace(tzinfo=timezone.utc)
    timestamp = timestamp.astimezone(timezone.utc)
    return timestamp.isoformat(timespec="milliseconds").replace("+00:00", "Z")


def _json_safe(value: Any) -> Any:
    if value is None or isinstance(value, (str, bool, int)):
        return value
    if isinstance(value, float):
        if not math.isfinite(value):
            raise ValueError("Time-series values cannot contain NaN or infinity")
        return value
    if isinstance(value, Mapping):
        return {str(key): _json_safe(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_json_safe(item) for item in value]
    raise ValueError("Time-series payload contains a non-JSON value")


def _retryable_status(status_code: int) -> bool:
    return status_code in {408, 425, 429} or 500 <= status_code <= 599


def _is_duplicate_acknowledgement(
    response: HTTPResponse,
    event_id: str,
    *,
    payload: Optional[Mapping[str, Any]] = None,
) -> bool:
    """Accept a conflict only when it identifies this event as a duplicate.

    Treating every HTTP 409 as success can silently discard an event when the
    conflict concerns a device, schema, or another idempotency key. The
    dedicated DataHub endpoint must echo the event id and an explicit duplicate
    status so a retry can be acknowledged safely.
    """

    if response.status_code != 409:
        return False
    if payload is None:
        payload = _json_mapping(response)
    if not isinstance(payload, Mapping):
        return False
    duplicate_status = str(payload.get("status") or "").strip().lower()
    returned_event_id = str(
        payload.get("event_id") or payload.get("idempotency_key") or ""
    )
    return returned_event_id == event_id and duplicate_status in {
        "duplicate",
        "already_recorded",
    }


def _json_mapping(response: HTTPResponse) -> Optional[Mapping[str, Any]]:
    try:
        payload = json.loads(response.body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    if not isinstance(payload, Mapping):
        return None
    return payload


def _dataset_asset_id(payload: Optional[Mapping[str, Any]]) -> Optional[int]:
    """Extract a strict SupraWorx dataset asset id from a receipt."""

    if payload is None:
        return None
    dataset = payload.get("dataset")
    if not isinstance(dataset, Mapping):
        return None
    value = dataset.get("asset_id")
    if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
        return None
    return value


def _is_capture_frame_event(event: OutboxEvent) -> bool:
    """Return true only for bounded deterministic capture-frame events."""

    if event.kind != "image" or event.payload.get("series") != "camera.image":
        return False
    attributes = event.payload.get("attributes")
    return (
        isinstance(attributes, Mapping)
        and attributes.get("capture_schema") == CAPTURE_DATASET_SCHEMA
    )
