"""Versioned chunks remain evidence snapshots; publication uses real PostgreSQL locks."""

import hashlib
from uuid import UUID, uuid4

import pytest
from fastapi import HTTPException
from sqlalchemy import func, select, text
from sqlalchemy.exc import OperationalError
from sqlalchemy.orm import Session
from test_retrieval import add_item

from app.models.indexing import Chunk, IndexPublication, IndexState, IndexVersion
from app.services import indexing
from app.services.indexing import locked_state


def build(client, **payload):
    response = client.post("/api/indexes", json=payload)
    assert response.status_code == 201, response.text
    return response.json()


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


def chunks(client, version):
    return client.get(f"/api/indexes/{version['id']}/chunks?page_size=100").json()["items"]


def test_build_preserves_original_offsets_content_hash_and_provenance(client):
    content = ("Ａ字符\n那拉提🙂占位资料。" * 37).strip()
    item = add_item(client, content=content)
    first = build(client, chunk_size=100, overlap=20)
    second = build(client, chunk_size=100, overlap=20)
    assert first["status"] == "ready" and first["item_count"] == 1
    assert first["corpus_fingerprint"] == second["corpus_fingerprint"]
    assert first["config"] == second["config"] and first["id"] != second["id"]
    actual = chunks(client, first)
    assert len(actual) == first["chunk_count"]
    assert [c["ordinal"] for c in actual] == list(range(len(actual)))
    assert actual[0]["start_offset"] == 0 and actual[-1]["end_offset"] == len(content)
    restored = [""] * len(content)
    for chunk in actual:
        start, end = chunk["start_offset"], chunk["end_offset"]
        assert chunk["text"] == content[start:end]
        assert chunk["text_sha256"] == hashlib.sha256(chunk["text"].encode()).hexdigest()
        assert chunk["provenance"]["knowledge_version"] == item["version"]
        assert chunk["provenance"]["source_url"] == "https://example.com/"
        restored[start:end] = list(chunk["text"])
    assert "".join(restored) == content
    assert [(c["text"], c["start_offset"], c["end_offset"]) for c in actual] == [
        (c["text"], c["start_offset"], c["end_offset"]) for c in chunks(client, second)
    ]
    state = client.get("/api/indexes/active").json()
    assert state["generation"] == 0 and state["active_version"] is None and not state["is_current"]


def test_only_valid_reviewed_evidence_is_chunked(client):
    good = add_item(client)
    add_item(client, status="pending")
    add_item(client, status="rejected")
    removed = add_item(client)
    assert (
        client.delete(
            f"/api/knowledge/{removed['id']}?expected_version={removed['version']}"
        ).status_code
        == 204
    )
    version = build(client)
    assert version["item_count"] == 1
    assert {c["knowledge_id"] for c in chunks(client, version)} == {good["id"]}


def test_publication_is_atomic_optimistic_and_audited(client):
    add_item(client)
    first, second = build(client), build(client)
    result = publish(client, first)
    assert result.status_code == 200 and result.json()["generation"] == 1
    assert result.json()["is_current"]
    assert publish(client, second).status_code == 409
    assert publish(client, second, 1).json()["generation"] == 2
    assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "superseded"
    assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "published"
    # Same active version with the current generation is an idempotent no-op.
    assert publish(client, second, 2).json()["generation"] == 2
    history = client.get("/api/indexes/publications").json()
    assert history["total"] == 2
    assert history["items"][0]["previous_version_id"] == first["id"]
    assert history["items"][0]["version_id"] == second["id"]
    assert client.get("/api/indexes/active").json()["active_version"]["id"] == second["id"]


def test_catalog_mutation_marks_active_stale_and_rejects_old_build(client):
    item = add_item(client)
    version = build(client)
    assert publish(client, version).status_code == 200
    old = chunks(client, version)
    updated = client.patch(
        f"/api/knowledge/{item['id']}",
        json={"expected_version": item["version"], "content": "修订的占位正文"},
    )
    assert updated.status_code == 200
    active = client.get("/api/indexes/active").json()
    assert not active["is_current"] and active["generation"] == 1
    assert active["active_version"]["id"] == version["id"]
    assert chunks(client, version) == old
    assert publish(client, version, 1).status_code == 409
    replacement = build(client)
    assert replacement["item_count"] == 0  # revised body is pending, not approved evidence
    assert publish(client, replacement, 1).json()["is_current"]
    assert chunks(client, replacement) == []


