"""内存替身：项目/章节 repo + 写章网关（端点测试用，无 DB/无网络）。

绝对导入 `from fakes_projects import ...`（包目录无 __init__.py，见 memory/gotchas）。
全局唯一名（避免跨包同名碰撞）。
"""

from __future__ import annotations

import uuid
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from typing import Any

from pydantic import BaseModel
from ww_core.domain import ForeshadowLedgerView, OutlineWriteView, StyleFingerprintView
from ww_core.domain.chapter_repo import ChapterDraftView, ChapterView
from ww_core.domain.foreshadow_state import (
    ForeshadowStatus,
    apply_overdue_scan,
)
from ww_core.domain.foreshadow_state import (
    transition as apply_transition,
)
from ww_core.domain.job_repo import (
    PROGRESS_COMPLETE,
    STATUS_AWAITING,
    STATUS_DONE,
    STATUS_FAILED,
    STATUS_QUEUED,
    STATUS_RUNNING,
    JobView,
)
from ww_core.domain.project_repo import ProjectCreate, ProjectView
from ww_core.domain.repositories import DigestView, OutlineView
from ww_core.domain.review_repo import ReviewView
from ww_llm_gateway.types import Delta, LlmRequest, LlmResponse, ServedBy, Usage


class FakeProjectRepo:
    """实现 `ProjectRepo` Protocol 的内存版（按 owner_id 隔离）。"""

    def __init__(self) -> None:
        self.rows: dict[uuid.UUID, tuple[uuid.UUID, ProjectView]] = {}
        self.now = datetime(2026, 6, 28, tzinfo=UTC)

    async def create(self, owner_id: uuid.UUID, data: ProjectCreate) -> ProjectView:
        pid = uuid.uuid4()
        view = ProjectView(
            id=pid,
            title=data.title,
            genre=data.genre,
            logline=data.logline,
            premise=data.premise,
            theme=data.theme,
            selling_points=list(data.selling_points),
            structure=data.structure,
            tone=data.tone,
            ending_type=data.ending_type,
            narrative_pov=data.narrative_pov,
            updated_at=self.now,
        )
        self.rows[pid] = (owner_id, view)
        return view

    async def list_for_owner(self, owner_id: uuid.UUID) -> list[ProjectView]:
        return [v for (o, v) in self.rows.values() if o == owner_id]

    async def get(self, owner_id: uuid.UUID, project_id: uuid.UUID) -> ProjectView | None:
        entry = self.rows.get(project_id)
        if entry is None or entry[0] != owner_id:
            return None
        return entry[1]


class FakeChapterRepo:
    """实现 `ChapterRepo` Protocol：幂等覆盖同 (project_id, chapter_no) 草稿。"""

    def __init__(self) -> None:
        self.drafts: dict[tuple[uuid.UUID, int], ChapterDraftView] = {}
        self.accepted_versions: dict[tuple[uuid.UUID, int], int] = {}
        self.accepted_content: dict[tuple[uuid.UUID, int, int], str] = {}

    async def save_draft(
        self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1
    ) -> ChapterDraftView:
        view = ChapterDraftView(
            project_id=project_id,
            chapter_no=chapter_no,
            volume=volume,
            content=text,
            status="draft",
            version=1,
        )
        self.drafts[(project_id, chapter_no)] = view
        return view

    async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> ChapterDraftView | None:
        return self.drafts.get((project_id, chapter_no))

    async def max_version(self, project_id: uuid.UUID, chapter_no: int) -> int:
        versions = [
            v for (p, c), v in self.accepted_versions.items() if p == project_id and c == chapter_no
        ]
        draft = 1 if (project_id, chapter_no) in self.drafts else 0
        return max([*versions, draft], default=0)

    async def promote_to_accepted(
        self, project_id: uuid.UUID, chapter_no: int, *, content: str, volume: int = 1
    ) -> ChapterView:
        next_version = (await self.max_version(project_id, chapter_no)) + 1
        self.accepted_versions[(project_id, chapter_no)] = next_version
        self.accepted_content[(project_id, chapter_no, next_version)] = content
        return ChapterView(
            project_id=project_id,
            chapter_no=chapter_no,
            volume=volume,
            content=content,
            status="accepted",
            version=next_version,
        )

    async def latest_accepted(self, project_id: uuid.UUID, chapter_no: int) -> ChapterView | None:
        ver = self.accepted_versions.get((project_id, chapter_no))
        if ver is None:
            return None
        return ChapterView(
            project_id=project_id,
            chapter_no=chapter_no,
            volume=1,
            content=self.accepted_content[(project_id, chapter_no, ver)],
            status="accepted",
            version=ver,
        )


