"""Toy vectors test database mathematics and isolation, never model/domain quality."""

from dataclasses import replace
from uuid import UUID, uuid4

import pytest
from fastapi import HTTPException
from sqlalchemy import func, select, text, update
from sqlalchemy.exc import DBAPIError, OperationalError
from sqlalchemy.orm import Session
from test_indexing import build, publish
from test_retrieval import add_item

from app.models.indexing import Chunk
from app.models.retrieval import RetrievalRun
from app.models.vectors import ChunkVector, EmbeddingSpace, VectorPublication, VectorVersion
from app.services import vectors
from app.services.embedding import EmbeddingError, EmbeddingSpec
from app.services.embedding_runtime import EmbeddingRuntime, get_embedding_provider
from app.services.indexing import locked_state, publish_index


class ToyProvider:
    spec = EmbeddingSpec("test-only-coordinate-fixture", "a" * 40, 3)
    callback = None

    def encode(self, texts, *, query=False):
        if self.callback:
            self.callback(query)
        return [[1.0, 0.0, 0.0] if query or "alpha" in t else [0.0, 1.0, 0.0] for t in texts]


@pytest.fixture
def provider(client):
    value = ToyProvider()
    client.app.dependency_overrides[get_embedding_provider] = lambda: value
    return value


def text_version(client, **kwargs):
    version = build(client, chunk_size=100, overlap=0, **kwargs)
    generation = client.get("/api/indexes/active").json()["generation"]
    assert publish(client, version, generation).status_code == 200
    return version


def vector_build(client, version):
    response = client.post("/api/vector-indexes", json={"text_index_version_id": version["id"]})
    assert response.status_code == 201, response.text
    return response.json()


def vector_publish(client, version, generation=0):
    return client.post(
        f"/api/vector-indexes/{version['id']}/publish", json={"expected_generation": generation}
    )


def vector_search(client, **kwargs):
    return client.post("/api/retrieval/vector", json={"query": "alpha query", **kwargs})


def test_exact_cosine_original_chunks_category_and_mixed_history(client, provider):
    first = add_item(client, content="alpha 占位正文", category="交通")
    second = add_item(client, content="beta 占位正文", category="其他")
    third = add_item(client, content="gamma 占位正文", category="其他")
    add_item(client, status="pending", content="alpha 待审核不应进入")
    text_ = text_version(client)
    version = vector_build(client, text_)
    assert (
        version["status"] == "ready" and version["dimension"] == 3 and version["chunk_count"] == 3
    )
    assert vector_search(client).status_code == 409
    assert vector_publish(client, version).json()["is_current"]
    result = vector_search(client).json()
    assert result["mode"] == "vector" and result["total_hits"] == 3
    assert [h["knowledge_id"] for h in result["results"]] == [
        first["id"],
        second["id"],
        third["id"],
    ]
    assert [h["score"] for h in result["results"]] == [1.0, 0.0, 0.0]
    assert result["config"]["space"]["id"] == provider.spec.space_id
    assert result["text_index_version_id"] == text_["id"]
    assert result["vector_version_id"] == version["id"]
    assert result["corpus_fingerprint"] == text_["corpus_fingerprint"]
    assert result["elapsed_ms"] > 0
    assert result["results"][0]["snippet"] == first["content"]
    assert result["results"][0]["snippet_start"] == 0
    assert result["results"][0]["snippet_end"] == len(first["content"])
    assert result["results"][0]["source_url"] == "https://example.com/"
    filtered = vector_search(client, category="其他", top_k=1).json()
    assert filtered["corpus_size"] == 2 and filtered["results"][0]["knowledge_id"] == second["id"]
    empty = vector_search(client, category="不存在分类").json()
    assert empty["status"] == "no_evidence" and empty["results"] == []
    keyword = client.post("/api/retrieval", json={"query": "alpha"}).json()
    history = client.get("/api/retrieval/runs?page_size=100").json()
    assert {r["mode"] for r in history["items"]} == {"keyword", "vector"}
    assert client.get(f"/api/retrieval/runs/{result['run_id']}").json() == result
    assert client.get(f"/api/retrieval/runs/{keyword['run_id']}").json() == keyword
    client.cookies.clear()
    assert vector_search(client).status_code == 200
    assert client.get("/api/retrieval/runs").status_code == 401


