from __future__ import annotations

import json
import os
from collections.abc import Callable
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

try:
    import keyring
except ImportError:
    keyring = None

from .config import get_profiles_path, normalize_base_url
from .json_types import JSONObject, KeyringBackend


KEYRING_SERVICE_NAME = "qingflow-mcp"
_UNSET = object()


def _utcnow() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass(slots=True)
class SessionProfile:
    profile: str
    base_url: str
    qf_version: str | None
    qf_version_source: str | None
    token: str | None
    login_token: str | None
    credential: str | None
    uid: int
    email: str | None
    nick_name: str | None
    selected_ws_id: int | None
    selected_ws_name: str | None
    persisted: bool
    created_at: str
    updated_at: str

    @classmethod
    def from_dict(cls, value: JSONObject) -> "SessionProfile":
        return cls(
            profile=value["profile"],
            base_url=value["base_url"],
            qf_version=value.get("qf_version"),
            qf_version_source=value.get("qf_version_source"),
            token=value.get("token"),
            login_token=value.get("login_token"),
            credential=value.get("credential"),
            uid=value["uid"],
            email=value.get("email"),
            nick_name=value.get("nick_name"),
            selected_ws_id=value.get("selected_ws_id"),
            selected_ws_name=value.get("selected_ws_name"),
            persisted=bool(value.get("persisted", False)),
            created_at=value.get("created_at", _utcnow()),
            updated_at=value.get("updated_at", _utcnow()),
        )


@dataclass(slots=True)
class BackendSession:
    token: str
    login_token: str | None
    credential: str | None
    profile: str
    base_url: str
    qf_version: str | None
    qf_version_source: str | None = None


