"""知识入库服务契约测试。

入库链路：文件落存储 → 元数据落库 → 切分 chunk → 逐块写 `fin_knowledge_meta`
→ 投递 `knowledge.vector_sync_requested`（Milvus 写入由 Outbox Worker 负责，本层不连 Milvus）。

固定口径（Task 5 派发说明裁定 1）：`ingest` 返回 **`list[int]`**（一份文档切出多块即是多行知识），
集合名只能取白名单映射、调用方不得随意指定。
"""

import json
import re
from typing import Any

import pytest

from app.core.errors import ValidationAgentError
from app.service.document_parser import ParsedChunk
from app.service.knowledge_ingest_service import (
    TYPE_TO_COLLECTION,
    KnowledgeIngestService,
)

_TEXT_QUERY = re.compile(r"[A-Za-z_]+")


class FakeStorage:
    def __init__(self) -> None:
        self.saved: list[dict[str, Any]] = []

    async def save(self, *, key: str, content: bytes, content_type: str) -> str:
        self.saved.append({"key": key, "content": content, "content_type": content_type})
        return key


class FakeParser:
    def __init__(self, chunks: list[ParsedChunk]) -> None:
        self._chunks = chunks
        self.calls: list[dict[str, Any]] = []

    def parse(self, *, filename: str, content: bytes) -> list[ParsedChunk]:
        self.calls.append({"filename": filename, "content": content})
        return list(self._chunks)


class FakeSession:
    """记录每条带参数的语句；`SELECT LAST_INSERT_ID()` 按预设序列返回。"""

    def __init__(self, ids: list[Any] | None = None) -> None:
        self._ids = list(ids if ids is not None else [])
        self.statements: list[tuple[str, dict[str, Any]]] = []
        self.commits = 0

    async def execute(self, statement: Any, params: Any = None) -> None:
        self.statements.append((str(statement), dict(params or {})))
        return None

    async def scalar(self, statement: Any, params: Any = None) -> Any:
        self.statements.append((str(statement), dict(params or {})))
        return self._ids.pop(0) if self._ids else None

    async def commit(self) -> None:  # pragma: no cover - 事务归调用方
        self.commits += 1


class RecordingEmbedder:
    async def embed(self, endpoints: Any, text: str, *, max_attempts: int = 2) -> Any:
        raise AssertionError("入库服务不得自己调用 embedding（向量由 Outbox Worker 负责）")


class RecordingResolver:
    async def resolve(self, *, agent_type: str, task_type: str) -> list[Any]:
        raise AssertionError("入库服务不得自己解析模型端点（交给 Outbox Worker）")


def _chunks() -> list[ParsedChunk]:
    return [
        ParsedChunk(text="第一块正文", heading_path=("申购",), index=0),
        ParsedChunk(text="第二块正文", heading_path=("申购", "确认"), index=1),
    ]


def _service(
    *,
    ids: list[Any] | None = None,
    chunks: list[ParsedChunk] | None = None,
    created_by: int | None = 9003,
) -> tuple[KnowledgeIngestService, FakeSession, FakeStorage, FakeParser]:
    """默认给 2 个 chunk 配 2 个自增 id（`LAST_INSERT_ID()` 的替身）。"""
    session = FakeSession(ids if ids is not None else [11, 12])
    storage = FakeStorage()
    parser = FakeParser(chunks if chunks is not None else _chunks())
    service = KnowledgeIngestService(
        session=session,  # type: ignore[arg-type]
        parser=parser,  # type: ignore[arg-type]
        storage=storage,
        embedder=RecordingEmbedder(),
        endpoint_resolver=RecordingResolver(),
        created_by=created_by,
    )
    return service, session, storage, parser


def _by_table(session: FakeSession, table: str) -> list[dict[str, Any]]:
    prefix = f"INSERT INTO {table}"
    return [params for sql, params in session.statements if sql.startswith(prefix)]