def test_multiple_chunks_keep_original_offsets_and_order(client, provider):
    item = add_item(client, content="alpha 占位正文。" * 45)
    version = vector_build(client, text_version(client))
    vector_publish(client, version)
    hits = vector_search(client, top_k=20).json()["results"]
    assert len(hits) > 1
    assert [h["chunk_ordinal"] for h in hits] == list(range(len(hits)))
    assert len({h["chunk_id"] for h in hits}) == len(hits)
    for h in hits:
        assert h["snippet"] == item["content"][h["snippet_start"] : h["snippet_end"]]


def test_same_dimension_different_revision_is_a_separate_space(client, provider, migrated_database):
    add_item(client)
    text_ = text_version(client)
    first = vector_build(client, text_)
    vector_publish(client, first)
    old_space = provider.spec.space_id
    provider.spec = replace(provider.spec, revision="b" * 40)
    assert vector_search(client).status_code == 409
    second = vector_build(client, text_)
    assert second["space_id"] != old_space and second["dimension"] == first["dimension"]
    assert vector_publish(client, second, 1).status_code == 200
    assert vector_search(client).json()["config"]["space"]["id"] == provider.spec.space_id
    assert client.get("/api/vector-indexes/spaces").json()["total"] == 2
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 2
        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 2


@pytest.mark.parametrize("bad", ["dimension", "zero", "nan", "count", "overflow", "unavailable"])
def test_invalid_or_failed_encode_never_persists_partial_vectors(
    client, provider, migrated_database, bad
):
    add_item(client, content="alpha 占位")
    text_ = text_version(client)
    original = vector_build(client, text_)
    vector_publish(client, original)
    before = client.get("/api/vector-indexes/active").json()

    def fail(texts, **kwargs):
        if bad == "overflow":
            raise EmbeddingError("token limit")
        if bad == "unavailable":
            raise OSError("private-path-must-not-appear")
        return {
            "dimension": [[1, 0]],
            "zero": [[0, 0, 0]],
            "nan": [[float("nan"), 0, 0]],
            "count": [],
        }[bad]

    provider.encode = fail
    response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]})
    assert response.status_code == (503 if bad == "unavailable" else 422)
    assert "private-path" not in response.text
    assert client.get("/api/vector-indexes/active").json() == before
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 1
        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 1


def test_build_and_query_encode_release_locks_but_recheck_catalog(
    client, provider, migrated_database
):
    item = add_item(client, content="alpha 占位")
    text_ = text_version(client)
    original = vector_build(client, text_)
    vector_publish(client, original)

    def mutate(query):
        with Session(migrated_database) as other:
            other.execute(text("SET LOCAL lock_timeout = '100ms'"))
            state = locked_state(other)  # Must succeed during both document and query inference.
            state.catalog_revision += 1
            other.execute(
                text("UPDATE knowledge_items SET content=:value WHERE id=:id"),
                {"value": "修订占位", "id": item["id"]},
            )
            other.commit()

    provider.callback = mutate
    response = client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]})
    assert response.status_code == 409
    assert not client.get("/api/vector-indexes/active").json()["is_current"]
    assert vector_search(client).status_code == 409
    # Restore an active text/vector snapshot, then race during query encoding too.
    provider.callback = None
    text_ = text_version(client)
    replacement = vector_build(client, text_)
    vector_publish(client, replacement, 1)
    provider.callback = mutate
    assert vector_search(client).status_code == 409
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 2
        assert db.scalar(select(func.count()).select_from(RetrievalRun)) == 0


def test_query_rejects_publication_changed_during_encode(client, provider, migrated_database):
    add_item(client)
    text_ = text_version(client)
    first, second = vector_build(client, text_), vector_build(client, text_)
    vector_publish(client, first)

    def switch(query):
        if query:
            with Session(migrated_database) as other:
                vectors.publish_vectors(other, UUID(second["id"]), 1, second["created_by"])

    provider.callback = switch
    assert vector_search(client).status_code == 409
    assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == second["id"]


