from __future__ import annotations

import json
import logging
import os
import sys
import time
from dataclasses import dataclass, field
from typing import Any

LOGGER = logging.getLogger("fivetran_sync")
fivetran_log: Any | None
try:  # pragma: no cover - exercised by fivetran debug, not pure unit tests.
    from fivetran_connector_sdk import Logging as _fivetran_log
except Exception:
    fivetran_log = None
else:
    fivetran_log = _fivetran_log


def structured_log(level: int, message: str, **fields: Any) -> None:
    payload = {"message": message, **fields}
    encoded = json.dumps(payload, sort_keys=True, default=str)
    if fivetran_log is not None and getattr(fivetran_log, "LOG_LEVEL", None) is not None:
        if level >= logging.ERROR:
            fivetran_log.error(encoded)
        elif level >= logging.WARNING:
            fivetran_log.warning(encoded)
        elif level <= logging.DEBUG:
            fivetran_log.debug(encoded)
        else:
            fivetran_log.info(encoded)
        return
    LOGGER.log(level, encoded)


def info(message: str, **fields: Any) -> None:
    structured_log(logging.INFO, message, **fields)


def warning(message: str, **fields: Any) -> None:
    structured_log(logging.WARNING, message, **fields)


def info_with_memory(message: str, **fields: Any) -> None:
    info(message, **_add_process_memory_fields(fields))


def warning_with_memory(message: str, **fields: Any) -> None:
    warning(message, **_add_process_memory_fields(fields))


def timer_start() -> float:
    return time.perf_counter()


def elapsed_ms(start: float) -> float:
    return round((time.perf_counter() - start) * 1000, 3)


def process_memory_fields() -> dict[str, float]:
    fields: dict[str, float] = {}
    rss_mb = _current_rss_mb()
    if rss_mb is not None:
        fields["process_rss_mb"] = rss_mb
    peak_rss_mb = _peak_rss_mb()
    if peak_rss_mb is not None:
        fields["process_peak_rss_mb"] = peak_rss_mb
    return fields


def _add_process_memory_fields(fields: dict[str, Any]) -> dict[str, Any]:
    result = fields.copy()
    result.update(process_memory_fields())
    return result


def _current_rss_mb() -> float | None:
    if not sys.platform.startswith("linux"):
        return None
    try:
        with open("/proc/self/statm", encoding="utf-8") as statm:
            parts = statm.read().split()
    except OSError:
        return None
    if len(parts) < 2:
        return None
    try:
        resident_pages = int(parts[1])
        page_size = os.sysconf("SC_PAGE_SIZE")
    except (OSError, TypeError, ValueError):
        return None
    return round((resident_pages * page_size) / (1024 * 1024), 3)


def _peak_rss_mb() -> float | None:
    try:
        import resource
    except ImportError:  # pragma: no cover - resource is unavailable on Windows.
        return None
    usage = resource.getrusage(resource.RUSAGE_SELF)
    if usage.ru_maxrss <= 0:
        return None
    if sys.platform == "darwin":
        return round(usage.ru_maxrss / (1024 * 1024), 3)
    return round(usage.ru_maxrss / 1024, 3)


@dataclass
class OperationStats:
    counts: dict[str, int] = field(default_factory=dict)
    durations_ms: dict[str, float] = field(default_factory=dict)

    def record(self, operation: str, duration_ms: float) -> None:
        self.record_many(operation, count=1, duration_ms=duration_ms)

    def record_many(self, operation: str, *, count: int, duration_ms: float) -> None:
        if count <= 0:
            return
        self.counts[operation] = self.counts.get(operation, 0) + count
        self.durations_ms[operation] = self.durations_ms.get(operation, 0.0) + duration_ms

    def merge(self, other: OperationStats) -> None:
        for operation, count in other.counts.items():
            self.counts[operation] = self.counts.get(operation, 0) + count
        for operation, duration in other.durations_ms.items():
            self.durations_ms[operation] = self.durations_ms.get(operation, 0.0) + duration

    @property
    def total_count(self) -> int:
        return sum(self.counts.values())

    @property
    def total_duration_ms(self) -> float:
        return round(sum(self.durations_ms.values()), 3)

    def as_log_fields(self, prefix: str = "operation_") -> dict[str, Any]:
        fields: dict[str, Any] = {
            f"{prefix}total_count": self.total_count,
            f"{prefix}total_duration_ms": self.total_duration_ms,
        }
        for operation in sorted(set(self.counts) | set(self.durations_ms)):
            fields[f"{prefix}{operation}_count"] = self.counts.get(operation, 0)
            fields[f"{prefix}{operation}_duration_ms"] = round(self.durations_ms.get(operation, 0.0), 3)
        return fields


__all__ = [
    "OperationStats",
    "elapsed_ms",
    "info",
    "info_with_memory",
    "process_memory_fields",
    "structured_log",
    "timer_start",
    "warning",
    "warning_with_memory",
]