class FakeReviewRepo:
    """实现 `ReviewRepo` Protocol（内存留痕 + 历史 + 裁决；只 flush 语义无 DB）。"""

    def __init__(self, *, fail_set_decisions: bool = False) -> None:
        self.rows: list[ReviewView] = []
        self._fail_set_decisions = fail_set_decisions

    async def record(
        self,
        project_id: uuid.UUID,
        chapter_no: int,
        *,
        chapter_version: int | None = None,
        conflicts: list[dict[str, Any]],
        foreshadow_sug: list[dict[str, Any]] | None = None,
        style: dict[str, Any] | None = None,
        pace: dict[str, Any] | None = None,
        characterization: dict[str, Any] | None = None,
        health_score: int | None = None,
    ) -> ReviewView:
        view = ReviewView(
            id=uuid.uuid4(),
            project_id=project_id,
            chapter_no=chapter_no,
            chapter_version=chapter_version,
            conflicts=[dict(c) for c in conflicts],
            foreshadow_sug=[dict(s) for s in (foreshadow_sug or [])],
            style=style,
            pace=pace,
            characterization=characterization,
            health_score=health_score,
        )
        # 新→旧：插到列表头部。
        self.rows.insert(0, view)
        return view

    async def list_for_chapter(self, project_id: uuid.UUID, chapter_no: int) -> list[ReviewView]:
        return [r for r in self.rows if r.project_id == project_id and r.chapter_no == chapter_no]

    async def set_decisions(self, review_id: uuid.UUID, *, decisions: dict[str, Any]) -> ReviewView:
        if self._fail_set_decisions:
            raise RuntimeError("boom: set_decisions failed (transaction rollback test)")
        for i, r in enumerate(self.rows):
            if r.id == review_id:
                updated = r.model_copy(update={"decisions": dict(decisions)})
                self.rows[i] = updated
                return updated
        raise LookupError(f"review not found: {review_id}")


class FakeDigestAppendRepo:
    """实现 `DigestAppendRepo` Protocol（内存 append-only）。"""

    def __init__(self) -> None:
        self.rows: list[tuple[uuid.UUID, int, dict[str, Any]]] = []

    async def append(
        self, project_id: uuid.UUID, chapter_no: int, *, facts: dict[str, Any]
    ) -> DigestView:
        self.rows.append((project_id, chapter_no, dict(facts)))
        return DigestView(chapter_no=chapter_no, facts=dict(facts))


class FakeSession:
    """最小 fake：记录 commit/rollback 次数（验收事务 + 端点提交边界断言）。"""

    def __init__(self) -> None:
        self.commits = 0
        self.rollbacks = 0

    async def commit(self) -> None:
        self.commits += 1

    async def rollback(self) -> None:
        self.rollbacks += 1


class FakeForeshadowRepo:
    """实现 `ForeshadowLedgerRepo` Protocol（内存；状态机/进展/扫描复用纯函数）。

    `raise_integrity_on` 模拟重复 code 的 DB 唯一约束冲突（register 抛 IntegrityError）。
    """

    def __init__(self, *, raise_integrity_on: set[str] | None = None) -> None:
        self.rows: dict[tuple[uuid.UUID, str], ForeshadowLedgerView] = {}
        self._raise_integrity_on = raise_integrity_on or set()

    async def register(
        self,
        project_id: uuid.UUID,
        *,
        code: str,
        title: str,
        planted_at: int | None = None,
        content: str | None = None,
        expected_close_from: int | None = None,
        expected_close_to: int | None = None,
        importance: str | None = None,
    ) -> ForeshadowLedgerView:
        if code in self._raise_integrity_on or (project_id, code) in self.rows:
            from sqlalchemy.exc import IntegrityError

            raise IntegrityError("duplicate code", None, Exception("unique"))
        view = ForeshadowLedgerView(
            code=code,
            title=title,
            status=ForeshadowStatus.OPEN.value,
            planted_at=planted_at,
            content=content,
            expected_close_from=expected_close_from,
            expected_close_to=expected_close_to,
            importance=importance,
        )
        self.rows[(project_id, code)] = view
        return view

    async def get(self, project_id: uuid.UUID, code: str) -> ForeshadowLedgerView | None:
        return self.rows.get((project_id, code))

    async def list_by_status(
        self, project_id: uuid.UUID, status: str | None = None
    ) -> list[ForeshadowLedgerView]:
        rows = [v for (p, _), v in self.rows.items() if p == project_id]
        if status is not None:
            rows = [v for v in rows if v.status == status]
        return sorted(rows, key=lambda v: v.code)

    async def transition(
        self, project_id: uuid.UUID, code: str, *, to_status: str
    ) -> ForeshadowLedgerView:
        row = self.rows.get((project_id, code))
        if row is None:
            raise LookupError(f"foreshadow not found: {code}")
        new_status = apply_transition(
            ForeshadowStatus(row.status), ForeshadowStatus(to_status)
        ).value
        updated = row.model_copy(update={"status": new_status})
        self.rows[(project_id, code)] = updated
        return updated

    async def record_progress(
        self, project_id: uuid.UUID, code: str, *, entry: dict[str, Any]
    ) -> ForeshadowLedgerView:
        row = self.rows.get((project_id, code))
        if row is None:
            raise LookupError(f"foreshadow not found: {code}")
        updated = row.model_copy(update={"progress": [*row.progress, dict(entry)]})
        self.rows[(project_id, code)] = updated
        return updated

    async def scan_overdue(
        self, project_id: uuid.UUID, *, current_chapter: int
    ) -> list[ForeshadowLedgerView]:
        changed: list[ForeshadowLedgerView] = []
        for key, row in list(self.rows.items()):
            new_status = apply_overdue_scan(
                ForeshadowStatus(row.status),
                current_chapter=current_chapter,
                expected_close_to=row.expected_close_to,
            ).value
            if new_status != row.status:
                updated = row.model_copy(update={"status": new_status})
                self.rows[key] = updated
                changed.append(updated)
        return changed


