"""大屏的业务逻辑：存取、乐观锁、发布、历史、复制、封面、线上读取和数据源取数

存储层只提供"匹配并更新"这类原语，404 / 409 的判断都在这里
"""

import asyncio
import copy
import secrets
import weakref
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from typing import Any, Protocol
from uuid import uuid4

from loguru import logger
from pydantic import ValidationError

from app.core.exceptions import BizException
from app.schemas.saved_screen import (
    NAME_MAX_LENGTH,
    ListStatus,
    PublishEntry,
    PublishInfo,
    PublishSnapshot,
    ScreenRecord,
    ScreenSummary,
    SortField,
)
from app.schemas.screen import Screen
from app.services.datasource_proxy import FetchCache, fetch_json
from app.services.screen_store import ScreenStore
from app.services.thumbnails import ThumbnailStorage

# 事件代码和数据源转换函数是字符串形式的 JS，线上会在浏览器里执行；写入权限只给 owner，再限制长度
SCRIPT_MAX_CHARS = 64 * 1024
COPY_SUFFIX = " 副本"
EMPTY_PAGE = {
    # 不设背景色：用主题的背景色，组件颜色也跟着主题走
    "canvas": {"width": 1920, "height": 1080, "theme": "light"},
    "nodes": [],
    "dataSource": [],
}

# 回收站过期清理：一批最多清多少个，两次扫描的间隔（秒）
PURGE_BATCH = 100
TRASH_SWEEP_INTERVAL = 3600

STALE = "大屏已在别处修改"
# 下线 / 删除 / 换封面遇到并发修改时的重读次数
UPDATE_ATTEMPTS = 3


def _now() -> datetime:
    return datetime.now(UTC)


class FilePinner(Protocol):
    """数据文件服务里和大屏有关的两个方法"""

    async def sync_screen_files(
        self, user_id: int, screen_id: str, file_ids: set[str]
    ) -> None: ...

    async def release_screen_files(self, screen_id: str) -> None: ...


def referenced_file_ids(page: dict[str, Any]) -> set[str]:
    """页面里图表绑定的数据文件：nodes[].fileQuery.file_id"""
    ids = set()
    nodes = page.get("nodes")
    for node in nodes if isinstance(nodes, list) else []:
        query = node.get("fileQuery") if isinstance(node, dict) else None
        file_id = query.get("file_id") if isinstance(query, dict) else None
        if isinstance(file_id, str) and file_id:
            ids.add(file_id)
    return ids


def _publish_is(publish: PublishInfo | None) -> dict[str, Any]:
    """ "线上指针还是我读到的那个"的更新条件"""
    if publish is None:
        return {"publish": None}
    return {"publish.version": publish.version, "publish.status": publish.status}


def _not_found() -> BizException:
    # 别人的大屏与不存在同样返回 404
    return BizException(404, "大屏不存在或已删除")


def validate_page(page: dict[str, Any]) -> dict[str, Any]:
    """按 Screen 校验结构，但返回的仍是原始 JSON：model_dump 会给没传的字段补出 null，前端读回来就变了"""
    try:
        Screen.model_validate(page)
    except ValidationError as error:
        first = error.errors()[0]
        where = ".".join(str(part) for part in first["loc"])
        raise BizException(422, f"大屏数据不合法：{where}：{first['msg']}") from None
    _check_script_length(page)
    return page


def _check_script_length(page: dict[str, Any]) -> None:
    scripts = [
        event.get("code")
        for node in page["nodes"]
        for event in (node.get("events") or [])
        if isinstance(event, dict)
    ] + [
        source.get("transform")
        for source in page.get("dataSource", [])
        if isinstance(source, dict)
    ]
    if any(isinstance(text, str) and len(text) > SCRIPT_MAX_CHARS for text in scripts):
        raise BizException(
            422, f"事件代码和数据转换函数不能超过 {SCRIPT_MAX_CHARS // 1024} KB"
        )