class SessionStore:
    def __init__(self, base_dir: Path | None = None, keyring_backend: KeyringBackend | None = None) -> None:
        profiles_path = get_profiles_path() if base_dir is None else Path(base_dir) / "profiles.json"
        self._profiles_path = profiles_path
        self._profiles_path.parent.mkdir(parents=True, exist_ok=True)
        self._secrets_path = self._profiles_path.parent / "secrets.json"
        self._keyring = keyring_backend if keyring_backend is not None else keyring
        self._memory_sessions: dict[str, BackendSession] = {}
        self._logged_out_profiles: set[str] = set()
        self._profile_refresher: Callable[[str, SessionProfile | None], bool] | None = None
        self._refreshing_profiles: set[str] = set()

    def set_profile_refresher(self, refresher: Callable[[str, SessionProfile | None], bool] | None) -> None:
        self._profile_refresher = refresher

    def save_session(
        self,
        *,
        profile: str,
        base_url: str,
        qf_version: str | None,
        qf_version_source: str | None = None,
        token: str,
        login_token: str | None,
        credential: str | None = None,
        uid: int,
        email: str | None,
        nick_name: str | None,
        persist: bool,
    ) -> SessionProfile:
        now = _utcnow()
        previous = self.peek_profile(profile)
        persisted = False
        if persist:
            persisted = self._set_secret(self._token_key(profile), token)
            if login_token:
                self._set_secret(self._login_token_key(profile), login_token)
            else:
                self._delete_secret(self._login_token_key(profile))
            if credential:
                self._set_secret(self._credential_key(profile), credential)
            else:
                self._delete_secret(self._credential_key(profile))
        else:
            self._delete_secret(self._token_key(profile))
            self._delete_secret(self._login_token_key(profile))
            self._delete_secret(self._credential_key(profile))
        session_profile = SessionProfile(
            profile=profile,
            base_url=normalize_base_url(base_url) or base_url,
            qf_version=(str(qf_version).strip() or None) if qf_version is not None else None,
            qf_version_source=(str(qf_version_source).strip() or None) if qf_version_source is not None else None,
            token=str(token).strip() or None,
            login_token=(str(login_token).strip() or None) if login_token is not None else None,
            credential=(str(credential).strip() or None) if credential is not None else None,
            uid=uid,
            email=email,
            nick_name=nick_name,
            selected_ws_id=None,
            selected_ws_name=None,
            persisted=persisted,
            created_at=previous.created_at if previous else now,
            updated_at=now,
        )
        self._memory_sessions[profile] = BackendSession(
            token=session_profile.token or token,
            login_token=session_profile.login_token,
            credential=session_profile.credential,
            profile=profile,
            base_url=session_profile.base_url,
            qf_version=session_profile.qf_version,
            qf_version_source=session_profile.qf_version_source,
        )
        self._logged_out_profiles.discard(profile)
        self._upsert_profile(session_profile)
        return session_profile

    def get_profile(self, profile: str) -> SessionProfile | None:
        payload = self._load_profiles()
        raw_profile = payload.get("profiles", {}).get(profile)
        session_profile = SessionProfile.from_dict(raw_profile) if isinstance(raw_profile, dict) else None
        if profile in self._refreshing_profiles:
            return session_profile
        if self._should_refresh_profile(payload, session_profile) and self._refresh_profile(profile, session_profile):
            payload = self._load_profiles()
            raw_profile = payload.get("profiles", {}).get(profile)
            session_profile = SessionProfile.from_dict(raw_profile) if isinstance(raw_profile, dict) else None
        return session_profile

    def peek_profile(self, profile: str) -> SessionProfile | None:
        """读取 profile，但不触发缺失、过期会话的自动刷新。"""
        payload = self._load_profiles()
        raw_profile = payload.get("profiles", {}).get(profile)
        return SessionProfile.from_dict(raw_profile) if isinstance(raw_profile, dict) else None

    def get_backend_session(self, profile: str) -> BackendSession | None:
        if profile in self._logged_out_profiles:
            return None
        memory_session = self._memory_sessions.get(profile)
        session_profile = self.get_profile(profile)
        if memory_session:
            if session_profile is not None:
                memory_session.base_url = session_profile.base_url
                memory_session.qf_version = session_profile.qf_version
                memory_session.qf_version_source = session_profile.qf_version_source
            return memory_session
        if not session_profile:
            return None
        token = session_profile.token
        if not token and session_profile.persisted:
            token = self._get_secret(self._token_key(profile))
        if not token:
            return None
        login_token = session_profile.login_token
        if not login_token and session_profile.persisted:
            login_token = self._get_secret(self._login_token_key(profile))
        credential = session_profile.credential
        if not credential and session_profile.persisted:
            credential = self._get_secret(self._credential_key(profile))
        backend_session = BackendSession(
            token=token,
            login_token=login_token,
            credential=credential,
            profile=profile,
            base_url=session_profile.base_url,
            qf_version=session_profile.qf_version,
            qf_version_source=session_profile.qf_version_source,
        )
        self._memory_sessions[profile] = backend_session
        return backend_session

    def select_workspace(self, profile: str, ws_id: int, ws_name: str | None) -> SessionProfile:
        session_profile = self.get_profile(profile)
        if session_profile is None:
            raise KeyError(profile)
        session_profile.selected_ws_id = ws_id
        session_profile.selected_ws_name = ws_name
        session_profile.updated_at = _utcnow()
        self._upsert_profile(session_profile)
        return session_profile

    def update_route(self, profile: str, *, qf_version: str | None, qf_version_source: str | None) -> SessionProfile:
        session_profile = self.get_profile(profile)
        if session_profile is None:
            raise KeyError(profile)
        session_profile.qf_version = (str(qf_version).strip() or None) if qf_version is not None else None
        session_profile.qf_version_source = (str(qf_version_source).strip() or None) if qf_version_source is not None else None
        session_profile.updated_at = _utcnow()
        self._upsert_profile(session_profile)
        backend_session = self._memory_sessions.get(profile)
        if backend_session is not None:
            backend_session.qf_version = session_profile.qf_version
            backend_session.qf_version_source = session_profile.qf_version_source
        return session_profile

    def update_profile_metadata(
        self,
        profile: str,
        *,
        uid: int | object = _UNSET,
        email: str | None | object = _UNSET,
        nick_name: str | None | object = _UNSET,
        selected_ws_id: int | None | object = _UNSET,
        selected_ws_name: str | None | object = _UNSET,
    ) -> SessionProfile:
        session_profile = self.get_profile(profile)
        if session_profile is None:
            raise KeyError(profile)
        if uid is not _UNSET:
            session_profile.uid = int(uid)
        if email is not _UNSET:
            session_profile.email = email if isinstance(email, str) or email is None else session_profile.email
        if nick_name is not _UNSET:
            session_profile.nick_name = nick_name if isinstance(nick_name, str) or nick_name is None else session_profile.nick_name
        if selected_ws_id is not _UNSET:
            session_profile.selected_ws_id = (
                int(selected_ws_id)
                if isinstance(selected_ws_id, int) and not isinstance(selected_ws_id, bool)
                else None
            )
        if selected_ws_name is not _UNSET:
            session_profile.selected_ws_name = (
                selected_ws_name
                if isinstance(selected_ws_name, str) or selected_ws_name is None
                else session_profile.selected_ws_name
            )
        session_profile.updated_at = _utcnow()
        self._upsert_profile(session_profile)
        return session_profile

    def logout(self, profile: str, forget_persisted: bool = False) -> None:
        self._memory_sessions.pop(profile, None)
        if forget_persisted:
            self.invalidate(profile)
            return
        if self.get_profile(profile):
            self._logged_out_profiles.add(profile)

    def invalidate(self, profile: str) -> None:
        self._memory_sessions.pop(profile, None)
        self._logged_out_profiles.discard(profile)
        self._delete_secret(self._token_key(profile))
        self._delete_secret(self._login_token_key(profile))
        self._delete_secret(self._credential_key(profile))
        payload = self._load_profiles()
        profiles = payload.get("profiles", {})
        if profile in profiles:
            profiles.pop(profile)
            self._save_profiles(payload)

    def has_profile(self, profile: str) -> bool:
        return self.get_profile(profile) is not None

    def _token_key(self, profile: str) -> str:
        return f"{profile}:token"

    def _login_token_key(self, profile: str) -> str:
        return f"{profile}:login-token"

    def _credential_key(self, profile: str) -> str:
        return f"{profile}:credential"

    def _upsert_profile(self, profile: SessionProfile) -> None:
        payload = self._load_profiles()
        payload.setdefault("profiles", {})[profile.profile] = asdict(profile)
        self._save_profiles(payload)

    def _load_profiles(self) -> JSONObject:
        if not self._profiles_path.exists():
            return {"profiles": {}}
        try:
            with self._profiles_path.open("r", encoding="utf-8") as handle:
                payload = json.load(handle)
        except (OSError, json.JSONDecodeError):
            return {"profiles": {}}
        if not isinstance(payload, dict):
            return {"profiles": {}}
        profiles = payload.get("profiles")
        if not isinstance(profiles, dict):
            payload["profiles"] = {}
        return payload

    def _should_refresh_profile(self, payload: JSONObject, session_profile: SessionProfile | None) -> bool:
        if self._profile_refresher is None:
            return False
        profiles = payload.get("profiles")
        if not isinstance(profiles, dict) or not profiles:
            return True
        if session_profile is None:
            return True
        return self._is_profile_stale(session_profile)

    def _refresh_profile(self, profile: str, session_profile: SessionProfile | None) -> bool:
        if self._profile_refresher is None:
            return False
        self._refreshing_profiles.add(profile)
        try:
            return bool(self._profile_refresher(profile, session_profile))
        except Exception:
            return False
        finally:
            self._refreshing_profiles.discard(profile)

    def _is_profile_stale(self, session_profile: SessionProfile) -> bool:
        timestamp_text = session_profile.updated_at or session_profile.created_at
        try:
            timestamp = datetime.fromisoformat(timestamp_text)
        except (TypeError, ValueError):
            return True
        if timestamp.tzinfo is None:
            timestamp = timestamp.replace(tzinfo=timezone.utc)
        return datetime.now(timezone.utc) - timestamp > timedelta(days=7)

    def _save_profiles(self, payload: JSONObject) -> None:
        self._profiles_path.parent.mkdir(parents=True, exist_ok=True)
        with self._profiles_path.open("w", encoding="utf-8") as handle:
            json.dump(payload, handle, ensure_ascii=False, indent=2)

    def _set_secret(self, key: str, value: str) -> bool:
        if self._keyring is not None:
            try:
                self._keyring.set_password(KEYRING_SERVICE_NAME, key, value)
                self._delete_file_secret(key)
                return True
            except Exception:
                pass
        return self._set_file_secret(key, value)

    def _get_secret(self, key: str) -> str | None:
        if self._keyring is not None:
            try:
                value = self._keyring.get_password(KEYRING_SERVICE_NAME, key)
            except Exception:
                value = None
            if value:
                return value
        return self._get_file_secret(key)

    def _delete_secret(self, key: str) -> None:
        if self._keyring is not None:
            try:
                self._keyring.delete_password(KEYRING_SERVICE_NAME, key)
            except Exception:
                pass
        self._delete_file_secret(key)

    def _load_file_secrets(self) -> dict[str, str]:
        if not self._secrets_path.exists():
            return {}
        try:
            with self._secrets_path.open("r", encoding="utf-8") as handle:
                payload = json.load(handle)
        except (OSError, json.JSONDecodeError):
            return {}
        if not isinstance(payload, dict):
            return {}
        return {str(key): str(value) for key, value in payload.items() if isinstance(value, str)}

    def _save_file_secrets(self, payload: dict[str, str]) -> bool:
        self._secrets_path.parent.mkdir(parents=True, exist_ok=True)
        try:
            fd = os.open(self._secrets_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(payload, handle, ensure_ascii=False, indent=2)
            try:
                os.chmod(self._secrets_path, 0o600)
            except OSError:
                pass
            return True
        except OSError:
            return False

    def _set_file_secret(self, key: str, value: str) -> bool:
        payload = self._load_file_secrets()
        payload[key] = value
        return self._save_file_secrets(payload)

    def _get_file_secret(self, key: str) -> str | None:
        return self._load_file_secrets().get(key)

    def _delete_file_secret(self, key: str) -> None:
        payload = self._load_file_secrets()
        if key not in payload:
            return
        payload.pop(key, None)
        if payload:
            self._save_file_secrets(payload)
            return
        try:
            self._secrets_path.unlink()
        except FileNotFoundError:
            return
        except OSError:
            return