"""Skill usage sidecar — per-skill telemetry stored in .usage.json."""

from __future__ import annotations

import json
import logging
import os
import tempfile
import threading
from collections import OrderedDict
from datetime import UTC, datetime
from pathlib import Path

logger = logging.getLogger(__name__)

_USAGE_FILENAME = ".usage.json"

# 串行化对 .usage.json 的"读-改-写",避免并发(尤其 curator 多线程)丢更新。
# .usage.json 是全局单文件;此锁覆盖进程内并发。leader 选举已使 curator 只在
# 单进程跑,故进程内锁即可保证状态流转不丢。
_usage_lock = threading.Lock()

# One user run can first load SKILL.md and then execute several helper scripts.
# Those are implementation details of one logical skill invocation, so retain a
# small process-local LRU keyed by (run, skill) and increment the public counter
# only once. A run stays on one worker; persistence across restarts is neither
# needed nor desirable here.
_MAX_RECENT_USE_EVENTS = 8192
_recent_use_events: OrderedDict[tuple[str, str], None] = OrderedDict()

STATE_ACTIVE = "active"
STATE_STALE = "stale"
STATE_ARCHIVED = "archived"

SOURCE_AGENT = "agent"
SOURCE_UPLOAD = "upload"

_DEFAULT_ENTRY: dict = {
    "use_count": 0,
    "view_count": 0,
    "patch_count": 0,
    "last_used_at": None,
    "last_viewed_at": None,
    "last_patched_at": None,
    "created_at": None,
    "state": STATE_ACTIVE,
    "pinned": False,
    "featured": False,
    "archived_at": None,
    "source": SOURCE_UPLOAD,
}


def _usage_path() -> Path:
    from deerflow.skills.storage import get_or_new_skill_storage

    return Path(get_or_new_skill_storage().get_skills_root_path()) / _USAGE_FILENAME


def _read_all() -> dict:
    path = _usage_path()
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        logger.debug("Failed to read %s", path, exc_info=True)
        return {}


def _write_all(data: dict) -> None:
    path = _usage_path()
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        tmp_fd, tmp_path = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
        try:
            with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
                json.dump(data, f, indent=2, default=str)
            os.replace(tmp_path, path)
        except Exception:
            try:
                os.unlink(tmp_path)
            except Exception:
                pass
            raise
    except Exception:
        logger.debug("Failed to write %s", path, exc_info=True)


def _now_iso() -> str:
    return datetime.now(UTC).isoformat()


def _update_entry(name: str, updates) -> None:
    """更新一个 usage 条目。``updates`` 可以是 dict,或 ``(entry) -> dict`` 回调。

    用回调形式可让"读当前值 -> 算增量"也落在锁内(例如计数自增),
    避免并发下的 lost update。
    """
    try:
        with _usage_lock:
            data = _read_all()
            entry = {**_DEFAULT_ENTRY, **data.get(name, {})}
            resolved = updates(entry) if callable(updates) else updates
            entry.update(resolved)
            data[name] = entry
            _write_all(data)
    except Exception:
        logger.debug("Failed to update usage entry for %r", name, exc_info=True)


def get_entry(name: str) -> dict:
    """Return the usage entry for *name*, filling missing fields with defaults."""
    data = _read_all()
    return {**_DEFAULT_ENTRY, **data.get(name, {})}


def all_entries() -> dict:
    """Return all usage entries keyed by skill name."""
    return _read_all()


def agent_created_entries() -> dict:
    """Return only entries where source == 'agent' (eligible for curation)."""
    return {k: v for k, v in _read_all().items() if v.get("source") == SOURCE_AGENT}


def bump_view(name: str) -> None:
    # 用回调:读当前计数 + 自增在同一把锁内完成,避免并发丢计数。
    _update_entry(name, lambda e: {"view_count": e.get("view_count", 0) + 1, "last_viewed_at": _now_iso()})


async def _increment_use_projection(name: str, use_count: int, used_at: datetime) -> None:
    """Keep the MySQL skill-list projection current without importing ``app``."""
    try:
        from sqlalchemy import case, update

        from deerflow.persistence.engine import get_session_factory
        from deerflow.persistence.skills.model import SkillRow

        session_factory = get_session_factory()
        if session_factory is None:
            return
        async with session_factory() as session:
            await session.execute(
                update(SkillRow)
                .where(SkillRow.name == name)
                .values(
                    # Fire-and-forget writes may finish out of order. Never let
                    # an older absolute sidecar count overwrite a newer value.
                    use_count=case((SkillRow.use_count < use_count, use_count), else_=SkillRow.use_count),
                    last_activity_at=case(
                        (SkillRow.last_activity_at.is_(None), used_at),
                        (SkillRow.last_activity_at < used_at, used_at),
                        else_=SkillRow.last_activity_at,
                    ),
                    updated_at=case(
                        (SkillRow.updated_at < used_at, used_at),
                        else_=SkillRow.updated_at,
                    ),
                )
            )
            await session.commit()
    except Exception:
        logger.debug("Failed to update skill usage projection for %r", name, exc_info=True)