def test_failed_catalog_write_does_not_invalidate_current_version(client):
    item = add_item(client)
    version = build(client)
    publish(client, version)
    before = client.get("/api/indexes/active").json()
    assert (
        client.patch(
            f"/api/knowledge/{item['id']}", json={"expected_version": 999, "content": "占位"}
        ).status_code
        == 409
    )
    assert client.post("/api/knowledge", json={"title": "", "content": "占位"}).status_code == 422
    assert client.get("/api/indexes/active").json() == before


def test_source_and_document_mutations_preserve_old_provenance(client):
    src = client.post(
        "/api/sources", json={"name": "原始占位来源", "source_url": "https://example.com/"}
    ).json()
    doc = client.post(
        "/api/documents",
        json={
            "source_id": src["id"],
            "title": "占位文档",
            "content": "占位正文",
            "collected_at": "2026-10-16T00:00:00Z",
        },
    ).json()
    add_item(client, document_id=doc["id"])
    version = build(client)
    publish(client, version)
    snapshot = chunks(client, version)
    assert snapshot[0]["provenance"]["document_version"] == doc["version"]
    assert snapshot[0]["provenance"]["source_name"] == src["name"]
    assert (
        client.patch(
            f"/api/documents/{doc['id']}", json={"expected_version": 1, "content": "文档新正文"}
        ).status_code
        == 200
    )
    assert not client.get("/api/indexes/active").json()["is_current"]
    empty = build(client)
    publish(client, empty, 1)
    assert (
        client.patch(
            f"/api/sources/{src['id']}", json={"expected_version": 1, "name": "新来源名"}
        ).status_code
        == 200
    )
    assert not client.get("/api/indexes/active").json()["is_current"]
    assert chunks(client, version) == snapshot
    assert publish(client, version, 2).status_code == 409


def test_fingerprint_recheck_rejects_out_of_band_catalog_changes(client, migrated_database):
    item = add_item(client)
    version = build(client)
    publish(client, version)
    with migrated_database.begin() as c:
        c.execute(
            text("UPDATE knowledge_items SET content=:body WHERE id=:id"),
            {"body": "直接修改的占位正文", "id": item["id"]},
        )
    assert not client.get("/api/indexes/active").json()["is_current"]
    assert publish(client, version, 1).status_code == 409


@pytest.mark.parametrize("failure", ["chunk_budget", "corpus_budget", "database_write"])
def test_failed_build_keeps_previous_publication_and_leaves_no_partial_rows(
    client, migrated_database, monkeypatch, failure
):
    add_item(client, content="占位正文" * 100)
    previous = build(client)
    publish(client, previous)
    if failure == "chunk_budget":
        monkeypatch.setattr(indexing, "MAX_CHUNKS", 1)
        response = client.post("/api/indexes", json={"chunk_size": 100, "overlap": 0})
        assert response.status_code == 503
    elif failure == "corpus_budget":
        from app.services import retrieval

        monkeypatch.setattr(retrieval, "MAX_ITEMS", 0)
        assert client.post("/api/indexes", json={}).status_code == 503
        monkeypatch.undo()
    else:
        # Real PostgreSQL constraint failure during chunk insert; the new version rolls back too.
        with migrated_database.begin() as c:
            c.execute(
                text("ALTER TABLE chunks ADD CONSTRAINT test_reject_chunk CHECK (false) NOT VALID")
            )
        assert client.post("/api/indexes", json={}).status_code == 409
    with Session(migrated_database) as db:
        assert db.scalar(select(func.count()).select_from(IndexVersion)) == 1
        assert db.scalar(select(func.count()).select_from(Chunk)) == previous["chunk_count"]
        assert db.get(IndexState, 1).active_version_id == previous["id"]
        assert db.scalar(select(func.count()).select_from(IndexPublication)) == 1
    assert client.get("/api/indexes/active").json()["active_version"]["id"] == previous["id"]


