group_fqcd_jr/app/service/config_release_service.py

import logging
from datetime import UTC, datetime
from typing import Any
from uuid import uuid4

from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.model.audit import InteractionAudit
from app.model.configuration import ConfigRelease, PlatformConfigItem
from app.model.platform import DomainEventOutbox

logger = logging.getLogger(__name__)

# 受 `config_release` **整版本替换**影响的表，以及各自的**逻辑键**。
#
# 逻辑键用来判断"同一份配置"在新版本里还在不在：它不含 `release_id`、不含自增 id，
# 也**不含 version** —— 同名提示词在不同版本里可以用不同 version，那仍是同一份配置。
#
# 这张清单是**穷举**的：`information_schema` 里带 `release_id` 列的表只有这三张。
# 漏掉任何一张的后果都是**静默失效**，而且不会报错：客服闲聊提示词就这么失效过一次
# —— 它挂在 release 174，active 变成 181 后 `load_active_prompt` 读不到，而 Agent 侧
# 有逐字段兜底、回落到代码默认值，于是功能看着正常、没有任何人发现、也没有任何告警。
RELEASE_SCOPED_TABLES: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("platform_config_item", ("namespace", "config_key")),
    ("prompt_template_version", ("prompt_code", "task_type", "agent_type")),
    ("model_routing_rule", ("rule_code",)),
)

# 搬运配置项时要剥掉的列：`id` 与 `release_id` 由新版本自己生成，
# 带过去要么主键冲突、要么把内容挂到旧版本上。
NOT_PORTABLE_COLUMNS = frozenset({"id", "release_id"})


class ConfigReleaseError(ValueError):
    pass


