"""MySQL-backed projection for the Gateway skill list endpoint."""

from __future__ import annotations

import asyncio
import json
import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import or_, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.schema import CreateColumn

from deerflow.config.app_config import AppConfig
from deerflow.persistence.engine import get_engine, get_session_factory
from deerflow.persistence.skills.model import SkillCatalogSnapshotRow, SkillRow
from deerflow.skills.storage import get_or_new_skill_storage

logger = logging.getLogger(__name__)
_CATALOG_CACHE_KEY = "global"
_catalog_warm_lock = asyncio.Lock()
_memory_catalog_snapshot: list[dict[str, Any]] | None = None
_memory_catalog_is_fallback = False


_SKILL_LIST_COLUMNS = (
    SkillRow.name,
    SkillRow.category,
    SkillRow.description,
    SkillRow.license,
    SkillRow.enabled,
    SkillRow.always_on,
    SkillRow.use_count,
    SkillRow.view_count,
    SkillRow.patch_count,
    SkillRow.last_activity_at,
    SkillRow.created_at,
    SkillRow.state,
    SkillRow.pinned,
    SkillRow.featured,
    SkillRow.source,
    SkillRow.owner_user_id,
    SkillRow.published,
    SkillRow.name_zh,
    SkillRow.square_id,
    SkillRow.detail_preview,
)

_SKILL_CATALOG_COLUMNS = {
    "category",
    "description",
    "license",
    "enabled",
    "available",
    "use_count",
    "view_count",
    "patch_count",
    "last_activity_at",
    "state",
    "pinned",
    "featured",
    "source",
    "source_mtime_ns",
    "detail_preview",
}


async def ensure_skill_catalog_schema() -> None:
    """Add compact catalog columns to an existing MySQL/SQLite skills table.

    ``create_all`` creates new tables but intentionally does not alter existing
    ones. These additive columns all have safe defaults (or are nullable), so
    this migration preserves existing ownership/detail data and is repeatable.
    """

    engine = get_engine()
    if engine is None:
        return

    def _ensure(sync_conn) -> None:
        from sqlalchemy import func, inspect, update

        table = SkillRow.__table__
        inspector = inspect(sync_conn)
        if table.name not in inspector.get_table_names():
            table.create(sync_conn, checkfirst=True)
            return

        existing = {column["name"] for column in inspector.get_columns(table.name)}
        preparer = sync_conn.dialect.identifier_preparer
        quoted_table = preparer.quote(table.name)
        for name in sorted(_SKILL_CATALOG_COLUMNS - existing):
            column = table.columns[name]
            definition = str(CreateColumn(column).compile(dialect=sync_conn.dialect))
            sync_conn.exec_driver_sql(f"ALTER TABLE {quoted_table} ADD COLUMN {definition}")
            existing.add(name)

        # Preserve the short description previously exposed from ``detail``
        # without selecting that potentially large field in normal list reads.
        sync_conn.execute(
            update(table)
            .where(table.c.detail_preview.is_(None), table.c.detail.is_not(None))
            .values(detail_preview=func.substr(table.c.detail, 1, 1024))
        )

        existing_indexes = {index["name"] for index in inspector.get_indexes(table.name)}
        for index in table.indexes:
            if index.name not in existing_indexes:
                index.create(sync_conn, checkfirst=True)

    async with engine.begin() as conn:
        await conn.run_sync(_ensure)


async def load_skill_catalog_projection(
    *,
    user_id: str,
    admin: bool = False,
) -> list[dict[str, Any]] | None:
    """Read lightweight skill catalog columns from MySQL, never SKILL.md or LONGTEXT.

    ``None`` means persistence is unavailable; an empty list is a valid empty
    catalog. Access filtering happens in SQL so private skill metadata never
    crosses into the request process for an unrelated user.
    """

    session_factory = get_session_factory()
    if session_factory is None:
        return None
    stmt = select(*_SKILL_LIST_COLUMNS).where(SkillRow.available.is_(True))
    if not admin:
        stmt = stmt.where(
            or_(
                SkillRow.owner_user_id.is_(None),
                SkillRow.owner_user_id == user_id,
                SkillRow.published.is_(True),
            )
        )
    stmt = stmt.order_by(SkillRow.created_at.asc())
    try:
        async with session_factory() as session:
            rows = (await session.execute(stmt)).mappings().all()
    except Exception:
        logger.warning("Skill MySQL catalog projection unavailable; retaining legacy fallback", exc_info=True)
        return None
    result: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        item["builtin"] = item.get("owner_user_id") is None
        result.append(item)
    return result


def _parse_catalog_datetime(value: Any) -> datetime | None:
    if isinstance(value, datetime):
        return value
    if not isinstance(value, str) or not value.strip():
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