def public_page(page: dict[str, Any]) -> dict[str, Any]:
    """线上页面里去掉数据源的 headers：它们常放 API key，线上取数由服务端按快照里的完整配置发起，浏览器用不着"""
    sources = page.get("dataSource")
    if not isinstance(sources, list):
        return {**page}
    return {
        **page,
        "dataSource": [
            {key: value for key, value in item.items() if key != "headers"}
            if isinstance(item, dict)
            else item
            for item in sources
        ],
    }


class ScreenService:
    def __init__(
        self,
        store: ScreenStore,
        thumbnails: ThumbnailStorage,
        files: FilePinner | None = None,
        trash_days: int = 30,
    ):
        self._store = store
        self._thumbnails = thumbnails
        # 已保存的大屏引用的数据文件不能过期，否则"刷新数据"会 404；没有文件服务时不管
        self._files = files
        if trash_days < 1:
            raise ValueError("trash_days 必须是正整数")
        self.trash_days = trash_days
        # 按大屏的锁：保存 / 回滚 / 删除 / 还原都是"改库再同步固定的文件"两步，
        # 两个请求交错会让固定的文件跟着先开始、后结束的那个请求的旧页面走。单实例部署，进程内锁就够
        self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = (
            weakref.WeakValueDictionary()
        )
        self._datasource_cache = FetchCache()

    def _lock(self, screen_id: str) -> asyncio.Lock:
        lock = self._locks.get(screen_id)
        if lock is None:
            lock = self._locks[screen_id] = asyncio.Lock()
        return lock

    async def _pin_files(self, owner_id: int, screen_id: str, page: dict[str, Any]):
        """让数据文件跟着大屏的引用走。主操作已经成功，这一步失败只记日志，不让保存报错"""
        if self._files is None:
            return
        try:
            await self._files.sync_screen_files(
                owner_id, screen_id, referenced_file_ids(page)
            )
        except Exception as error:
            logger.opt(exception=error).error(
                "同步大屏 {} 引用的数据文件失败", screen_id
            )

    async def _release_files(self, screen_id: str) -> None:
        if self._files is None:
            return
        try:
            await self._files.release_screen_files(screen_id)
        except Exception as error:
            logger.opt(exception=error).error(
                "放开大屏 {} 固定的数据文件失败", screen_id
            )

    # ---- 草稿 ----

    async def create(
        self, owner_id: int, name: str, page: dict[str, Any] | None
    ) -> ScreenRecord:
        now = _now()
        record = ScreenRecord(
            screen_id=f"s_{uuid4().hex}",
            owner_id=owner_id,
            name=name,
            page=validate_page(copy.deepcopy(EMPTY_PAGE) if page is None else page),
            revision=1,
            created_at=now,
            updated_at=now,
        )
        await self._store.insert(record)
        await self._pin_files(owner_id, record.screen_id, record.page)
        return record

    async def get(self, screen_id: str, owner_id: int) -> ScreenRecord:
        record = await self._store.get(screen_id, owner_id)
        if record is None:
            raise _not_found()
        return record

    async def save(
        self, screen_id: str, owner_id: int, page: dict[str, Any], revision: int
    ) -> ScreenSummary:
        validate_page(page)
        async with self._lock(screen_id):
            saved = await self._update_with_revision(
                screen_id,
                owner_id,
                {"page": page, "revision": revision + 1, "updated_at": _now()},
                revision,
            )
            await self._pin_files(owner_id, screen_id, page)
        return saved

    async def _update_with_revision(
        self, screen_id: str, owner_id: int, changes: dict[str, Any], revision: int
    ) -> ScreenSummary:
        updated = await self._store.update(
            screen_id, owner_id, changes, revision=revision
        )
        if updated is not None:
            return updated
        # 没更新成功：大屏不存在是 404，存在但 revision 不一致是 409
        if await self._store.get_summary(screen_id, owner_id) is None:
            raise _not_found()
        raise BizException(409, STALE)

    async def rename(self, screen_id: str, owner_id: int, name: str) -> None:
        # 改名不算改内容，revision 不动
        changes = {"name": name, "updated_at": _now()}
        if await self._store.update(screen_id, owner_id, changes) is None:
            raise _not_found()

    async def _update_from_summary(
        self,
        screen_id: str,
        owner_id: int,
        build: Callable[[ScreenSummary], tuple[dict[str, Any], dict[str, Any]] | None],
    ) -> None:
        """读出当前状态，据此算出修改，再带着"状态没变"的条件写回；并发改了就重读再来

        下线、删除要整段改写 publish，不带条件就会把并发发布写进去的新指针盖回旧值
        build 返回 (changes, where)；返回 None 表示已经是目标状态，什么都不用做
        """
        for _ in range(UPDATE_ATTEMPTS):
            summary = await self._store.get_summary(screen_id, owner_id)
            if summary is None:
                raise _not_found()
            built = build(summary)
            if built is None:
                return
            changes, where = built
            if await self._store.update(screen_id, owner_id, changes, where=where):
                return
        raise BizException(409, "操作冲突，请重试")

    async def delete(self, screen_id: str, owner_id: int) -> None:
        def build(summary: ScreenSummary):
            changes: dict[str, Any] = {"deleted_at": _now()}
            if summary.publish is None:
                return changes, {"publish": None}
            # 线上版本同时下线，分享链接立即 404
            changes["publish"] = summary.publish.model_copy(
                update={"status": "offline"}
            )
            return changes, _publish_is(summary.publish)

        async with self._lock(screen_id):
            await self._update_from_summary(screen_id, owner_id, build)
            # 进了回收站的大屏不再固定文件，还原时再固定回来
            await self._release_files(screen_id)

    async def restore_deleted(self, screen_id: str, owner_id: int) -> None:
        async with self._lock(screen_id):
            record = await self._store.get(screen_id, owner_id, deleted=True)
            if record is None:
                raise _not_found()
            changes = {"deleted_at": None}
            if (
                await self._store.update(screen_id, owner_id, changes, deleted=True)
                is None
            ):
                raise _not_found()
            await self._pin_files(owner_id, screen_id, record.page)

    # ---- 回收站 ----

    async def _purge(self, summary: ScreenSummary) -> bool:
        """彻底删除回收站里的一个大屏

        先删大屏本身（带"仍在回收站"的条件，清理期间被还原的不会误删），再清理附属数据
        """
        if not await self._store.purge(summary.screen_id, summary.owner_id):
            return False
        await self._store.delete_publishes(summary.screen_id)
        self._thumbnails.delete(summary.thumbnail)
        await self._release_files(summary.screen_id)
        return True

    async def empty_trash(self, owner_id: int) -> int:
        """清空自己的回收站，返回彻底删除的个数"""
        count = 0
        previous: set[str] = set()
        while True:
            batch, _ = await self._store.search(
                owner_id,
                keyword=None,
                status=None,
                sort="updated_at",
                descending=True,
                limit=PURGE_BATCH,
                offset=0,
                deleted=True,
            )
            ids = {summary.screen_id for summary in batch}
            purged = [await self._purge(summary) for summary in batch]
            count += sum(purged)
            # 空批才算清完；整批都没删成（被并发还原了）就再查一次，
            # 但同一批连续两次都删不掉说明有东西卡住，别空转
            if not batch or (not any(purged) and ids == previous):
                return count
            previous = ids

    async def purge_expired_trash(self) -> int:
        """彻底删除在回收站里超过保留天数的大屏（所有人的）"""
        deadline = _now() - timedelta(days=self.trash_days)
        count = 0
        previous: set[str] = set()
        while True:
            batch = await self._store.list_expired_trash(deadline, PURGE_BATCH)
            ids = {summary.screen_id for summary in batch}
            purged = [await self._purge(summary) for summary in batch]
            count += sum(purged)
            if not batch or (not any(purged) and ids == previous):
                return count
            previous = ids

    async def run_trash_sweeper(self) -> None:
        while True:
            try:
                if removed := await self.purge_expired_trash():
                    logger.info(
                        "已彻底删除 {} 个超过 {} 天的回收站大屏",
                        removed,
                        self.trash_days,
                    )
            except Exception as error:
                logger.opt(exception=error).error("清理回收站失败")
            await asyncio.sleep(TRASH_SWEEP_INTERVAL)

    # ---- 管理页 ----

    async def list_screens(
        self,
        owner_id: int,
        *,
        keyword: str | None,
        status: ListStatus | None,
        sort: SortField,
        descending: bool,
        limit: int,
        offset: int,
        deleted: bool,
    ) -> tuple[list[ScreenSummary], int]:
        return await self._store.search(
            owner_id,
            keyword=keyword,
            status=status,
            sort=sort,
            descending=descending,
            limit=limit,
            offset=offset,
            deleted=deleted,
        )

    async def copy_screen(self, screen_id: str, owner_id: int, name: str | None) -> str:
        """只复制草稿和封面，不复制发布状态和历史"""
        source = await self.get(screen_id, owner_id)
        base = source.name[: NAME_MAX_LENGTH - len(COPY_SUFFIX)]
        copied = await self.create(
            owner_id,
            name or f"{base}{COPY_SUFFIX}",
            copy.deepcopy(source.page),
        )
        if source.thumbnail is not None:
            thumbnail = self._thumbnails.copy(source.thumbnail)
            if thumbnail is not None:
                try:
                    attached = await self._store.update(
                        copied.screen_id, owner_id, {"thumbnail": thumbnail}
                    )
                except BaseException:
                    self._thumbnails.delete(thumbnail)
                    raise
                if not attached:
                    self._thumbnails.delete(thumbnail)
        return copied.screen_id

    async def set_thumbnail(self, screen_id: str, owner_id: int, data: bytes) -> str:
        thumbnail = self._thumbnails.save(data)
        replaced: list[str | None] = []

        def build(summary: ScreenSummary):
            # 封面只是展示用的，不算改内容：不动 revision 和 updated_at
            replaced[:] = [summary.thumbnail]
            return {"thumbnail": thumbnail}, {"thumbnail": summary.thumbnail}

        try:
            await self._update_from_summary(screen_id, owner_id, build)
        except BaseException:
            self._thumbnails.delete(thumbnail)
            raise
        # 删的是这次真正换掉的那张；并发上传时各自换掉各自读到的，不会留下孤儿文件
        self._thumbnails.delete(replaced[0])
        return thumbnail

    def open_thumbnail(self, thumbnail_id: str):
        return self._thumbnails.open(thumbnail_id)

    # ---- 发布 ----

    async def publish(
        self, screen_id: str, owner_id: int, revision: int
    ) -> PublishInfo:
        record = await self.get(screen_id, owner_id)
        if record.revision != revision:
            raise BizException(409, "大屏有未保存或已在别处修改的内容，请先保存")
        previous = record.publish
        snapshot = PublishSnapshot(
            screen_id=screen_id,
            # 版本号取已有快照的最大值加一：写快照后没来得及更新指针留下的孤儿快照也不会撞号
            version=await self._store.last_publish_version(screen_id) + 1,
            revision=record.revision,
            page=record.page,
            published_at=_now(),
            published_by=owner_id,
        )
        if not await self._store.insert_publish(snapshot):
            raise BizException(409, "发布冲突，请重试")
        info = PublishInfo(
            status="published",
            version=snapshot.version,
            revision=snapshot.revision,
            # 第一次发布生成 share_id，之后沿用；下线再发布也沿用
            share_id=previous.share_id
            if previous
            else f"p_{secrets.token_urlsafe(16)}",
            published_at=snapshot.published_at,
        )
        # 条件：草稿没被再保存过（revision），线上指针还是读到的那个——并发发布只有一个能成，
        # 否则第一次发布时两边各生成一个 share_id，先返回给前端的那个会失效
        try:
            updated = await self._store.update(
                screen_id,
                owner_id,
                {"publish": info},
                revision=revision,
                where=_publish_is(previous),
            )
        except BaseException:
            await self._store.delete_publish(screen_id, snapshot.version)
            raise
        if updated is None:
            await self._store.delete_publish(screen_id, snapshot.version)
            raise BizException(409, STALE)
        return info

    async def unpublish(self, screen_id: str, owner_id: int) -> None:
        def build(summary: ScreenSummary):
            if summary.publish is None or summary.publish.status == "offline":
                return None
            offline = summary.publish.model_copy(update={"status": "offline"})
            return {"publish": offline}, _publish_is(summary.publish)

        await self._update_from_summary(screen_id, owner_id, build)

    async def exists(self, screen_id: str, owner_id: int) -> bool:
        return await self._store.get_summary(screen_id, owner_id) is not None

    async def list_publishes(
        self, screen_id: str, owner_id: int, limit: int, offset: int
    ) -> tuple[list[PublishEntry], int]:
        if await self._store.get_summary(screen_id, owner_id) is None:
            raise _not_found()
        return await self._store.list_publishes(screen_id, limit, offset)

    async def restore_publish(self, screen_id: str, owner_id: int, version: int) -> int:
        """把历史版本的页面覆盖回草稿，不自动发布"""
        async with self._lock(screen_id):
            record = await self.get(screen_id, owner_id)
            snapshot = await self._store.get_publish(screen_id, version)
            if snapshot is None:
                raise BizException(404, "发布版本不存在")
            updated = await self._update_with_revision(
                screen_id,
                owner_id,
                {
                    "page": snapshot.page,
                    "revision": record.revision + 1,
                    "updated_at": _now(),
                },
                record.revision,
            )
            await self._pin_files(owner_id, screen_id, snapshot.page)
        return updated.revision

    # ---- 线上（匿名） ----

    async def _live(self, share_id: str) -> tuple[ScreenSummary, PublishSnapshot]:
        """线上版本：已下线 / 已删除 / 不存在都是 404；返回的是发布快照，不是草稿"""
        summary = await self._store.get_by_share_id(share_id)
        if (
            summary is None
            or summary.publish is None
            or summary.publish.status != "published"
        ):
            raise _not_found()
        snapshot = await self._store.get_publish(
            summary.screen_id, summary.publish.version
        )
        if snapshot is None:
            raise _not_found()
        return summary, snapshot

    async def get_live(self, share_id: str) -> tuple[ScreenSummary, PublishSnapshot]:
        return await self._live(share_id)

    async def fetch_live_datasource(self, share_id: str, datasource_id: str):
        """前端只传数据源 ID，请求地址取自已发布快照；不能做成"前端传 URL 后端转发"，否则是匿名开放代理"""
        _, snapshot = await self._live(share_id)
        source = next(
            (
                item
                for item in snapshot.page.get("dataSource", [])
                if isinstance(item, dict) and item.get("id") == datasource_id
            ),
            None,
        )
        if source is None or source.get("type") != "api" or not source.get("url"):
            raise BizException(404, "数据源不存在")
        method = "POST" if source.get("method") == "POST" else "GET"
        params = source.get("params")
        headers = source.get("headers")
        return await self._datasource_cache.get(
            (share_id, snapshot.version, datasource_id),
            lambda: fetch_json(
                str(source["url"]),
                method,
                params if isinstance(params, dict) else None,
                headers if isinstance(headers, dict) else None,
            ),
        )