from __future__ import annotations

import os
from typing import Any, Optional

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

from ...config.ai_prompt_registry import get_prompt_registry_entry
from ...models import PromptRelease, PromptReleaseCheck
from ...utils.timezone import now_cn_naive


_RELEASE_CHECK_TYPES = ("variable_contract", "schema_smoke", "golden_replay")
PLATFORM_RUNTIME_TENANT_ID = int(os.getenv("PMFLOW_PLATFORM_RUNTIME_TENANT_ID", "1") or "1")


class PromptReleaseService:
    def __init__(self, db: AsyncSession):
        self.db = db

    @staticmethod
    def _snapshot(row: PromptRelease) -> dict[str, Any]:
        return {
            "id": int(row.id),
            "tenant_id": int(row.tenant_id),
            "prompt_key": row.prompt_key,
            "version": int(row.version),
            "body_snapshot": row.body_snapshot,
            "status": row.status,
            "notes": row.notes,
            "created_by": int(row.created_by) if row.created_by is not None else None,
            "activated_by": int(row.activated_by) if row.activated_by is not None else None,
            "activated_at": row.activated_at,
        }

    async def create_release(
        self,
        *,
        tenant_id: int,
        prompt_key: str,
        body_snapshot: str,
        created_by: Optional[int],
        notes: Optional[str] = None,
    ) -> dict[str, Any]:
        prompt_key = str(prompt_key or "").strip()
        if not prompt_key:
            raise ValueError("prompt_key is required")
        if get_prompt_registry_entry(prompt_key) is None:
            raise ValueError("unsupported prompt_key")
        body_snapshot = str(body_snapshot or "").strip()
        if not body_snapshot:
            raise ValueError("body_snapshot is required")
        current_max = (
            await self.db.execute(
                select(func.max(PromptRelease.version)).where(
                    PromptRelease.tenant_id == int(tenant_id),
                    PromptRelease.prompt_key == prompt_key,
                )
            )
        ).scalar_one_or_none()
        row = PromptRelease(
            tenant_id=int(tenant_id),
            prompt_key=prompt_key,
            version=int(current_max or 0) + 1,
            body_snapshot=body_snapshot,
            status="draft",
            notes=str(notes or "").strip() or None,
            created_by=created_by,
        )
        self.db.add(row)
        await self.db.commit()
        await self.db.refresh(row)
        return self._snapshot(row)

    async def activate_release(
        self,
        *,
        tenant_id: int,
        release_id: int,
        actor_id: Optional[int],
        prompt_key: Optional[str] = None,
    ) -> dict[str, Any]:
        filters = [
            PromptRelease.tenant_id == int(tenant_id),
            PromptRelease.id == int(release_id),
        ]
        if prompt_key:
            filters.append(PromptRelease.prompt_key == str(prompt_key).strip())
        row = (
            await self.db.execute(
                select(PromptRelease).where(*filters)
            )
        ).scalar_one_or_none()
        if row is None:
            raise ValueError("prompt release not found")
        registry_entry = get_prompt_registry_entry(row.prompt_key)
        if registry_entry and str(registry_entry.get("editable_policy") or "") == "guarded_release":
            await self._assert_guarded_gate_passed(tenant_id=tenant_id, release_id=int(row.id))
        previous = (
            await self.db.execute(
                select(PromptRelease).where(
                    PromptRelease.tenant_id == int(tenant_id),
                    PromptRelease.prompt_key == row.prompt_key,
                    PromptRelease.status == "active",
                )
            )
        ).scalars().all()
        for item in previous:
            item.status = "archived"
        row.status = "active"
        row.activated_by = actor_id
        row.activated_at = now_cn_naive().isoformat()
        await self.db.commit()
        await self.db.refresh(row)
        return self._snapshot(row)

    async def _assert_guarded_gate_passed(self, *, tenant_id: int, release_id: int) -> None:
        rows = (
            await self.db.execute(
                select(PromptReleaseCheck)
                .where(
                    PromptReleaseCheck.tenant_id == int(tenant_id),
                    PromptReleaseCheck.release_id == int(release_id),
                )
                .order_by(PromptReleaseCheck.check_type.asc(), PromptReleaseCheck.id.desc())
            )
        ).scalars().all()
        latest: dict[str, PromptReleaseCheck] = {}
        for row in rows:
            check_type = str(row.check_type or "")
            if check_type and check_type not in latest:
                latest[check_type] = row
        blocking: list[str] = []
        for check_type in _RELEASE_CHECK_TYPES:
            row = latest.get(check_type)
            if row is None:
                blocking.append(f"{check_type} not run")
            elif str(row.status or "") != "passed":
                blocking.append(f"{check_type}: {row.summary}")
        if blocking:
            raise ValueError(f"guarded prompt release is blocked by gate: {'; '.join(blocking)}")

    async def rollback_release(
        self,
        *,
        tenant_id: int,
        prompt_key: str,
        release_id: int,
        actor_id: Optional[int],
    ) -> dict[str, Any]:
        row = (
            await self.db.execute(
                select(PromptRelease).where(
                    PromptRelease.tenant_id == int(tenant_id),
                    PromptRelease.prompt_key == str(prompt_key or "").strip(),
                    PromptRelease.id == int(release_id),
                )
            )
        ).scalar_one_or_none()
        if row is None:
            raise ValueError("prompt release not found")
        if row.status != "archived":
            raise ValueError("only archived prompt releases can be rolled back")
        return await self.activate_release(
            tenant_id=tenant_id,
            prompt_key=prompt_key,
            release_id=release_id,
            actor_id=actor_id,
        )

    async def list_releases(
        self,
        *,
        tenant_id: int,
        prompt_key: Optional[str] = None,
    ) -> dict[str, Any]:
        stmt = select(PromptRelease).where(PromptRelease.tenant_id == int(tenant_id))
        if prompt_key:
            stmt = stmt.where(PromptRelease.prompt_key == str(prompt_key).strip())
        rows = (
            await self.db.execute(
                stmt.order_by(PromptRelease.prompt_key.asc(), PromptRelease.version.desc(), PromptRelease.id.desc())
            )
        ).scalars().all()
        return {"items": [self._snapshot(row) for row in rows], "total": len(rows)}

    async def get_active_template(
        self,
        *,
        tenant_id: int,
        prompt_key: str,
    ) -> Optional[str]:
        row = await self.get_active_release(tenant_id=tenant_id, prompt_key=prompt_key)
        return str(row.get("body_snapshot") or "").strip() if row else None

    async def get_active_release(
        self,
        *,
        tenant_id: int,
        prompt_key: str,
    ) -> Optional[dict[str, Any]]:
        row = (
            await self.db.execute(
                select(PromptRelease).where(
                    PromptRelease.tenant_id == int(tenant_id),
                    PromptRelease.prompt_key == str(prompt_key or "").strip(),
                    PromptRelease.status == "active",
                )
            )
        ).scalar_one_or_none()
        return self._snapshot(row) if row else None

    async def render_prompt(
        self,
        *,
        tenant_id: int,
        prompt_key: str,
        fallback_template: str,
        _override_template: Optional[str] = None,
        **kwargs: Any,
    ) -> str:
        template = str(_override_template or "").strip() if _override_template is not None else ""
        if not template:
            template = str(await self.get_active_template(tenant_id=tenant_id, prompt_key=prompt_key) or "").strip()
        if not template and int(tenant_id) != PLATFORM_RUNTIME_TENANT_ID:
            template = str(
                await self.get_active_template(tenant_id=PLATFORM_RUNTIME_TENANT_ID, prompt_key=prompt_key) or ""
            ).strip()
        if not template:
            raise RuntimeError(f"active prompt release not found: {prompt_key}")
        try:
            return str(template).format(**kwargs).strip()
        except Exception as exc:
            raise RuntimeError(f"active prompt release render failed: {prompt_key}") from exc