from __future__ import annotations

import json
import logging
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Callable

from niannian.storage.cloud_base import CloudBackend
from niannian.storage.cloud_dual import DualUploadResult
from niannian.storage.local_degraded import PendingUploadQueue
from niannian.storage.manifest import Manifest, utc_now_iso
from niannian.storage.reconcile_report import ReconcileReporter, sha256_hex
from niannian.storage.tier_chain import (
    build_role_map,
    provider_for_role,
    resolve_tier_chain,
)
from niannian.storage.tier_notifier import TierNotifier
from niannian.storage.tier_probe import DEFAULT_TIER_PROBE_TTL_HOURS, probe_is_fresh_at

logger = logging.getLogger(__name__)

PingFn = Callable[[CloudBackend], bool]
MANIFEST_PATH = "sync/manifest.json"
LOCAL_DEGRADED = "local_degraded"


@dataclass
class UploadResult:
    rel_path: str
    ok: bool
    active_role: str


@dataclass
class TierHealth:
    active_role: str
    quark_ok: bool
    baidu_ready: bool
    onedrive_ready: bool = False
    feimao_ready: bool = False
    message: str = "ok"
    degraded: bool = False


@dataclass
class ReconcileReport:
    ok: bool
    active_role: str = "quark"
    files_uploaded: list[str] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


def default_ping(backend: CloudBackend) -> bool:
    custom = getattr(backend, "tier_ping", None)
    if callable(custom):
        try:
            return bool(custom())
        except Exception:
            return False
    if not backend.is_configured():
        return False
    try:
        backend.download(MANIFEST_PATH)
        return True
    except Exception:
        return False


def read_tier_state(home: Path) -> dict[str, Any]:
    path = home / "sync" / "tier_state.json"
    if not path.is_file():
        return _default_tier_state()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        data.setdefault("active_role", "quark")
        data.setdefault("chain", list(resolve_tier_chain()))
        data.setdefault("providers", {})
        return data
    except (json.JSONDecodeError, OSError):
        return _default_tier_state()


def _default_tier_state() -> dict[str, Any]:
    chain = list(resolve_tier_chain())
    return {
        "active_role": "quark",
        "chain": chain,
        "providers": {},
        "failover_count": 0,
        "failover_since": None,
    }


def write_tier_state(home: Path, state: dict[str, Any]) -> None:
    path = home / "sync" / "tier_state.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