def _schedule_use_projection_increment(name: str, use_count: int, used_at: datetime) -> None:
    """Best-effort bridge from synchronous sandbox tools to the DB home loop."""
    try:
        from deerflow.persistence.engine import fire_and_forget_db_write

        fire_and_forget_db_write(_increment_use_projection(name, use_count, used_at))
    except Exception:
        logger.debug("Failed to schedule skill usage projection for %r", name, exc_info=True)


def runtime_use_event_key(runtime: Any | None) -> str | None:
    """Return the stable user-run identifier carried by ``ToolRuntime``."""
    if runtime is None:
        return None
    context = getattr(runtime, "context", None)
    if isinstance(context, dict):
        for key in ("run_id", "langgraph_run_id"):
            value = context.get(key)
            if value is not None and str(value).strip():
                return str(value).strip()
    config = getattr(runtime, "config", None)
    if isinstance(config, dict):
        for bucket_name in ("configurable", "metadata"):
            bucket = config.get(bucket_name)
            if not isinstance(bucket, dict):
                continue
            for key in ("run_id", "langgraph_run_id"):
                value = bucket.get(key)
                if value is not None and str(value).strip():
                    return str(value).strip()
    return None


def bump_use(name: str, *, event_key: str | None = None) -> bool:
    """Count one logical use, optionally de-duplicated by user run.

    Returns ``True`` only when the counter changed. This makes callers and tests
    able to distinguish a new invocation from a second file/script access in
    the same run.
    """
    normalized_name = str(name or "").strip()
    if not normalized_name:
        return False
    normalized_event = str(event_key or "").strip() or None
    dedupe_key = (normalized_event, normalized_name) if normalized_event else None
    used_at = datetime.now(UTC)
    new_count = 0
    try:
        with _usage_lock:
            if dedupe_key is not None and dedupe_key in _recent_use_events:
                _recent_use_events.move_to_end(dedupe_key)
                return False
            data = _read_all()
            entry = {**_DEFAULT_ENTRY, **data.get(normalized_name, {})}
            new_count = max(0, int(entry.get("use_count") or 0)) + 1
            entry.update({"use_count": new_count, "last_used_at": used_at.isoformat()})
            data[normalized_name] = entry
            _write_all(data)
            if dedupe_key is not None:
                _recent_use_events[dedupe_key] = None
                _recent_use_events.move_to_end(dedupe_key)
                while len(_recent_use_events) > _MAX_RECENT_USE_EVENTS:
                    _recent_use_events.popitem(last=False)
    except Exception:
        logger.debug("Failed to bump usage entry for %r", normalized_name, exc_info=True)
        return False

    _schedule_use_projection_increment(normalized_name, new_count, used_at)
    return True


def bump_patch(name: str) -> None:
    _update_entry(name, lambda e: {"patch_count": e.get("patch_count", 0) + 1, "last_patched_at": _now_iso()})


def set_state(name: str, state: str) -> None:
    updates: dict = {"state": state}
    if state == STATE_ARCHIVED:
        updates["archived_at"] = _now_iso()
    elif state == STATE_ACTIVE:
        updates["archived_at"] = None
    _update_entry(name, updates)


def set_pinned(name: str, pinned: bool) -> None:
    _update_entry(name, {"pinned": pinned})


def set_featured(name: str, featured: bool) -> None:
    """置顶/取消置顶。与 pinned(防归档)正交,仅影响列表排序展示。"""
    _update_entry(name, {"featured": featured})


def forget(name: str) -> None:
    try:
        with _usage_lock:
            data = _read_all()
            data.pop(name, None)
            _write_all(data)
    except Exception:
        logger.debug("Failed to forget usage entry for %r", name, exc_info=True)


def mark_agent_created(name: str) -> None:
    """Record a skill as agent-created; sets source='agent' and initialises created_at."""
    _update_entry(name, {"source": SOURCE_AGENT, "created_at": _now_iso(), "state": STATE_ACTIVE})


def mark_uploaded(name: str) -> None:
    """Record a skill as user-uploaded; sets source='upload'."""
    _update_entry(name, {"source": SOURCE_UPLOAD, "created_at": _now_iso()})


def latest_activity_at(name: str) -> str | None:
    """Return the most recent activity timestamp, falling back to created_at."""
    entry = get_entry(name)
    candidates = [entry.get("last_used_at"), entry.get("last_viewed_at"), entry.get("last_patched_at")]
    valid = [ts for ts in candidates if ts]
    if valid:
        return max(valid)
    return entry.get("created_at")