@pytest.mark.parametrize(
    "payload",
    [
        {"chunk_size": 99},
        {"chunk_size": 2001},
        {"overlap": -1},
        {"chunk_size": 100, "overlap": 51},
        {"unexpected": 1},
    ],
)
def test_build_validation_has_no_side_effects(client, payload):
    assert client.post("/api/indexes", json=payload).status_code == 422
    assert client.get("/api/indexes").json()["total"] == 0


def test_index_management_permissions_csrf_and_pagination(client, anonymous_client):
    # Same TestClient fixture is authenticated as admin here; first verify authorized validation.
    version = build(client)
    assert client.post("/api/indexes", json={}, headers={"X-CSRF-Token": ""}).status_code == 403
    assert client.get("/api/indexes?page_size=101").status_code == 422
    assert client.get(f"/api/indexes/{uuid4()}").status_code == 404
    assert client.get(f"/api/indexes/{uuid4()}/chunks").status_code == 404
    assert client.get("/api/indexes/invalid").status_code == 422
    assert client.get(f"/api/indexes/{version['id']}/chunks?page=2").json()["items"] == []
    client.cookies.clear()
    for method, path in [
        ("GET", "/api/indexes"),
        ("GET", "/api/indexes/active"),
        ("GET", "/api/indexes/publications"),
        ("POST", "/api/indexes"),
        ("GET", f"/api/indexes/{version['id']}/chunks"),
        ("POST", f"/api/indexes/{version['id']}/publish"),
    ]:
        assert client.request(method, path, json={}).status_code == 401
    assert (
        client.post(
            "/api/auth/register", json={"username": "normal_user", "password": "test-password-123"}
        ).status_code
        == 201
    )
    auth = client.post(
        "/api/auth/login", json={"username": "normal_user", "password": "test-password-123"}
    ).json()
    client.headers["X-CSRF-Token"] = auth["csrf_token"]
    assert client.post("/api/indexes", json={}).status_code == 403
    assert client.get("/api/indexes/active").status_code == 403


def test_publication_lock_serializes_competing_publishers(client, migrated_database):
    add_item(client)
    first, second = build(client), build(client)
    actor = first["created_by"]
    with Session(migrated_database) as owner, Session(migrated_database) as competitor:
        locked_state(owner)
        competitor.execute(text("SET LOCAL lock_timeout = '100ms'"))
        with pytest.raises(OperationalError) as blocked:
            indexing.publish_index(competitor, UUID(second["id"]), 0, actor)
        assert blocked.value.orig.sqlstate == "55P03"
        competitor.rollback()
        assert indexing.publish_index(owner, UUID(first["id"]), 0, actor).generation == 1
        with pytest.raises(HTTPException) as stale:
            indexing.publish_index(competitor, UUID(second["id"]), 0, actor)
        assert stale.value.status_code == 409
        competitor.rollback()
    assert client.get("/api/indexes/publications").json()["total"] == 1
    assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"]


def test_catalog_writes_use_the_publication_lock(client, migrated_database):
    # The guard must run before a catalog mutation, even for creation with no parent row.
    with Session(migrated_database) as owner:
        locked_state(owner)
        from app.db.session import get_db

        factory = client.app.dependency_overrides[get_db]

        def short_timeout():
            for db in factory():
                db.execute(text("SET LOCAL lock_timeout = '100ms'"))
                yield db

        client.app.dependency_overrides[get_db] = short_timeout
        response = client.post("/api/knowledge", json={"title": "占位", "content": "占位"})
        assert response.status_code == 503
        client.app.dependency_overrides[get_db] = factory
        owner.rollback()
    assert client.get("/api/knowledge").json() == []


def test_publication_failure_rolls_back_old_status_pointer_and_audit(client, migrated_database):
    add_item(client)
    first, second = build(client), build(client)
    publish(client, first)
    with migrated_database.begin() as c:
        c.execute(
            text(
                "ALTER TABLE index_publications ADD CONSTRAINT test_reject_publication CHECK (false) NOT VALID"
            )
        )
    assert publish(client, second, 1).status_code == 409
    assert client.get("/api/indexes/active").json()["active_version"]["id"] == first["id"]
    assert client.get(f"/api/indexes/{first['id']}").json()["status"] == "published"
    assert client.get(f"/api/indexes/{second['id']}").json()["status"] == "ready"
    assert client.get("/api/indexes/publications").json()["total"] == 1