class ConfigReleaseService:
    def __init__(self, session: AsyncSession) -> None:
        self.session = session

    async def submit_for_review(self, release_id: int, actor_id: int) -> ConfigRelease:
        release = await self._get(release_id)
        if release.created_by != actor_id or release.status != "draft":
            raise ConfigReleaseError("only draft owner can submit")
        release.status = "pending_review"
        release.updated_at = self._now()
        self._audit("config.submit_review", actor_id, release)
        await self.session.flush()
        return release

    async def approve(self, release_id: int, reviewer_id: int) -> ConfigRelease:
        # 只保留状态机约束：必须处于待审核状态才能审核通过；不再要求审核人不同于创建人
        # （单管理员部署下自审是唯一可行路径）。
        #
        # 自审与多人复核一样**如实写入 `reviewer_id`**：该列的含义就是"审核人"，留空会让
        # `admin_service` 的激活分支（要求 `reviewer_id is not None`）把已审核版本判为
        # "未审核"而拒绝激活。
        #
        # 自审要落库，前提是数据库侧不再有 `chk_config_release_separation`
        # （`reviewer_id IS NULL OR reviewer_id <> created_by`）。合并后的仓库里
        # `20260910_drop_review_separation` 与 `20260911_drop_review_separation`
        # 都会撤下它（两条分支各补一次，均幂等）；若约束仍在，写
        # `reviewer_id = created_by` 会抛
        # `Check constraint 'chk_config_release_separation' is violated`，表现为审核接口
        # 500 而不是业务异常。
        release = await self._get(release_id)
        if release.status != "pending_review":
            raise ConfigReleaseError("release is not pending review")
        now = self._now()
        release.status = "approved"
        release.reviewer_id = reviewer_id
        release.reviewed_at = now
        release.updated_at = now
        self._audit("config.approve", reviewer_id, release)
        await self.session.flush()
        return release

    async def active_release_id(self) -> int | None:
        """当前生效版本的 id；没有任何生效版本时返回 None。"""
        found = await self.session.scalar(
            select(ConfigRelease.id).where(ConfigRelease.status == "active")
        )
        return int(found) if found is not None else None

    async def effective_snapshot(
        self, release_id: int | None = None
    ) -> dict[str, list[dict[str, Any]]]:
        """读某个发布版本在**全部受管表**里的内容；默认读当前生效版本。

        发布脚本应当**先取这份快照**，把它原样搬到新版本、再追加本次变更 —— 因为
        `config_release` 是整版本替换，不搬就等于删（详见 RELEASE_SCOPED_TABLES 的说明）。
        每行已剥掉 `id` 与 `release_id`（见 NOT_PORTABLE_COLUMNS），可直接作为写入载荷。
        """
        if release_id is None:
            release_id = await self.active_release_id()
        snapshot: dict[str, list[dict[str, Any]]] = {}
        for table, _keys in RELEASE_SCOPED_TABLES:
            if release_id is None:
                snapshot[table] = []
                continue
            rows = (await self.session.execute(
                text(f"SELECT * FROM {table} WHERE release_id = :rid"), {"rid": release_id}
            )).mappings().all()
            snapshot[table] = [
                {
                    name: value
                    for name, value in dict(row).items()
                    if name not in NOT_PORTABLE_COLUMNS
                }
                for row in rows
            ]
        return snapshot

    async def _table_keys(
        self, table: str, keys: tuple[str, ...], release_ids: list[int]
    ) -> dict[int, set[tuple[Any, ...]]]:
        """按 release 分组取出某张表的逻辑键集合。"""
        columns = ", ".join(keys)
        grouped: dict[int, set[tuple[Any, ...]]] = {}
        for release_id in release_ids:
            rows = (await self.session.execute(
                text(f"SELECT {columns} FROM {table} WHERE release_id = :rid"),
                {"rid": release_id},
            )).mappings().all()
            grouped[release_id] = {tuple(row[key] for key in keys) for row in rows}
        return grouped

    async def _warn_dropped_items(
        self, release: ConfigRelease, previous_active: list[ConfigRelease]
    ) -> None:
        """点名"旧版本有、新版本没有"的配置 —— 它们在激活后会静默失效。

        **逐张覆盖全部受管表**。原先只比对 `platform_config_item`，于是客服闲聊提示词在
        `prompt_template_version` 里被静默丢掉时，连一行告警都没有：它挂在 release 174，
        active 变成 181 后读不到，而 Agent 侧有兜底、回落到代码默认值，功能看着正常，
        于是没有人发现（详见 RELEASE_SCOPED_TABLES 的说明）。

        不阻断激活 —— 有时确实是要主动撤下某项配置；这里要的是"事后能查到是谁弄没的"。
        """
        if not previous_active:
            return
        release_ids = [previous.id for previous in previous_active] + [release.id]
        for table, keys in RELEASE_SCOPED_TABLES:
            grouped = await self._table_keys(table, keys, release_ids)
            new_keys = grouped.get(release.id, set())
            for previous in previous_active:
                dropped = sorted(
                    (key for key in grouped.get(previous.id, set()) if key not in new_keys),
                    key=str,
                )
                if dropped:
                    logger.warning(
                        "配置发布 %s 取代 %s：%s 里有 %d 条配置在新版本中不存在，"
                        "激活后即失效 → %s",
                        release.release_no,
                        previous.release_no,
                        table,
                        len(dropped),
                        [dict(zip(keys, key, strict=True)) for key in dropped],
                    )

    async def activate(self, release_id: int, actor_id: int) -> ConfigRelease:
        release = await self._get(release_id)
        if release.status != "approved":
            raise ConfigReleaseError("only approved release can activate")
        active = await self.session.scalars(
            select(ConfigRelease).where(ConfigRelease.status == "active")
        )
        now = self._now()
        previous_active = [previous for previous in active if previous.id != release.id]
        # 把"本次激活会让哪些配置项失效"显式记进日志，见 _warn_dropped_items 的说明。
        await self._warn_dropped_items(release, previous_active)
        for previous in previous_active:
            previous.status = "superseded"
            previous.updated_at = now
        release.status = "active"
        release.activated_at = now
        release.updated_at = now
        # 记录"本版本取代了谁"。`supersedes_release_id` 的列含义就是被取代的版本，但此前
        # 从未写入，导致版本链断裂：清理或回滚旧数据时无法知道"谁顶掉了谁"，一旦某次实验
        # 创建的临时版本被删掉，平台就会静默停在"零个 active 版本"的状态（工具白名单随之
        # 变成空集，所有 Agent 工具按 fail-closed 被拒）。唯一键 `uk_config_release_active_one`
        # 保证同时最多一个 active，所以这里最多记录一条。
        if len(previous_active) == 1:
            release.supersedes_release_id = previous_active[0].id
        self._audit("config.activate", actor_id, release)
        self.session.add(
            DomainEventOutbox(
                id=0,
                event_id=str(uuid4()),
                event_type="config.cache_invalidate_requested",
                aggregate_type="config_release",
                aggregate_id=str(release.id),
                trace_id=release.release_no,
                payload={"release_id": release.id, "release_no": release.release_no},
                status="pending",
                retry_count=0,
                occurred_at=now,
                created_at=now,
                updated_at=now,
            )
        )
        await self.session.flush()
        return release

    async def rollback(self, release_id: int, actor_id: int) -> ConfigRelease:
        target = await self._get(release_id)
        if target.status not in {"active", "superseded"}:
            raise ConfigReleaseError("release cannot be rolled back")
        now = self._now()
        current = await self.session.scalars(
            select(ConfigRelease).where(ConfigRelease.status == "active")
        )
        for release in current:
            release.status = "superseded"
            release.updated_at = now
        await self.session.flush()
        rollback = ConfigRelease(
            release_no=f"rollback-{uuid4().hex[:16]}",
            title=f"Rollback {target.release_no}",
            change_summary=f"Rollback to {target.release_no}",
            status="active",
            created_by=actor_id,
            reviewer_id=None,
            reviewed_at=None,
            activated_at=now,
            created_at=now,
            updated_at=now,
            rollback_of_release_id=target.id,
        )
        self.session.add(rollback)
        await self.session.flush()
        self.session.add(
            DomainEventOutbox(
                event_id=str(uuid4()),
                event_type="config.cache_invalidate_requested",
                aggregate_type="config_release",
                aggregate_id=str(target.id),
                trace_id=target.release_no,
                payload={"release_id": target.id, "release_no": target.release_no},
                status="pending",
                retry_count=0,
                occurred_at=now,
                created_at=now,
                updated_at=now,
            )
        )
        self._audit("config.rollback", actor_id, rollback)
        await self.session.flush()
        return rollback

    async def get_value(
        self, release_id: int, namespace: str, config_key: str
    ) -> dict[str, object] | None:
        item = await self.session.scalar(
            select(PlatformConfigItem).where(
                PlatformConfigItem.release_id == release_id,
                PlatformConfigItem.namespace == namespace,
                PlatformConfigItem.config_key == config_key,
            )
        )
        return item.value_json if item else None

    @staticmethod
    def assert_if_match(current_version: str, if_match: str | None) -> None:
        if if_match is None or if_match.strip('"') != current_version:
            raise ConfigReleaseError("CONFIG_VERSION_CONFLICT")

    async def _get(self, release_id: int) -> ConfigRelease:
        release = await self.session.get(ConfigRelease, release_id)
        if release is None:
            raise ConfigReleaseError("release not found")
        return release

    def _audit(self, action: str, actor_id: int, release: ConfigRelease) -> None:
        self.session.add(
            InteractionAudit(
                actor_type="user",
                actor_id=actor_id,
                target_customer_id=None,
                session_id=None,
                portal="admin",
                action_type=action,
                detail={"release_id": release.id, "release_no": release.release_no},
                created_at=self._now(),
            )
        )

    @staticmethod
    def _now() -> datetime:
        return datetime.now(UTC).replace(tzinfo=None)