async def refresh_skill_catalog_projection(
    app_config: AppConfig | None = None,
    *,
    skills: list[Any] | None = None,
) -> int:
    """Synchronize filesystem metadata into one compact MySQL row per skill.

    This runs at startup and after explicit skill changes. Request handlers
    only read the projection and never walk the skill tree or usage sidecars.
    Ownership, publication, and long-form detail columns are preserved.
    """

    session_factory = get_session_factory()
    if session_factory is None:
        return 0
    await ensure_skill_catalog_schema()
    storage = get_or_new_skill_storage(app_config=app_config)
    if skills is None:
        skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
    usage_entries = await asyncio.to_thread(_load_usage_entries)
    catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
    metadata_by_name: dict[str, tuple[str, int]] = {}
    for skill in skills:
        try:
            stat = skill.skill_file.stat()
            metadata_by_name[skill.name] = (
                datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
                int(stat.st_mtime_ns),
            )
        except OSError:
            metadata_by_name[skill.name] = ("", 0)

    async with session_factory() as session:
        existing_rows = (
            await session.execute(select(SkillRow.name, SkillRow.owner_user_id, SkillRow.always_on))
        ).all()
        existing = {str(name): (owner_id, bool(always_on)) for name, owner_id, always_on in existing_rows}
        catalog_names = {str(item.get("name") or "").strip() for item in catalog}
        removed_names = sorted(name for name in existing if name not in catalog_names)
        if removed_names:
            await session.execute(
                update(SkillRow)
                .where(SkillRow.name.in_(removed_names))
                .values(available=False, updated_at=datetime.now(UTC))
            )
        for item in catalog:
            name = str(item.get("name") or "").strip()
            if not name:
                continue
            category = str(item.get("category") or "custom")[:16]
            created_at_text, source_mtime_ns = metadata_by_name.get(name, ("", 0))
            created_at = _parse_catalog_datetime(item.get("created_at") or created_at_text) or datetime.now(UTC)
            fields: dict[str, Any] = {
                "category": category,
                "description": str(item.get("description") or "")[:1024],
                "license": str(item.get("license") or "")[:255] or None,
                "enabled": bool(item.get("enabled", True)),
                "available": True,
                "use_count": max(0, int(item.get("use_count") or 0)),
                "view_count": max(0, int(item.get("view_count") or 0)),
                "patch_count": max(0, int(item.get("patch_count") or 0)),
                "last_activity_at": _parse_catalog_datetime(item.get("last_activity_at")),
                "state": str(item.get("state") or "active")[:16],
                "pinned": bool(item.get("pinned")),
                "featured": bool(item.get("featured")),
                "source": str(item.get("source") or "upload")[:16],
                "source_mtime_ns": source_mtime_ns,
                "updated_at": datetime.now(UTC),
            }
            if name in existing:
                # The per-skill always_on preference is admin-editable in MySQL
                # and must survive filesystem metadata refreshes.
                await session.execute(update(SkillRow).where(SkillRow.name == name).values(**fields))
            else:
                owner_id = None
                session.add(
                    SkillRow(
                        name=name,
                        owner_user_id=owner_id,
                        published=category != "custom",
                        always_on=bool(item.get("always_on", False)),
                        created_at=created_at,
                        **fields,
                    )
                )
        await session.commit()
    return len(catalog)


def _build_catalog_payload(skills: list[Any], usage_entries: dict[str, Any]) -> list[dict[str, Any]]:
    payload: list[dict[str, Any]] = []
    defaults = {
        "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": "active",
        "pinned": False,
        "featured": False,
        "source": "upload",
    }
    for skill in skills:
        usage = {**defaults, **(usage_entries.get(skill.name) or {})}
        created_at = usage.get("created_at")
        if not created_at:
            try:
                created_at = datetime.fromtimestamp(skill.skill_file.stat().st_mtime, tz=UTC).isoformat()
            except OSError:
                created_at = None
        activity = [
            value
            for value in (
                usage.get("last_used_at"),
                usage.get("last_viewed_at"),
                usage.get("last_patched_at"),
            )
            if value
        ]
        last_activity_at = max(activity) if activity else created_at
        payload.append(
            {
                "name": skill.name,
                "description": skill.description or "",
                "license": skill.license,
                "category": str(getattr(skill.category, "value", skill.category)),
                "enabled": bool(skill.enabled),
                "always_on": bool(getattr(skill, "always_on", False)),
                "use_count": int(usage.get("use_count") or 0),
                "view_count": int(usage.get("view_count") or 0),
                "patch_count": int(usage.get("patch_count") or 0),
                "last_activity_at": last_activity_at,
                "created_at": created_at,
                "state": str(usage.get("state") or "active"),
                "pinned": bool(usage.get("pinned")),
                "featured": bool(usage.get("featured")),
                "source": str(usage.get("source") or "upload"),
            }
        )
    return payload