def test_text_switch_during_build_discards_new_vectors(client, provider, migrated_database):
    add_item(client)
    first = text_version(client)
    replacement = build(client)

    def switch(query):
        with Session(migrated_database) as other:
            publish_index(other, UUID(replacement["id"]), 1, replacement["created_by"])

    provider.callback = switch
    assert (
        client.post("/api/vector-indexes", json={"text_index_version_id": first["id"]}).status_code
        == 409
    )
    assert client.get("/api/vector-indexes").json()["total"] == 0


def test_out_of_band_evidence_change_fails_closed(client, provider, migrated_database):
    item = add_item(client)
    text_ = text_version(client)
    version = vector_build(client, text_)
    vector_publish(client, version)
    with migrated_database.begin() as c:
        c.execute(
            text("UPDATE knowledge_items SET content='SQL 更新占位' WHERE id=:id"),
            {"id": item["id"]},
        )
    assert not client.get("/api/vector-indexes/active").json()["is_current"]
    assert vector_search(client).status_code == 409
    assert vector_publish(client, version, 1).status_code == 409


def test_optimistic_publication_audit_and_rollback(client, provider, migrated_database):
    add_item(client)
    text_ = text_version(client)
    first, second = vector_build(client, text_), vector_build(client, text_)
    assert vector_publish(client, first).json()["generation"] == 1
    assert vector_publish(client, second).status_code == 409
    assert vector_publish(client, first, 1).json()["generation"] == 1
    with migrated_database.begin() as c:
        c.execute(
            text(
                "ALTER TABLE vector_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID"
            )
        )
    assert vector_publish(client, second, 1).status_code == 409
    assert client.get("/api/vector-indexes/active").json()["active_version"]["id"] == first["id"]
    assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "published"
    assert client.get(f"/api/vector-indexes/{second['id']}").json()["status"] == "ready"
    assert client.get("/api/vector-indexes/publications").json()["total"] == 1
    with migrated_database.begin() as c:
        c.execute(text("ALTER TABLE vector_publications DROP CONSTRAINT test_reject_publication"))
    assert vector_publish(client, second, 1).json()["generation"] == 2
    assert client.get(f"/api/vector-indexes/{first['id']}").json()["status"] == "superseded"


def test_vector_insert_failure_rolls_back_entire_build(client, provider, migrated_database):
    add_item(client)
    text_ = text_version(client)
    with migrated_database.begin() as c:
        c.execute(
            text(
                "ALTER TABLE chunk_vectors ADD CONSTRAINT test_reject_vector CHECK (false) NOT VALID"
            )
        )
    assert (
        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
        == 409
    )
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(VectorVersion)) == 0
        assert db.scalar(select(func.count()).select_from(ChunkVector)) == 0
        assert db.scalar(select(func.count()).select_from(EmbeddingSpace)) == 0


@pytest.mark.parametrize(
    "column,value",
    [
        ("dimension", 2),
        ("space_id", "0" * 64),
        ("text_index_version_id", str(uuid4())),
        ("embedding", [1, 0]),
        ("embedding", [2, 0, 0]),
        ("embedding", [float("nan"), 0, 0]),
    ],
)
def test_postgresql_rejects_cross_scope_dimensions_and_nonunit_values(
    client, provider, migrated_database, column, value
):
    add_item(client)
    version = vector_build(client, text_version(client))
    with Session(migrated_database) as db:
        with pytest.raises(DBAPIError):
            db.execute(
                update(ChunkVector)
                .where(ChunkVector.version_id == version["id"])
                .values(**{column: value})
            )
            db.commit()
        db.rollback()
        row = db.scalar(select(ChunkVector))
        assert row.space_id == provider.spec.space_id and row.dimension == 3