class FakeOutlineWriteRepo:
    """实现 `OutlineWriteRepo` Protocol（内存；幂等覆盖同 (project_id, chapter_no)）。"""

    def __init__(self) -> None:
        # (project_id, chapter_no) -> (volume, beats, foreshadow_windows)
        self.rows: dict[tuple[uuid.UUID, int], tuple[int, list[str], list[dict[str, Any]]]] = {}
        self.upsert_calls = 0

    async def upsert_chapter(
        self,
        project_id: uuid.UUID,
        *,
        volume: int,
        chapter_no: int,
        beats: list[str],
        foreshadow_windows: list[dict[str, Any]],
    ) -> OutlineWriteView:
        self.upsert_calls += 1
        windows = [dict(w) for w in foreshadow_windows]
        self.rows[(project_id, chapter_no)] = (volume, list(beats), windows)
        return OutlineWriteView(
            chapter_no=chapter_no,
            volume=volume,
            beats=list(beats),
            foreshadow_windows=windows,
        )


class FakeOutlineReadRepo:
    """实现读侧 `OutlineRepo` Protocol（内存；DB beats 形 `{"beats": [...]}`）。

    `add_chapter` 以裸 `list[str]` 便利入种，内部包成 dict（与 DB JSONB round-trip 一致）。
    `list_for_project` 按 chapter_no 升序返回（仿 SqlOutlineRepo 的 order_by）。
    """

    def __init__(self) -> None:
        self.rows: dict[tuple[uuid.UUID, int], OutlineView] = {}

    def add_chapter(
        self,
        project_id: uuid.UUID,
        *,
        volume: int,
        chapter_no: int,
        beats: list[str],
        foreshadow_windows: list[dict[str, Any]] | None = None,
    ) -> None:
        self.rows[(project_id, chapter_no)] = OutlineView(
            volume=volume,
            chapter_no=chapter_no,
            beats={"beats": list(beats)},
            foreshadow_windows=[dict(w) for w in (foreshadow_windows or [])],
        )

    async def get(self, project_id: uuid.UUID, chapter_no: int) -> OutlineView | None:
        return self.rows.get((project_id, chapter_no))

    async def list_for_project(self, project_id: uuid.UUID) -> list[OutlineView]:
        views = [v for (p, _), v in self.rows.items() if p == project_id]
        return sorted(views, key=lambda v: v.chapter_no)


class FakeSessionFactory:
    """到期扫描的独立 session 工厂替身：`()` → async-CM 产 `FakeSession`。

    记录开了几次 session + 各次 commit，供「扫描自建 session 并提交」断言。
    """

    def __init__(self) -> None:
        self.sessions: list[FakeSession] = []

    def __call__(self) -> Any:
        session = FakeSession()
        self.sessions.append(session)

        @asynccontextmanager
        async def _cm() -> AsyncIterator[FakeSession]:
            yield session

        return _cm()