async def refresh_skill_catalog_snapshot(
    app_config: AppConfig | None = None,
    *,
    skills: list[Any] | None = None,
) -> int:
    """Read skill files off-request and persist a single JSON snapshot in MySQL."""

    session_factory = get_session_factory()
    if session_factory is None:
        return 0

    storage = get_or_new_skill_storage(app_config=app_config)
    if skills is None:
        skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
    usage_entries = await asyncio.to_thread(_load_usage_entries)
    catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
    encoded = json.dumps(catalog, ensure_ascii=False, separators=(",", ":"))

    await _persist_catalog_snapshot(encoded)
    global _memory_catalog_snapshot, _memory_catalog_is_fallback
    _memory_catalog_snapshot = catalog
    _memory_catalog_is_fallback = False
    return len(catalog)


async def _persist_catalog_snapshot(encoded: str) -> None:
    session_factory = get_session_factory()
    if session_factory is None:
        return
    row = SkillCatalogSnapshotRow(
        cache_key=_CATALOG_CACHE_KEY,
        catalog_json=encoded,
        updated_at=datetime.now(UTC),
    )
    for attempt in range(2):
        try:
            async with session_factory() as session:
                await session.merge(row)
                await session.commit()
            return
        except IntegrityError:
            if attempt:
                raise
            # Two workers can warm an empty table at the same time. The loser
            # retries as an update after the winner's transaction is visible.


def _load_usage_entries() -> dict[str, Any]:
    from deerflow.skills.usage import all_entries

    return all_entries()


async def load_skill_catalog_snapshot() -> list[dict[str, Any]] | None:
    """Load the prepared catalogue using MySQL only; None means not warmed."""

    session_factory = get_session_factory()
    if session_factory is None:
        return None
    async with session_factory() as session:
        row = await session.get(SkillCatalogSnapshotRow, _CATALOG_CACHE_KEY)
        if row is None:
            return None
        try:
            payload = json.loads(row.catalog_json)
        except (TypeError, ValueError):
            logger.error("Invalid skill catalogue snapshot JSON; refresh is required")
            return None
    if not isinstance(payload, list):
        logger.error("Invalid skill catalogue snapshot payload type; refresh is required")
        return None
    catalog = [item for item in payload if isinstance(item, dict) and item.get("name")]
    global _memory_catalog_snapshot, _memory_catalog_is_fallback
    _memory_catalog_snapshot = catalog
    _memory_catalog_is_fallback = False
    return catalog


async def load_or_warm_skill_catalog_snapshot(
    app_config: AppConfig | None = None,
) -> list[dict[str, Any]] | None:
    """Load MySQL snapshot, warming it once on a cold/missing snapshot.

    Startup warm-up is best-effort. If it missed because of transient startup
    timing, the first list request builds the catalogue off the event loop and
    stores it before returning. If persistence is temporarily unavailable, a
    process-local copy keeps the endpoint usable without rescanning on every
    request.
    """

    global _memory_catalog_snapshot, _memory_catalog_is_fallback
    if get_session_factory() is None:
        return None
    if _memory_catalog_is_fallback and _memory_catalog_snapshot is not None:
        return _memory_catalog_snapshot

    try:
        catalog = await load_skill_catalog_snapshot()
    except Exception:
        logger.warning("Skill catalogue snapshot read failed; attempting a one-time warm-up", exc_info=True)
        catalog = None
    if catalog is not None:
        return catalog

    async with _catalog_warm_lock:
        if _memory_catalog_is_fallback and _memory_catalog_snapshot is not None:
            return _memory_catalog_snapshot
        try:
            catalog = await load_skill_catalog_snapshot()
        except Exception:
            logger.warning("Skill catalogue snapshot is unavailable during warm-up", exc_info=True)
            catalog = None
        if catalog is not None:
            return catalog

        storage = get_or_new_skill_storage(app_config=app_config)
        skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
        usage_entries = await asyncio.to_thread(_load_usage_entries)
        catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
        encoded = json.dumps(catalog, ensure_ascii=False, separators=(",", ":"))
        try:
            await _persist_catalog_snapshot(encoded)
        except Exception:
            logger.exception("Skill catalogue snapshot persistence failed; serving the warmed process-local copy")
            _memory_catalog_snapshot = catalog
            _memory_catalog_is_fallback = True
            return catalog
        _memory_catalog_snapshot = catalog
        _memory_catalog_is_fallback = False
        logger.info("Skill catalogue snapshot warmed on first request: skills=%d", len(catalog))
        return catalog