def test_same_width_cross_space_and_cross_text_chunks_are_rejected_by_database(
    client, provider, migrated_database
):
    add_item(client)
    original_text = text_version(client)
    first = vector_build(client, original_text)
    provider.spec = replace(provider.spec, revision="c" * 40)
    second = vector_build(client, original_text)
    newer_text = text_version(client)
    with Session(migrated_database) as db:
        other_chunk = db.scalar(select(Chunk.id).where(Chunk.index_version_id == newer_text["id"]))
        for fields in [{"space_id": second["space_id"]}, {"chunk_id": other_chunk}]:
            with pytest.raises(DBAPIError):
                db.execute(
                    update(ChunkVector)
                    .where(ChunkVector.version_id == first["id"])
                    .values(**fields)
                )
                db.commit()
            db.rollback()


def test_incomplete_vector_set_cannot_be_published_or_queried(client, provider, migrated_database):
    add_item(client)
    version = vector_build(client, text_version(client))
    vector_publish(client, version)
    with migrated_database.begin() as c:
        c.execute(text("DELETE FROM chunk_vectors"))
    assert vector_publish(client, version, 1).status_code == 409
    assert vector_search(client).status_code == 409


def test_only_current_text_and_bounded_complete_snapshots_can_build(
    client, provider, migrated_database, monkeypatch
):
    assert vector_search(client).status_code == 409
    add_item(client)
    unpublished = build(client)
    assert (
        client.post(
            "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]}
        ).status_code
        == 409
    )
    text_ = text_version(client)
    assert (
        client.post(
            "/api/vector-indexes", json={"text_index_version_id": unpublished["id"]}
        ).status_code
        == 409
    )
    monkeypatch.setattr(vectors, "MAX_VECTOR_CHUNKS", 0)
    assert (
        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
        == 503
    )
    monkeypatch.undo()
    with migrated_database.begin() as c:
        c.execute(text("DELETE FROM chunks"))
    assert (
        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
        == 409
    )


def test_empty_reviewed_corpus_can_be_published_without_fabricated_results(client, provider):
    version = vector_build(client, text_version(client))
    assert version["chunk_count"] == 0
    assert vector_publish(client, version).json()["is_current"]
    result = vector_search(client).json()
    assert (
        result["status"] == "no_evidence" and result["results"] == [] and result["corpus_size"] == 0
    )


def test_metadata_and_build_require_admin_and_csrf(client, provider):
    add_item(client)
    version = vector_build(client, text_version(client))
    assert (
        client.post(
            "/api/vector-indexes",
            json={"text_index_version_id": version["text_index_version_id"]},
            headers={"X-CSRF-Token": ""},
        ).status_code
        == 403
    )
    client.cookies.clear()
    for method, path in [
        ("GET", "/api/vector-indexes"),
        ("GET", "/api/vector-indexes/active"),
        ("GET", "/api/vector-indexes/spaces"),
        ("GET", "/api/vector-indexes/publications"),
        ("GET", f"/api/vector-indexes/{version['id']}"),
        ("POST", "/api/vector-indexes"),
        ("POST", f"/api/vector-indexes/{version['id']}/publish"),
    ]:
        assert client.request(method, path, json={}).status_code == 401
    assert (
        client.post(
            "/api/auth/register", json={"username": "vector_user", "password": "test-password-123"}
        ).status_code
        == 201
    )
    auth = client.post(
        "/api/auth/login", json={"username": "vector_user", "password": "test-password-123"}
    ).json()
    client.headers["X-CSRF-Token"] = auth["csrf_token"]
    assert client.get("/api/vector-indexes").status_code == 403
    assert client.post("/api/vector-indexes", json={}).status_code == 403


def test_vector_publication_serializes_with_catalog_and_other_publishers(
    client, provider, migrated_database
):
    add_item(client)
    text_ = text_version(client)
    first, second = vector_build(client, text_), vector_build(client, text_)
    with Session(migrated_database) as owner, Session(migrated_database) as other:
        locked_state(owner)
        other.execute(text("SET LOCAL lock_timeout = '100ms'"))
        with pytest.raises(OperationalError) as blocked:
            vectors.publish_vectors(other, UUID(first["id"]), 0, first["created_by"])
        assert blocked.value.orig.sqlstate == "55P03"
        other.rollback()
        vectors.publish_vectors(owner, UUID(first["id"]), 0, first["created_by"])
        with pytest.raises(HTTPException) as stale:
            vectors.publish_vectors(other, UUID(second["id"]), 0, second["created_by"])
        assert stale.value.status_code == 409
        other.rollback()
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(VectorPublication)) == 1