class CloudTierRouter:
    """三云路由：夸克 → 百度 → OneDrive failover + reconcile + TierNotifier。"""

    MANIFEST_PATH = MANIFEST_PATH

    def __init__(
        self,
        home: Path,
        quark: CloudBackend,
        baidu: CloudBackend,
        onedrive: CloudBackend,
        *,
        tier_cfg: dict[str, Any] | None = None,
        notify_cfg: dict[str, Any] | None = None,
        credential_manager: Any | None = None,
        ping_fn: PingFn | None = None,
    ) -> None:
        self._home = home
        self._quark = quark
        self._baidu = baidu
        self._onedrive = onedrive
        self._tier_cfg = tier_cfg or {}
        self._cred_mgr = credential_manager
        self._notifier = TierNotifier(home, notify_cfg or {})
        self._ping_fn = ping_fn or default_ping
        self._priority = resolve_tier_chain(self._tier_cfg)
        self._role_for = build_role_map(self._priority)
        self._provider_for = {role: name for name, role in self._role_for.items()}
        self._provider_for[LOCAL_DEGRADED] = "local"
        self._backends = {
            "quark": self._quark,
            "baidu": self._baidu,
            "onedrive": self._onedrive,
        }
        self._state = read_tier_state(home)
        self._last_degraded = False
        self._pending = PendingUploadQueue(home)
        self._draining = False
        self._reconcile = ReconcileReporter(home, {"storage": {"tier": self._tier_cfg}})

    @property
    def PRIORITY(self) -> tuple[str, ...]:
        return self._priority

    @property
    def quark(self) -> CloudBackend:
        return self._quark

    @property
    def baidu(self) -> CloudBackend:
        return self._baidu

    @property
    def onedrive(self) -> CloudBackend:
        return self._onedrive

    @property
    def feimao(self) -> CloudBackend:
        """Deprecated — v0.2.9 飞猫已退出 tier；保留属性避免旧调用方 AttributeError。"""
        return self._onedrive

    @property
    def notifier(self) -> TierNotifier:
        return self._notifier

    @property
    def last_upload_degraded(self) -> bool:
        return self._last_degraded or self.active_role() != "quark"

    @property
    def credential_manager(self) -> Any | None:
        return self._cred_mgr

    @property
    def pending_queue(self) -> PendingUploadQueue:
        return self._pending

    @property
    def tier_state(self) -> dict[str, Any]:
        return dict(self._state)

    def _backend(self, provider: str) -> CloudBackend:
        backend = self._backends.get(provider)
        if backend is None:
            raise KeyError(f"unknown tier provider: {provider}")
        return backend

    def _failover_eligible(self, provider: str) -> bool:
        backend = self._backend(provider)
        return backend.is_configured() or bool(getattr(backend, "token", None))

    def active_role(self) -> str:
        self._state = read_tier_state(self._home)
        return str(self._state.get("active_role", "quark"))

    def active_provider(self) -> str:
        role = self.active_role()
        if role == LOCAL_DEGRADED:
            return "local"
        return self._provider_for.get(role, provider_for_role(role))

    def active_backend(self) -> CloudBackend:
        role = self.active_role()
        if role == LOCAL_DEGRADED:
            return self._quark
        return self._backend(self.active_provider())

    def cloud_ready(self) -> bool:
        role = self.active_role()
        if role == LOCAL_DEGRADED:
            return True
        backend = self.active_backend()
        return backend.is_configured() or bool(getattr(backend, "token", None))

    def primary_mode(self) -> str:
        if self.active_role() != "quark":
            return "degraded"
        return self._quark.primary_mode()

    def backend_status(self) -> dict[str, str]:
        return {
            "quark": self._quark.status(),
            "baidu": self._baidu.status(),
            "onedrive": self._onedrive.status(),
            "realtime": self.active_role(),
            "mode": "tiered",
        }

    def _probe_ttl_hours(self) -> float:
        raw = self._tier_cfg.get("probe_ttl_hours")
        if raw is not None:
            return float(raw)
        return DEFAULT_TIER_PROBE_TTL_HOURS

    def _read_cached_ping(self, name: str) -> bool | None:
        ps = self._provider_state(name)
        last_at = ps.get("last_probe_at")
        if not probe_is_fresh_at(last_at, ttl_hours=self._probe_ttl_hours()):
            return None
        if "last_probe_ok" in ps:
            return bool(ps["last_probe_ok"])
        return ps.get("status") == "ok"

    def _persist_probe_result(self, name: str, ok: bool, *, record_failure: bool) -> None:
        ps = self._provider_state(name)
        ps["last_probe_at"] = utc_now_iso()
        ps["last_probe_ok"] = ok
        self._update_provider_ping(name, ok, record_failure=record_failure)

    def record_provider_failure(self, name: str) -> None:
        """实际云请求失败时立即标 offline，不等探针 TTL。"""
        self._state = read_tier_state(self._home)
        ps = self._provider_state(name)
        ps["status"] = "offline"
        ps["last_probe_ok"] = False
        ps["last_probe_at"] = utc_now_iso()
        ps["consecutive_failures"] = int(ps.get("consecutive_failures", 0)) + 1
        write_tier_state(self._home, self._state)

    def _resolve_provider_ping(
        self,
        name: str,
        *,
        force: bool = False,
        record_failure: bool = True,
    ) -> bool:
        if not force:
            cached = self._read_cached_ping(name)
            if cached is not None:
                return cached
        ok = self._ping_fn(self._backend(name))
        self._persist_probe_result(name, ok, record_failure=record_failure)
        return ok

    def _provider_state(self, name: str) -> dict[str, Any]:
        providers = self._state.setdefault("providers", {})
        return providers.setdefault(name, {"status": "unknown", "consecutive_failures": 0})

    def _update_provider_ping(self, name: str, ok: bool, *, record_failure: bool) -> None:
        ps = self._provider_state(name)
        if ok:
            was_offline = ps.get("status") == "offline"
            ps["status"] = "ok"
            ps["last_ok"] = utc_now_iso()
            ps["consecutive_failures"] = 0
            if was_offline:
                self._notifier.emit(
                    "provider_recovered",
                    provider=name,
                    severity="info",
                    message=f"{name} 已恢复",
                )
        elif record_failure:
            ps["status"] = "offline"
            ps["consecutive_failures"] = int(ps.get("consecutive_failures", 0)) + 1

    def _consecutive_failures(self, provider: str) -> int:
        return int(self._provider_state(provider).get("consecutive_failures", 0))

    def _resolve_active_role(self, pings: dict[str, bool]) -> str:
        threshold = int(self._tier_cfg.get("failover_after_failures", 3))
        chain = self._priority

        for i, name in enumerate(chain):
            if not pings.get(name):
                continue
            if i == 0:
                return self._role_for[name]
            prev_failed = all(self._consecutive_failures(chain[j]) >= threshold for j in range(i))
            if prev_failed and self._failover_eligible(name):
                return self._role_for[name]

        if len(chain) >= 2 and all(self._consecutive_failures(n) >= threshold for n in chain[:-1]):
            return LOCAL_DEGRADED
        return self._state.get("active_role", "quark")

    def check_health(
        self,
        *,
        record_failure: bool = True,
        dry_run: bool = False,
        force_probe: bool = False,
    ) -> TierHealth:
        self._state = read_tier_state(self._home)
        threshold = int(self._tier_cfg.get("failover_after_failures", 3))
        old_role = self.active_role()
        chain = self._priority

        pings: dict[str, bool] = {}
        for name in chain:
            ok = self._resolve_provider_ping(name, force=force_probe, record_failure=record_failure)
            pings[name] = ok
            if not dry_run and not ok and record_failure:
                ps = self._provider_state(name)
                if int(ps.get("consecutive_failures", 0)) >= threshold:
                    self._notifier.emit(
                        "provider_offline",
                        provider=name,
                        severity="warning",
                        message=f"{name} 连续 {threshold} 次不可达",
                    )

        new_role = self._resolve_active_role(pings)
        if dry_run:
            all_down = new_role == LOCAL_DEGRADED or (
                not any(pings.values())
                and all(self._consecutive_failures(n) >= threshold for n in chain)
            )
            role = new_role
            degraded = role != "quark" or all_down
            message = "ok"
            if role == LOCAL_DEGRADED:
                message = "local_degraded_queue_active"
            elif all_down:
                message = "all_providers_down"
            elif role == "baidu_failover":
                message = "quark_unreachable_using_baidu"
            elif role == "onedrive_failover":
                message = "quark_baidu_unreachable_using_onedrive"
            elif not pings.get("quark"):
                message = "quark_probe_failed"
            return TierHealth(
                active_role=role,
                quark_ok=bool(pings.get("quark")),
                baidu_ready=self._failover_eligible("baidu"),
                onedrive_ready=self._failover_eligible("onedrive"),
                message=message,
                degraded=degraded,
            )

        old_provider = self._provider_for.get(old_role, provider_for_role(old_role))
        new_provider = self._provider_for.get(new_role, provider_for_role(new_role))
        old_idx = chain.index(old_provider) if old_role != LOCAL_DEGRADED and old_provider in chain else len(chain)
        new_idx = chain.index(new_provider) if new_role != LOCAL_DEGRADED and new_provider in chain else len(chain)

        if old_role == LOCAL_DEGRADED and new_role != LOCAL_DEGRADED and not self._draining:
            drain = self._pending.drain(self)
            if drain.drained:
                self._notifier.emit(
                    "queue_drained",
                    severity="info",
                    message=f"已上传 {drain.drained} 条 pending 队列",
                )
            if self._tier_cfg.get("reconcile_on_recovery", True):
                self.reconcile_on_recovery()
            new_role = self.active_role()
        elif new_idx < old_idx and self._tier_cfg.get("reconcile_on_recovery", True):
            self.reconcile_on_recovery()
            new_role = self.active_role()
        elif new_role != old_role:
            if new_role not in ("quark", LOCAL_DEGRADED):
                self._state["failover_since"] = utc_now_iso()
                self._state["failover_count"] = int(self._state.get("failover_count", 0)) + 1
            else:
                self._state["failover_since"] = None
            self._state["active_role"] = new_role
            if new_role == LOCAL_DEGRADED:
                self._notifier.emit(
                    "all_clouds_unreachable",
                    severity="critical",
                    message="日常链耗尽；写队列落盘",
                )
            else:
                self._notifier.emit(
                    "failover",
                    severity="warning",
                    **{"from": old_role, "to": new_role},
                    message=f"主库 {old_role} → {new_role}",
                )
            logger.warning("tier failover %s → %s", old_role, new_role)

        all_down = new_role == LOCAL_DEGRADED or (
            not any(pings.values())
            and all(self._consecutive_failures(n) >= threshold for n in chain)
        )
        if all_down and new_role != LOCAL_DEGRADED:
            self._notifier.emit(
                "all_providers_down",
                severity="critical",
                message="三云均不可用",
            )

        write_tier_state(self._home, self._state)
        role = self.active_role()
        message = "ok"
        degraded = role != "quark" or all_down
        if role == LOCAL_DEGRADED:
            message = "local_degraded_queue_active"
        elif all_down:
            message = "all_providers_down"
        elif role == "baidu_failover":
            message = "quark_unreachable_using_baidu"
        elif role == "onedrive_failover":
            message = "quark_baidu_unreachable_using_onedrive"
        elif not pings.get("quark"):
            message = "quark_probe_failed"

        return TierHealth(
            active_role=role,
            quark_ok=bool(pings.get("quark")),
            baidu_ready=self._failover_eligible("baidu"),
            onedrive_ready=self._failover_eligible("onedrive"),
            message=message,
            degraded=degraded,
        )

    def reconcile_on_recovery(self) -> ReconcileReport:
        self._state = read_tier_state(self._home)
        current_role = self.active_role()
        chain = self._priority

        if current_role == LOCAL_DEGRADED:
            top_idx: int | None = None
            for i, name in enumerate(chain):
                if self._ping_fn(self._backend(name)) and self._failover_eligible(name):
                    top_idx = i
                    break
            if top_idx is None:
                return ReconcileReport(ok=False, active_role=current_role, errors=["no_recovered_provider"])
            target_role = self._role_for[chain[top_idx]]
            self._state["active_role"] = target_role
            self._state["failover_since"] = None
            write_tier_state(self._home, self._state)
            drain = self._pending.drain(self)
            errors = [] if drain.ok else (drain.errors or [])
            return ReconcileReport(
                ok=drain.ok,
                active_role=target_role,
                files_uploaded=[f"queue:{drain.drained}"],
                errors=errors,
            )

        source_provider = self._provider_for.get(current_role, provider_for_role(current_role))
        source_idx = chain.index(source_provider) if source_provider in chain else 0

        top_idx: int | None = None
        for i, name in enumerate(chain):
            if self._ping_fn(self._backend(name)) and self._failover_eligible(name):
                top_idx = i
                break

        if top_idx is None:
            return ReconcileReport(ok=False, active_role=current_role, errors=["no_recovered_provider"])

        target_provider = chain[top_idx]
        target_role = self._role_for[target_provider]
        if source_idx <= top_idx:
            self._state["active_role"] = target_role
            write_tier_state(self._home, self._state)
            return ReconcileReport(ok=True, active_role=target_role)

        errors: list[str] = []
        uploaded: list[str] = []
        for step in range(source_idx, top_idx, -1):
            lower = chain[step]
            upper = chain[step - 1]
            step_uploaded, step_errors = self._backfill(
                self._backend(lower),
                self._backend(upper),
                source_name=lower,
                target_name=upper,
            )
            uploaded.extend(step_uploaded)
            errors.extend(step_errors)

        self._state["active_role"] = target_role
        self._state["failover_since"] = None
        for name in chain[: top_idx + 1]:
            self._provider_state(name)["consecutive_failures"] = 0
        write_tier_state(self._home, self._state)
        return ReconcileReport(ok=not errors, active_role=target_role, files_uploaded=uploaded, errors=errors)

    def reconcile_after_quark_recovery(self) -> ReconcileReport:
        return self.reconcile_on_recovery()

    def _backfill(
        self,
        source: CloudBackend,
        target: CloudBackend,
        *,
        source_name: str = "source",
        target_name: str = "target",
    ) -> tuple[list[str], list[str]]:
        errors: list[str] = []
        uploaded: list[str] = []
        s_raw = source.download(self.MANIFEST_PATH)
        t_raw = target.download(self.MANIFEST_PATH)
        s_manifest = Manifest.from_dict(json.loads(s_raw.decode("utf-8"))) if s_raw else Manifest()
        t_manifest = Manifest.from_dict(json.loads(t_raw.decode("utf-8"))) if t_raw else Manifest()
        use_source = s_manifest.version >= t_manifest.version
        paths = set(s_manifest.files.keys()) | set(t_manifest.files.keys())
        for rel_path in (
            "left/atoms.jsonl",
            "cerebellum/pages.json",
            "hippocampus/summaries.json",
            "hippocampus/lifecycle.json",
            self.MANIFEST_PATH,
        ):
            if source.download(rel_path) is not None or target.download(rel_path) is not None:
                paths.add(rel_path)
        for rel_path in sorted(paths):
            s_data = source.download(rel_path)
            t_data = target.download(rel_path)
            if s_data and t_data and sha256_hex(s_data) != sha256_hex(t_data):
                reason = "version_equal_sha_diff" if s_manifest.version == t_manifest.version else "content_diverged"
                winner = source_name if use_source else target_name
                self._reconcile.record_conflict(
                    rel_path=rel_path,
                    source_name=source_name,
                    target_name=target_name,
                    source_sha=sha256_hex(s_data),
                    target_sha=sha256_hex(t_data),
                    source_version=s_manifest.version,
                    target_version=t_manifest.version,
                    winner=winner,
                    reason=reason,
                )
            if use_source and s_data:
                if t_data is None or sha256_hex(t_data) != sha256_hex(s_data):
                    if target.upload(rel_path, s_data):
                        uploaded.append(rel_path)
                    else:
                        errors.append(f"reconcile upload failed: {rel_path}")
            elif t_data and s_data is None:
                if target.upload(rel_path, t_data):
                    uploaded.append(rel_path)
        if use_source and s_raw and s_manifest.version > t_manifest.version:
            if not target.upload(self.MANIFEST_PATH, s_raw):
                errors.append("reconcile manifest upload failed")
        return uploaded, errors

    def upload(self, rel_path: str, data: bytes, *, skip_health: bool = False) -> UploadResult:
        if not skip_health and not self._draining:
            self.check_health(record_failure=True, force_probe=False)
        role = self.active_role()
        if role == LOCAL_DEGRADED:
            self._pending.enqueue(rel_path, data)
            self._last_degraded = True
            return UploadResult(rel_path=rel_path, ok=True, active_role=role)
        backend = self.active_backend()
        provider = self.active_provider()
        ok = backend.upload(rel_path, data)
        if not ok and provider in self._priority:
            self.record_provider_failure(provider)
        self._last_degraded = role != "quark" or bool(getattr(backend, "last_upload_degraded", False))
        return UploadResult(rel_path=rel_path, ok=ok, active_role=role)

    def download(self, rel_path: str) -> bytes | None:
        provider = self.active_provider()
        try:
            return self.active_backend().download(rel_path)
        except Exception:
            if provider in self._priority:
                self.record_provider_failure(provider)
            raise

    def dual_upload(self, rel_path: str, data: bytes) -> DualUploadResult:
        result = self.upload(rel_path, data)
        role = self.active_role()
        if role == "onedrive_failover":
            return DualUploadResult(
                rel_path=rel_path, quark_ok=False, baidu_ok=False, onedrive_ok=result.ok
            )
        if role == "baidu_failover":
            return DualUploadResult(rel_path=rel_path, quark_ok=False, baidu_ok=result.ok, onedrive_ok=False)
        if role == LOCAL_DEGRADED:
            return DualUploadResult(rel_path=rel_path, quark_ok=result.ok, baidu_ok=False, onedrive_ok=False)
        return DualUploadResult(rel_path=rel_path, quark_ok=result.ok, baidu_ok=False, onedrive_ok=False)

    def dual_download(self, rel_path: str) -> bytes | None:
        return self.download(rel_path)

    def read_remote_manifest(self) -> Manifest | None:
        raw = self.dual_download(self.MANIFEST_PATH)
        if raw is None:
            return None
        try:
            return Manifest.from_dict(json.loads(raw.decode("utf-8")))
        except (json.JSONDecodeError, KeyError, TypeError, ValueError):
            return None

    def upload_manifest(self, manifest: Manifest) -> DualUploadResult:
        data = (json.dumps(manifest.to_dict(), ensure_ascii=False, indent=2) + "\n").encode("utf-8")
        return self.dual_upload(self.MANIFEST_PATH, data)

    def manifest_versions(self) -> dict[str, int | None]:
        out: dict[str, int | None] = {name: None for name in self._priority}
        for name in self._priority:
            backend = self._backend(name)
            raw = backend.download(self.MANIFEST_PATH)
            if raw is None:
                continue
            try:
                out[name] = int(Manifest.from_dict(json.loads(raw.decode("utf-8"))).version)
            except (json.JSONDecodeError, KeyError, TypeError, ValueError):
                continue
        return out

    def fast_manifest_versions(self) -> dict[str, int | None]:
        """Status 路由用：不拉远端 manifest，仅返回占位结构。"""
        return {name: None for name in self._priority}

    def tier_summary(self, mirror_state: dict[str, Any] | None = None, *, live_probe: bool = True) -> dict[str, Any]:
        mirror = mirror_state or {}
        chain = self._priority
        self._state = read_tier_state(self._home)
        if live_probe:
            pings = {
                name: self._resolve_provider_ping(name, force=False, record_failure=False)
                for name in chain
            }
            write_tier_state(self._home, self._state)
        else:
            pings = {
                name: self._provider_state(name).get("status", "unknown") == "ok"
                for name in chain
            }
        onedrive_cfg = self._tier_cfg.get("onedrive") or {}
        summary: dict[str, Any] = {
            "active_role": self.active_role(),
            "chain": list(chain),
            "quark": {"status": "ok" if pings.get("quark") else "offline", "role": "primary", "mode": self._quark.primary_mode()},
            "baidu": {
                "status": "ok" if pings.get("baidu") else "offline",
                "role": "failover+mirror",
                "mirror_last": mirror.get("last_mirror_at"),
                "failover_ready": self._failover_eligible("baidu"),
            },
            "onedrive": {
                "status": "ok" if pings.get("onedrive") else "offline",
                "role": onedrive_cfg.get("role", "failover+mirror"),
                "mirror_last": mirror.get("onedrive_last"),
                "failover_ready": self._failover_eligible("onedrive"),
                "mirror_status": mirror.get("onedrive_status"),
            },
            "feimao": {"role": "disabled", "status": "retired"},
            "pending_uploads": self._pending.pending_count(),
            "queue_path": "sync/pending_uploads.jsonl",
            "failover_since": self._state.get("failover_since"),
            "failover_count": int(self._state.get("failover_count", 0)),
            "recent_alerts": self._notifier.recent_alerts(),
        }
        return summary