async def test_ingest_returns_one_id_per_chunk() -> None:
    """裁定 1：返回值是 list[int]，一份文档多块 → 多行知识、多条事件。"""
    service, session, _, _ = _service(ids=[11, 12])

    ids = await service.ingest(filename="faq.md", content=b"# \xe7\x94\xb3\xe8\xb4\xad",
                               knowledge_type="faq")

    assert ids == [11, 12]
    assert len(_by_table(session, "fin_knowledge_meta")) == 2
    assert len(_by_table(session, "domain_event_outbox")) == 2


async def test_ingest_writes_allowlisted_collection_and_chunk_tags() -> None:
    service, session, _, _ = _service(ids=[11, 12])

    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq",
                         collection="fin_faq_collection")

    rows = _by_table(session, "fin_knowledge_meta")
    assert {row["milvus_collection"] for row in rows} == {"fin_faq_collection"}
    first = json.loads(rows[0]["tags"])
    assert first["heading_path"] == ["申购"]
    assert first["chunk_index"] == 0
    assert json.loads(rows[1]["tags"])["chunk_index"] == 1
    assert rows[0]["title"] == "第一块正文"
    assert rows[0]["reviewer_id"] == 9003
    assert rows[1]["reviewer_id"] == 9003


@pytest.mark.parametrize("knowledge_type", ["faq", "product", "policy"])
async def test_every_allowlisted_type_maps_to_its_collection(knowledge_type: str) -> None:
    service, session, _, _ = _service(ids=[11, 12])

    await service.ingest(filename="a.txt", content=b"x", knowledge_type=knowledge_type)

    assert _by_table(session, "fin_knowledge_meta")[0]["milvus_collection"] == (
        TYPE_TO_COLLECTION[knowledge_type]
    )


async def test_ingest_rejects_collection_outside_the_type_map() -> None:
    """全局约束：集合名不得由调用方随意指定，只能取白名单。"""
    service, session, _, _ = _service()

    with pytest.raises(ValidationAgentError):
        await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq",
                             collection="fin_policy_collection")
    assert session.statements == []


async def test_ingest_rejects_unknown_knowledge_type() -> None:
    service, _, _, _ = _service()

    with pytest.raises(ValidationAgentError):
        await service.ingest(filename="faq.md", content=b"x", knowledge_type="secret")


async def test_ingest_enqueues_sync_event_with_string_knowledge_id() -> None:
    """生产口径：payload 与 aggregate_id 都是 `str(knowledge_id)`（裁定 4）。"""
    service, session, _, _ = _service(ids=[11, 12])

    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")

    events = _by_table(session, "domain_event_outbox")
    assert [event["aggregate_id"] for event in events] == ["11", "12"]
    assert [json.loads(event["payload"])["knowledge_id"] for event in events] == ["11", "12"]
    assert {event["event_type"] for event in events} == {"knowledge.vector_sync_requested"}
    assert {event["aggregate_type"] for event in events} == {"knowledge_meta"}


async def test_event_type_constant_is_shared_with_worker() -> None:
    from app.worker.knowledge_vector_worker import VECTOR_SYNC_EVENT

    service, session, _, _ = _service(ids=[11, 12])
    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")

    assert _by_table(session, "domain_event_outbox")[0]["event_type"] == VECTOR_SYNC_EVENT


async def test_ingest_saves_document_before_splitting() -> None:
    service, _, storage, parser = _service(ids=[11, 12])

    await service.ingest(filename=r"C:\kb\faq.md", content=b"body", knowledge_type="faq")

    assert storage.saved[0]["content"] == b"body"
    assert storage.saved[0]["key"].startswith("kb/faq/")
    assert storage.saved[0]["key"].endswith("faq.md")
    assert storage.saved[0]["key"].count("-") == 1
    assert parser.calls == [{"filename": r"C:\kb\faq.md", "content": b"body"}]


async def test_ingest_uses_caller_created_by_when_given() -> None:
    service, session, _, _ = _service(ids=[11, 12], created_by=9003)

    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq", created_by=9004)

    assert {row["reviewer_id"] for row in _by_table(session, "fin_knowledge_meta")} == {9004}