def test_lazy_runtime_is_offline_and_does_not_wait_when_busy(monkeypatch):
    from app.services import embedding_runtime

    calls = []

    def load(spec, **kwargs):
        calls.append(kwargs)
        return ToyProvider()

    monkeypatch.setattr(embedding_runtime, "LocalEmbeddingProvider", load)
    runtime = EmbeddingRuntime(ToyProvider.spec, None, 8, 2)
    assert calls == []
    assert runtime.encode(["alpha"]) == [[1, 0, 0]]
    assert calls[0]["offline"] is True
    runtime.encode(["alpha"], query=True)
    assert len(calls) == 1
    with runtime._lock:
        with pytest.raises(HTTPException) as busy:
            runtime.encode(["alpha"])
        assert busy.value.status_code == 503


def test_disabled_runtime_is_explicit_and_keyword_baseline_remains_available(client):
    assert vector_search(client).status_code == 503
    assert client.post("/api/retrieval", json={"query": "那拉提"}).status_code == 200


def test_different_width_spaces_coexist_and_queries_never_compare_across_widths(client, provider):
    add_item(client)
    text_ = text_version(client)
    first = vector_build(client, text_)
    vector_publish(client, first)
    provider.spec = replace(provider.spec, dimension=2)
    provider.encode = lambda texts, **kwargs: [[1, 0] for _ in texts]
    second = vector_build(client, text_)
    assert vector_publish(client, second, 1).status_code == 200
    result = vector_search(client)
    assert result.status_code == 200, result.text
    assert result.json()["total_hits"] == 1 and result.json()["results"][0]["score"] == 1
    assert client.get("/api/vector-indexes/spaces").json()["total"] == 2


def test_new_text_publication_requires_corresponding_vectors_and_keeps_history(client, provider):
    add_item(client)
    old_text = text_version(client)
    old = vector_build(client, old_text)
    vector_publish(client, old)
    history = vector_search(client).json()
    new_text = text_version(client)
    active = client.get("/api/vector-indexes/active").json()
    assert not active["is_current"] and active["active_version"]["id"] == old["id"]
    assert vector_search(client).status_code == 409
    assert vector_publish(client, old, 1).status_code == 409
    replacement = vector_build(client, new_text)
    assert vector_publish(client, replacement, 1).json()["is_current"]
    assert client.get(f"/api/retrieval/runs/{history['run_id']}").json() == history
    assert client.get("/api/vector-indexes/publications").json()["total"] == 2


def test_snapshot_text_corruption_cannot_be_encoded_published_or_served(
    client, provider, migrated_database
):
    add_item(client)
    text_ = text_version(client)
    version = vector_build(client, text_)
    vector_publish(client, version)
    with migrated_database.begin() as c:
        c.execute(text("UPDATE chunks SET text='篡改占位'"))
    assert not client.get("/api/vector-indexes/active").json()["is_current"]
    assert (
        client.post("/api/vector-indexes", json={"text_index_version_id": text_["id"]}).status_code
        == 409
    )
    assert vector_publish(client, version, 1).status_code == 409
    assert vector_search(client).status_code == 409


def test_request_shape_pagination_and_missing_vector_version(client, provider):
    assert (
        client.post(
            "/api/vector-indexes", json={"text_index_version_id": str(uuid4()), "embedding": [1, 0]}
        ).status_code
        == 422
    )
    assert vector_search(client, top_k=21).status_code == 422
    assert vector_search(client, mode="keyword").status_code == 422
    assert vector_search(client, query="   ").status_code == 422
    assert client.get("/api/vector-indexes?page=0").status_code == 422
    assert client.get(f"/api/vector-indexes/{uuid4()}").status_code == 404
    assert client.get("/api/vector-indexes/spaces").json()["total"] == 0


def test_blank_optional_cache_path_is_not_the_working_directory():
    from app.core.config import Settings

    assert Settings(embedding_cache_dir="", _env_file=None).embedding_cache_dir is None