class FakeStyleWriteRepo:
    """实现 `StyleFingerprintWriteRepo` Protocol（内存版本化 append + 最新读取）。

    `append` 模拟 version=当前 max+1（首次 1）；`latest` 取最大 version 行。
    供 `GET /style` 读侧 + 学文风 work（mode=update → 第二次 append version+1）断言。
    """

    def __init__(self) -> None:
        # version -> (dimensions, evidence)
        self.rows: dict[int, tuple[dict[str, Any], dict[str, Any]]] = {}
        self.append_calls = 0

    async def append(
        self,
        project_id: uuid.UUID,
        *,
        dimensions_json: dict[str, Any],
        evidence_json: dict[str, Any],
    ) -> int:
        self.append_calls += 1
        next_version = (max(self.rows) if self.rows else 0) + 1
        self.rows[next_version] = (dict(dimensions_json), dict(evidence_json))
        return next_version

    async def latest(self, project_id: uuid.UUID) -> StyleFingerprintView | None:
        if not self.rows:
            return None
        version = max(self.rows)
        dimensions, evidence = self.rows[version]
        return StyleFingerprintView(dimensions=dimensions, evidence=evidence, version=version)


class FakeJobRepo:
    """实现 `JobRepo` Protocol（内存）：create/get/状态流转，供 `POST /style` 断言。"""

    def __init__(self) -> None:
        self.rows: dict[uuid.UUID, JobView] = {}

    async def create(self, project_id: uuid.UUID | None, kind: str) -> JobView:
        view = JobView(
            id=uuid.uuid4(),
            project_id=project_id,
            kind=kind,
            status=STATUS_QUEUED,
            progress=0,
        )
        self.rows[view.id] = view
        return view

    async def set_running(self, job_id: uuid.UUID) -> JobView:
        view = self.rows[job_id].model_copy(update={"status": STATUS_RUNNING})
        self.rows[job_id] = view
        return view

    async def set_progress(self, job_id: uuid.UUID, pct: int) -> JobView:
        view = self.rows[job_id].model_copy(update={"progress": max(0, min(100, pct))})
        self.rows[job_id] = view
        return view

    async def complete(self, job_id: uuid.UUID, result: dict[str, Any]) -> JobView:
        view = self.rows[job_id].model_copy(
            update={"status": STATUS_DONE, "progress": PROGRESS_COMPLETE, "result": dict(result)}
        )
        self.rows[job_id] = view
        return view

    async def set_awaiting(self, job_id: uuid.UUID, result: dict[str, Any]) -> JobView:
        view = self.rows[job_id].model_copy(
            update={"status": STATUS_AWAITING, "result": dict(result)}
        )
        self.rows[job_id] = view
        return view

    async def claim_awaiting_to_running(self, job_id: uuid.UUID) -> JobView | None:
        """原子抢占 awaiting→running（内存替身：仅当现态 awaiting 才抢到，否则 None）。"""
        current = self.rows.get(job_id)
        if current is None or current.status != STATUS_AWAITING:
            return None
        view = current.model_copy(update={"status": STATUS_RUNNING})
        self.rows[job_id] = view
        return view

    async def fail(self, job_id: uuid.UUID, error: str) -> JobView:
        view = self.rows[job_id].model_copy(update={"status": STATUS_FAILED, "error": error})
        self.rows[job_id] = view
        return view

    async def get(self, job_id: uuid.UUID) -> JobView | None:
        return self.rows.get(job_id)

    async def reap_zombies(self) -> int:
        return 0


class FakeReviewGateway:
    """实现审稿/digest 节点最小依赖（`run`）：吐固定结构化 `parsed`，绝不联网。"""

    def __init__(
        self,
        parsed: BaseModel | None = None,
        error: Exception | None = None,
        *,
        text: str | None = None,
    ) -> None:
        self._parsed = parsed
        self._error = error
        # `text` 覆盖纯文本产出（refiner output_schema=None → 端点消费 resp.text）。
        self._text = text
        self.requests: list[LlmRequest] = []

    async def run(self, req: LlmRequest) -> LlmResponse:
        self.requests.append(req)
        if self._error is not None:
            raise self._error
        if self._text is not None:
            out_text = self._text
        elif self._parsed is not None:
            out_text = self._parsed.model_dump_json()
        else:
            out_text = "{}"
        return LlmResponse(
            text=out_text,
            parsed=self._parsed,
            usage=Usage(
                provider="fake",
                model="fake",
                input_tokens=1,
                output_tokens=1,
                cost_minor=0,
                currency="USD",
            ),
            served_by=ServedBy(provider="fake", model="fake"),
        )


class FakeWriterGateway:
    """实现 write 节点最小依赖（`stream`）：吐固定 `Delta` 序列，绝不联网。

    可注入 `error` 以验证 SSE 归一层把异常归一为 error 事件。
    """

    def __init__(self, chunks: list[str] | None = None, error: Exception | None = None) -> None:
        self._chunks = chunks if chunks is not None else ["第一段。", "第二段。"]
        self._error = error
        self.requests: list[LlmRequest] = []

    async def stream(self, req: LlmRequest) -> AsyncIterator[Delta]:
        self.requests.append(req)
        for c in self._chunks:
            yield Delta(text=c)
        if self._error is not None:
            raise self._error