async def test_ingest_fails_when_no_created_by_anywhere() -> None:
    service, session, _, _ = _service(ids=[11, 12], created_by=None)

    with pytest.raises(ValidationAgentError):
        await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")
    assert session.statements == []


async def test_ingest_does_not_commit_or_embed() -> None:
    """事务与服务边界：ingest 只挂 SQL，commit 归调用方；向量化归 Outbox Worker。"""
    service, session, _, _ = _service(ids=[11, 12])
    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")
    assert session.commits == 0


async def test_empty_document_produces_no_rows() -> None:
    service, session, _, _ = _service(ids=[], chunks=[])

    assert await service.ingest(filename="empty.md", content=b"", knowledge_type="faq") == []
    assert session.statements == []


class FailingInsertSession(FakeSession):
    """模拟"INSERT 没生效"：`rowcount == 0`，且 `LAST_INSERT_ID()` 是 0。

    这与 `mysql` 真机行为一致 —— 插入 0 行时 `LAST_INSERT_ID()` 保持**上一次**的值
    （新连接上是 0），因此"拿 0"的真相是"插入没生效"，报错文案必须这么说。
    """

    async def execute(self, statement: Any, params: Any = None) -> Any:
        self.statements.append((str(statement), dict(params or {})))

        class Result:
            rowcount = 0

        return Result()


class ZeroIdSession(FakeSession):
    """模拟 `LAST_INSERT_ID()` 返回 0（连接上还没有过自增插入）但 INSERT 确实影响了 1 行。"""

    async def execute(self, statement: Any, params: Any = None) -> Any:
        self.statements.append((str(statement), dict(params or {})))

        class Result:
            rowcount = 1

        return Result()

    async def scalar(self, statement: Any, params: Any = None) -> Any:
        self.statements.append((str(statement), dict(params or {})))
        return 0


async def test_insert_not_effective_is_reported_as_such() -> None:
    """`rowcount == 0` 时必须在读 `LAST_INSERT_ID()` 之前就失败，且文案指向"插入没生效"。"""
    service, _, _, _ = _service(ids=[11, 12])
    service._session = FailingInsertSession([11, 12])  # type: ignore[assignment]

    with pytest.raises(ValidationAgentError, match="插入未生效"):
        await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")


async def test_zero_last_insert_id_is_reported_as_zero() -> None:
    """`LAST_INSERT_ID()` 为 0 时的文案必须点出 0，不能笼统说"取不到"（否则误导排障）。"""
    service, _, _, _ = _service(ids=[11, 12])
    service._session = ZeroIdSession([11, 12])  # type: ignore[assignment]

    with pytest.raises(ValidationAgentError, match="LAST_INSERT_ID"):
        await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")


async def test_last_insert_id_is_read_once_per_chunk_in_the_same_statement_order() -> None:
    """硬耦合护栏：每个 chunk 的 `INSERT` 后**紧随**一次 `LAST_INSERT_ID()`，中间不得有别的语句。

    这是 `LAST_INSERT_ID()` 是**连接级隐式状态**的直接后果：两者之间只要插入任何其它 SQL，
    拿到的就是别人的 id。本用例把"成对且相邻"固定下来，防止以后有人在中间加一条查询。
    """
    service, session, _, _ = _service(ids=[11, 12])

    await service.ingest(filename="faq.md", content=b"x", knowledge_type="faq")

    shape = [
        "INSERT INTO fin_knowledge_meta" if sql.startswith("INSERT INTO fin_knowledge_meta")
        else "SELECT LAST_INSERT_ID()" if "LAST_INSERT_ID" in sql
        else "INSERT INTO domain_event_outbox" if sql.startswith("INSERT INTO domain_event_outbox")
        else sql
        for sql, _ in session.statements
    ]
    assert shape == [
        "INSERT INTO fin_knowledge_meta",
        "SELECT LAST_INSERT_ID()",
        "INSERT INTO domain_event_outbox",
        "INSERT INTO fin_knowledge_meta",
        "SELECT LAST_INSERT_ID()",
        "INSERT INTO domain_event_outbox",
    ]