import json
from pathlib import Path

import pytest
from django.contrib.auth.models import Group
from django.contrib.auth.models import User
from guardian.shortcuts import assign_perm
from pytest_mock import MockerFixture

from documents.models import CustomField
from documents.models import CustomFieldInstance
from documents.models import Document
from documents.models import Note
from documents.search._backend import SearchMode
from documents.search._backend import TantivyBackend
from documents.search._backend import WriteBatch
from documents.search._backend import get_backend
from documents.search._backend import reset_backend
from documents.signals.handlers import add_to_index
from documents.tests.factories import CorrespondentFactory
from documents.tests.factories import DocumentFactory
from documents.tests.factories import DocumentTypeFactory
from documents.tests.factories import TagFactory
from documents.tests.factories import UserFactory

pytestmark = [pytest.mark.search, pytest.mark.django_db]

# Extensions of actual Tantivy segment data files, as opposed to its own
# bookkeeping files (meta.json, .managed.json, lock files).
_SEGMENT_FILE_EXTENSIONS = (
    ".fast",
    ".fieldnorm",
    ".idx",
    ".pos",
    ".store",
    ".term",
)


class TestWriteBatch:
    """Test WriteBatch context manager functionality."""

    def test_rolls_back_on_exception(self, backend: TantivyBackend) -> None:
        """Batch operations must rollback on exception to preserve index integrity."""
        doc = Document.objects.create(
            title="Rollback Target",
            content="should survive",
            checksum="RB1",
            pk=1,
        )
        backend.add_or_update(doc)

        try:
            with backend.batch_update() as batch:
                batch.remove(doc.pk)
                raise RuntimeError("simulated failure")
        except RuntimeError:
            pass

        ids = backend.search_ids("should survive", user=None)
        assert len(ids) == 1

    def test_writer_released_when_commit_fails(
        self,
        backend: TantivyBackend,
        mocker: MockerFixture,
    ) -> None:
        """A commit failure must still dispose the writer (released in finally).

        Otherwise the Tantivy IndexWriter lingers holding its internal lock and
        the next batch fails with LockBusy. The real writer is created in
        __enter__; here commit() is forced to raise via a mocked _writer.
        """
        doc = Document.objects.create(
            title="Commit Fail",
            content="indexable text",
            checksum="WBCF1",
            pk=42,
        )

        failing = mocker.MagicMock()
        failing.commit.side_effect = RuntimeError("simulated commit failure")
        mocker.patch.object(
            WriteBatch,
            "_writer",
            new_callable=mocker.PropertyMock,
            return_value=failing,
        )

        batch = backend.batch_update()
        with pytest.raises(RuntimeError, match="simulated commit failure"):
            with batch as b:
                b.add_or_update(doc)

        # Writer disposed despite the commit failure.
        assert batch._raw_writer is None

        # Drop the patch so a real writer can be created; a fresh batch must
        # succeed (would raise LockBusy if the previous writer had leaked).
        mocker.stopall()
        backend.add_or_update(doc)
        assert len(backend.search_ids("indexable", user=None)) == 1


class TestSearch:
    """Test search query parsing and matching via search_ids."""

    def test_empty_index_returns_no_ids(self, backend: TantivyBackend) -> None:
        """Empty indexes must not pass a zero limit to Tantivy."""
        assert backend.search_ids("missing", user=None) == []

    def test_text_mode_limits_default_search_to_title_and_content(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text mode must not match metadata-only fields."""
        doc = Document.objects.create(
            title="Invoice document",
            content="monthly statement",
            checksum="TXT1",
            pk=9,
        )
        backend.add_or_update(doc)

        assert (
            len(
                backend.search_ids(
                    "document_type:invoice",
                    user=None,
                    search_mode=SearchMode.TEXT,
                ),
            )
            == 0
        )
        assert (
            len(backend.search_ids("monthly", user=None, search_mode=SearchMode.TEXT))
            == 1
        )

    def test_title_mode_limits_default_search_to_title_only(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Title mode must not match content-only terms."""
        doc = Document.objects.create(
            title="Invoice document",
            content="monthly statement",
            checksum="TXT2",
            pk=10,
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("monthly", user=None, search_mode=SearchMode.TITLE))
            == 0
        )
        assert (
            len(backend.search_ids("invoice", user=None, search_mode=SearchMode.TITLE))
            == 1
        )

    def test_text_mode_matches_partial_term_substrings(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text mode should support substring matching within tokens."""
        doc = Document.objects.create(
            title="Account access",
            content="password reset instructions",
            checksum="TXT3",
            pk=11,
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("pass", user=None, search_mode=SearchMode.TEXT)) == 1
        )
        assert (
            len(backend.search_ids("sswo", user=None, search_mode=SearchMode.TEXT)) == 1
        )
        for query in ["sswo re", "re sswo"]:
            assert (
                len(backend.search_ids(query, user=None, search_mode=SearchMode.TEXT))
                == 1
            ), query

    def test_text_mode_matches_all_terms_without_requiring_adjacency(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text mode should match all terms in any order or field."""
        doc = Document.objects.create(
            title="complete-medical-history",
            content="Samsung Odyssey curved monitor",
            checksum="TXT13",
            pk=19,
        )
        backend.add_or_update(doc)

        for query in [
            "complete history",
            "history complete",
            "Samsung curved",
            "curved Samsung",
        ]:
            assert backend.search_ids(
                query,
                user=None,
                search_mode=SearchMode.TEXT,
            ) == [doc.pk], query

    def test_text_mode_matches_terms_across_title_and_content(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Each simple-search term may match either title or content."""
        doc = Document.objects.create(
            title="Complete record",
            content="Patient history",
            checksum="TXT14",
            pk=20,
        )
        backend.add_or_update(doc)

        assert backend.search_ids(
            "complete history",
            user=None,
            search_mode=SearchMode.TEXT,
        ) == [doc.pk]

    def test_text_mode_does_not_match_on_partial_term_overlap(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text mode should not match documents that merely share partial fragments."""
        doc = Document.objects.create(
            title="Adobe Acrobat PDF Files",
            content="Lorem ipsum dolor sit amet, consectetur adipiscing elit.",
            checksum="TXT7",
            pk=13,
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("raptor", user=None, search_mode=SearchMode.TEXT))
            == 0
        )

    def test_text_mode_anchors_numeric_tokens_regardless_of_query_order(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Numeric tokens must not match in the middle of a larger number."""
        exact_doc = Document.objects.create(
            title="Z-Berichte 6",
            content="monthly report",
            checksum="TXT9",
            pk=15,
        )
        prefix_doc = Document.objects.create(
            title="Z-Berichte 60",
            content="monthly report",
            checksum="TXT10",
            pk=16,
        )
        false_positive = Document.objects.create(
            title="Z-Berichte 16",
            content="monthly report",
            checksum="TXT11",
            pk=17,
        )
        backend.add_or_update(exact_doc)
        backend.add_or_update(prefix_doc)
        backend.add_or_update(false_positive)

        for query in ["Z-Berichte 6", "6 Z-Berichte"]:
            result_ids = set(
                backend.search_ids(query, user=None, search_mode=SearchMode.TEXT),
            )

            assert exact_doc.id in result_ids, query
            assert prefix_doc.id in result_ids, query
            assert false_positive.id not in result_ids, query

    def test_text_mode_ignores_queries_without_searchable_tokens(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text mode should safely return no hits for symbol-only strings."""
        doc = Document.objects.create(
            title="Guide",
            content="This is a guide.",
            checksum="TXT8",
            pk=14,
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("!!!", user=None, search_mode=SearchMode.TEXT)) == 0
        )

    def test_title_mode_matches_partial_term_substrings(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Title mode should support substring matching within title tokens."""
        doc = Document.objects.create(
            title="Password guide",
            content="reset instructions",
            checksum="TXT4",
            pk=12,
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("pass", user=None, search_mode=SearchMode.TITLE))
            == 1
        )
        assert (
            len(backend.search_ids("sswo", user=None, search_mode=SearchMode.TITLE))
            == 1
        )
        assert (
            len(backend.search_ids("sswo gu", user=None, search_mode=SearchMode.TITLE))
            == 1
        )

    @pytest.mark.parametrize(
        ("search_mode", "query"),
        [
            pytest.param(SearchMode.TITLE, "12345", id="title_search"),
            pytest.param(SearchMode.TEXT, "12345", id="text_search"),
            pytest.param(SearchMode.QUERY, None, id="query_title_exact"),
        ],
    )
    def test_search_modes_match_model_limit_title_tokens(
        self,
        backend: TantivyBackend,
        search_mode: SearchMode,
        query: str | None,
    ) -> None:
        """Search must keep filename-like title tokens up to the model limit."""
        long_title = "1234567890" * 12 + "12345678"
        doc = Document.objects.create(
            title=long_title,
            content="ordinary content",
            checksum="TXT12",
            pk=18,
        )
        backend.add_or_update(doc)

        assert backend.search_ids(
            query or f"title:{long_title}",
            user=None,
            search_mode=search_mode,
        ) == [doc.pk]

    @pytest.mark.parametrize(
        ("mode", "title", "content", "hits", "misses"),
        [
            pytest.param(
                SearchMode.QUERY,
                "CJK document",
                "東京都の人口は約1400万人です",
                ["東京", "人口"],
                ["大阪"],
                id="query_mode_cjk_content",
            ),
            pytest.param(
                SearchMode.TEXT,
                "CJK document",
                "東京都の人口は約1400万人です",
                ["東京"],
                ["大阪"],
                id="text_mode_cjk_content",
            ),
            pytest.param(
                SearchMode.TITLE,
                "東京都の報告書",
                "This document is about Tokyo.",
                ["東京", "報告"],
                ["大阪"],
                id="title_mode_cjk_title",
            ),
        ],
    )
    def test_cjk_search_finds_matching_documents(
        self,
        backend: TantivyBackend,
        mode: SearchMode,
        title: str,
        content: str,
        hits: list[str],
        misses: list[str],
    ) -> None:
        """CJK queries must match documents via bigram fields in all three search modes."""
        doc = DocumentFactory(title=title, content=content)
        backend.add_or_update(doc)

        for query in hits:
            assert len(backend.search_ids(query, user=None, search_mode=mode)) == 1, (
                f"Expected {query!r} to match in {mode} mode"
            )
        for query in misses:
            assert len(backend.search_ids(query, user=None, search_mode=mode)) == 0, (
                f"Expected {query!r} not to match in {mode} mode"
            )

    def test_title_mode_cjk_does_not_match_content_only(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Title-only CJK search must not return docs where CJK appears only in content."""
        doc = DocumentFactory(
            title="Tokyo report",
            content="東京都の人口は約1400万人です",
        )
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("東京", user=None, search_mode=SearchMode.TITLE))
            == 0
        )

    @pytest.mark.parametrize(
        ("field", "query", "miss"),
        [
            pytest.param("correspondent", "東京", "大阪", id="cjk_correspondent"),
            pytest.param("document_type", "請求書", "領収書", id="cjk_document_type"),
            pytest.param("tag", "重要", "普通", id="cjk_tag"),
        ],
    )
    def test_cjk_metadata_search_via_query_mode(
        self,
        backend: TantivyBackend,
        field: str,
        query: str,
        miss: str,
    ) -> None:
        """CJK in correspondent/document_type/tag names must be searchable via global search."""
        if field == "correspondent":
            doc = DocumentFactory(correspondent=CorrespondentFactory(name=query))
        elif field == "document_type":
            doc = DocumentFactory(document_type=DocumentTypeFactory(name=query))
        else:
            tag = TagFactory(name=query)
            doc = DocumentFactory()
            doc.tags.add(tag)
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids(query, user=None, search_mode=SearchMode.QUERY)) == 1
        ), f"Expected CJK {field} name {query!r} to match"
        assert (
            len(backend.search_ids(miss, user=None, search_mode=SearchMode.QUERY)) == 0
        ), f"Expected {miss!r} not to match"

    def test_cjk_text_mode_does_not_leak_field_query_semantics(
        self,
        backend: TantivyBackend,
    ) -> None:
        """TEXT mode is plain-text over content: a 'field:CJK' input must not be
        parsed as a structured query against that field. A doc tagged 重要 with
        no 重要 in its content must NOT match the TEXT-mode query 'tag:重要'."""
        tag = TagFactory(name="重要")
        doc = DocumentFactory(title="report", content="just english content")
        doc.tags.add(tag)
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids("tag:重要", user=None, search_mode=SearchMode.TEXT))
            == 0
        )
        # Sanity: the CJK run still matches when it is actually in the content.
        doc2 = DocumentFactory(title="report2", content="本文に重要な情報")
        backend.add_or_update(doc2)
        assert (
            len(backend.search_ids("tag:重要", user=None, search_mode=SearchMode.TEXT))
            == 1
        )

    @pytest.mark.parametrize(
        "query",
        [
            pytest.param("Straße", id="eszett"),
            pytest.param("Ærøskøbing", id="ae_and_oslash"),
            pytest.param("strasse", id="ascii_fold_form"),
        ],
    )
    def test_simple_search_folds_special_letters_like_index(
        self,
        backend: TantivyBackend,
        query: str,
    ) -> None:
        """Query-side folding must match index-side folding for non-decomposable
        letters (ß→ss, ø→o, ...). Searching the accented form must find the doc.
        A naive NFD fold deletes these letters and silently fails to match."""
        doc = DocumentFactory(title="report", content="Straße Ærøskøbing")
        backend.add_or_update(doc)

        assert (
            len(backend.search_ids(query, user=None, search_mode=SearchMode.TEXT)) == 1
        )

    def test_sort_field_ascending(self, backend: TantivyBackend) -> None:
        """Searching with sort_reverse=False must return results in ascending ASN order."""
        for asn in [30, 10, 20]:
            doc = Document.objects.create(
                title="sortable",
                content="sortable content",
                checksum=f"SFA{asn}",
                archive_serial_number=asn,
            )
            backend.add_or_update(doc)

        ids = backend.search_ids(
            "sortable",
            user=None,
            sort_field="archive_serial_number",
            sort_reverse=False,
        )
        assert len(ids) == 3
        asns = [Document.objects.get(pk=doc_id).archive_serial_number for doc_id in ids]
        assert asns == [10, 20, 30]

    def test_sort_field_descending(self, backend: TantivyBackend) -> None:
        """Searching with sort_reverse=True must return results in descending ASN order."""
        for asn in [30, 10, 20]:
            doc = Document.objects.create(
                title="sortable",
                content="sortable content",
                checksum=f"SFD{asn}",
                archive_serial_number=asn,
            )
            backend.add_or_update(doc)

        ids = backend.search_ids(
            "sortable",
            user=None,
            sort_field="archive_serial_number",
            sort_reverse=True,
        )
        assert len(ids) == 3
        asns = [Document.objects.get(pk=doc_id).archive_serial_number for doc_id in ids]
        assert asns == [30, 20, 10]


class TestSearchIds:
    """Test lightweight ID-only search."""

    def test_returns_matching_ids(self, backend: TantivyBackend) -> None:
        """search_ids must return IDs of all matching documents."""
        docs = []
        for i in range(5):
            doc = Document.objects.create(
                title=f"findable doc {i}",
                content="common keyword",
                checksum=f"SI{i}",
            )
            backend.add_or_update(doc)
            docs.append(doc)
        other = Document.objects.create(
            title="unrelated",
            content="nothing here",
            checksum="SI_other",
        )
        backend.add_or_update(other)

        ids = backend.search_ids(
            "common keyword",
            user=None,
            search_mode=SearchMode.QUERY,
        )
        assert set(ids) == {d.pk for d in docs}
        assert other.pk not in ids

    def test_respects_permission_filter(self, backend: TantivyBackend) -> None:
        """search_ids must respect user permission filtering."""
        owner = User.objects.create_user("ids_owner")
        other = User.objects.create_user("ids_other")
        doc = Document.objects.create(
            title="private doc",
            content="secret keyword",
            checksum="SIP1",
            owner=owner,
        )
        backend.add_or_update(doc)

        assert backend.search_ids(
            "secret",
            user=owner,
            search_mode=SearchMode.QUERY,
        ) == [doc.pk]
        assert (
            backend.search_ids("secret", user=other, search_mode=SearchMode.QUERY) == []
        )

    def test_respects_fuzzy_threshold(self, backend: TantivyBackend, settings) -> None:
        """search_ids must apply the same fuzzy threshold as search()."""
        doc = Document.objects.create(
            title="threshold test",
            content="unique term",
            checksum="SIT1",
        )
        backend.add_or_update(doc)

        settings.ADVANCED_FUZZY_SEARCH_THRESHOLD = 1.1
        ids = backend.search_ids("unique", user=None, search_mode=SearchMode.QUERY)
        assert ids == []

    def test_returns_ids_for_text_mode(self, backend: TantivyBackend) -> None:
        """search_ids must work with TEXT search mode."""
        doc = Document.objects.create(
            title="text mode doc",
            content="findable phrase",
            checksum="SIM1",
        )
        backend.add_or_update(doc)

        ids = backend.search_ids("findable", user=None, search_mode=SearchMode.TEXT)
        assert ids == [doc.pk]


class TestRebuild:
    """Test index rebuilding functionality."""

    def test_with_iter_wrapper_called(self, backend: TantivyBackend) -> None:
        """Index rebuild must pass (document, viewer_ids) pairs through iter_wrapper."""
        seen = []

        def wrapper(pairs):
            for doc, viewer_ids in pairs:
                seen.append(doc.pk)
                yield doc, viewer_ids

        Document.objects.create(title="Tracked", content="x", checksum="TW1", pk=30)
        backend.rebuild(Document.objects.all(), iter_wrapper=wrapper)
        assert 30 in seen

    def test_includes_group_granted_viewers(self, backend: TantivyBackend) -> None:
        """Rebuild must index viewer ids for group-only grants, not just direct ones.

        The batched viewer-id lookup used during rebuild() must mirror
        get_users_with_perms(with_group_users=True), which is the default
        used by the non-batched per-document indexing path. Without it, a
        user who can only see a document via group membership would lose
        search access to it after any full reindex.
        """
        owner = UserFactory()
        group_member = UserFactory()
        group = Group.objects.create(name="viewers")
        group_member.groups.add(group)

        doc = DocumentFactory(
            title="Group shared doc",
            content="group secret keyword",
            owner=owner,
        )
        assign_perm("view_document", group, doc)

        backend.rebuild(Document.objects.all())

        ids = backend.search_ids(
            "group secret",
            user=group_member,
            search_mode=SearchMode.QUERY,
        )
        assert ids == [doc.pk]


class TestAutocomplete:
    """Test autocomplete functionality."""

    def test_basic_functionality(self, backend: TantivyBackend) -> None:
        """Autocomplete must return words matching the given prefix."""
        doc = Document.objects.create(
            title="Invoice from Microsoft Corporation",
            content="payment details",
            checksum="AC1",
            pk=40,
        )
        backend.add_or_update(doc)

        results = backend.autocomplete("micro", limit=10)
        assert "microsoft" in results

    def test_results_ordered_by_document_frequency(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Autocomplete results must be ordered by document frequency to prioritize common terms."""
        # "payment" appears in 3 docs; "payslip" in 1 — "pay" prefix should
        # return "payment" before "payslip".
        for i, (title, checksum) in enumerate(
            [
                ("payment invoice", "AF1"),
                ("payment receipt", "AF2"),
                ("payment confirmation", "AF3"),
                ("payslip march", "AF4"),
            ],
            start=41,
        ):
            doc = Document.objects.create(
                title=title,
                content="details",
                checksum=checksum,
                pk=i,
            )
            backend.add_or_update(doc)

        results = backend.autocomplete("pay", limit=10)
        assert results.index("payment") < results.index("payslip")

    def test_folds_special_letters_consistently(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Autocomplete words must fold the same way as content (ß→ss), so a
        prefix of the folded form finds them. A naive NFD fold would store the
        word as 'strae' and the prefix 'stras' would never match it."""
        doc = DocumentFactory(title="Straße", content="details")
        backend.add_or_update(doc)

        assert "strasse" in backend.autocomplete("stras", limit=10)


class TestMoreLikeThis:
    """Test more like this functionality."""

    def test_more_like_this_ids_excludes_original(
        self,
        backend: TantivyBackend,
    ) -> None:
        """more_like_this_ids must return IDs of similar documents, excluding the original."""
        doc1 = Document.objects.create(
            title="Important document",
            content="financial information report",
            checksum="MLTI1",
            pk=150,
        )
        doc2 = Document.objects.create(
            title="Another document",
            content="financial information report",
            checksum="MLTI2",
            pk=151,
        )
        backend.add_or_update(doc1)
        backend.add_or_update(doc2)

        ids = backend.more_like_this_ids(doc_id=150, user=None)
        assert 150 not in ids
        assert 151 in ids


class TestSingleton:
    """Test get_backend() and reset_backend() singleton lifecycle."""

    @pytest.fixture(autouse=True)
    def _clean(self):
        reset_backend()
        yield
        reset_backend()

    def test_returns_same_instance_on_repeated_calls(self, index_dir) -> None:
        """Singleton pattern: repeated calls to get_backend() must return the same instance."""
        assert get_backend() is get_backend()

    def test_reinitializes_when_index_dir_changes(self, tmp_path, settings) -> None:
        """Backend singleton must reinitialize when INDEX_DIR setting changes for test isolation."""
        settings.INDEX_DIR = tmp_path / "a"
        (tmp_path / "a").mkdir()
        b1 = get_backend()

        settings.INDEX_DIR = tmp_path / "b"
        (tmp_path / "b").mkdir()
        b2 = get_backend()

        assert b1 is not b2
        assert b2._path == tmp_path / "b"

    def test_reset_forces_new_instance(self, index_dir) -> None:
        """reset_backend() must force creation of a new backend instance on next get_backend() call."""
        b1 = get_backend()
        reset_backend()
        b2 = get_backend()
        assert b1 is not b2


class TestFieldHandling:
    """Test handling of various document fields."""

    def test_none_values_handled_correctly(self, backend: TantivyBackend) -> None:
        """Document fields with None values must not cause indexing errors."""
        doc = Document.objects.create(
            title="Test Doc",
            content="test content",
            checksum="NV1",
            pk=60,
            original_filename=None,
            page_count=None,
        )
        # Should not raise an exception
        backend.add_or_update(doc)

        assert len(backend.search_ids("test", user=None)) == 1

    def test_custom_fields_include_name_and_value(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Custom fields must be indexed with both field name and value for structured queries."""
        field = CustomField.objects.create(
            name="Invoice Number",
            data_type=CustomField.FieldDataType.STRING,
        )
        doc = Document.objects.create(
            title="Invoice",
            content="test",
            checksum="CF1",
            pk=70,
        )
        CustomFieldInstance.objects.create(
            document=doc,
            field=field,
            value_text="INV-2024-001",
        )

        backend.add_or_update(doc)

        assert len(backend.search_ids("invoice", user=None)) == 1

    def test_select_custom_field_indexes_label_not_id(
        self,
        backend: TantivyBackend,
    ) -> None:
        """SELECT custom fields must index the human-readable label, not the opaque option ID."""
        field = CustomField.objects.create(
            name="Category",
            data_type=CustomField.FieldDataType.SELECT,
            extra_data={
                "select_options": [
                    {"id": "opt_abc", "label": "Invoice"},
                    {"id": "opt_def", "label": "Receipt"},
                ],
            },
        )
        doc = Document.objects.create(
            title="Categorised doc",
            content="test",
            checksum="SEL1",
            pk=71,
        )
        CustomFieldInstance.objects.create(
            document=doc,
            field=field,
            value_select="opt_abc",
        )
        backend.add_or_update(doc)

        assert len(backend.search_ids("custom_fields.value:invoice", user=None)) == 1
        assert len(backend.search_ids("custom_fields.value:opt_abc", user=None)) == 0

    def test_none_custom_field_value_not_indexed(self, backend: TantivyBackend) -> None:
        """Custom field instances with no value set must not produce an index entry."""
        field = CustomField.objects.create(
            name="Optional",
            data_type=CustomField.FieldDataType.SELECT,
            extra_data={"select_options": [{"id": "opt_1", "label": "Yes"}]},
        )
        doc = Document.objects.create(
            title="Unset field doc",
            content="test",
            checksum="SEL2",
            pk=72,
        )
        CustomFieldInstance.objects.create(
            document=doc,
            field=field,
            value_select=None,
        )
        backend.add_or_update(doc)

        assert len(backend.search_ids("custom_fields.value:none", user=None)) == 0

    def test_notes_include_user_information(self, backend: TantivyBackend) -> None:
        """Notes must be indexed with user information when available for structured queries."""
        user = User.objects.create_user("notewriter")
        doc = Document.objects.create(
            title="Doc with notes",
            content="test",
            checksum="NT1",
            pk=80,
        )
        Note.objects.create(document=doc, note="Important note", user=user)

        backend.add_or_update(doc)

        ids = backend.search_ids("test", user=None)
        assert len(ids) == 1, (
            f"Expected 1, got {len(ids)}. Document content should be searchable."
        )

        ids = backend.search_ids("notes.note:important", user=None)
        assert len(ids) == 1, (
            f"Expected 1, got {len(ids)}. Note content should be searchable via notes.note: prefix."
        )

    def test_notes_without_user_are_indexed(self, backend: TantivyBackend) -> None:
        """Notes whose user was deleted (SET_NULL) must not break indexing."""
        doc = Document.objects.create(
            title="Doc with orphaned note",
            content="test",
            checksum="NT2",
            pk=81,
        )
        Note.objects.create(document=doc, note="Orphaned note", user=None)

        backend.add_or_update(doc)

        ids = backend.search_ids("notes.note:orphaned", user=None)
        assert len(ids) == 1, (
            f"Expected 1, got {len(ids)}. Notes without a user should still be indexed."
        )


class TestHighlightHits:
    """Test highlight_hits returns proper HTML strings, not raw Snippet objects."""

    def test_highlights_simple_text_mode_returns_html_string(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Simple text search should still produce content highlights for exact-token hits."""
        doc = Document.objects.create(
            title="Highlight Test",
            content="The quick brown fox jumps over the lazy dog",
            checksum="HH0",
            pk=89,
        )
        backend.add_or_update(doc)

        hits = backend.highlight_hits("quick", [doc.pk], search_mode=SearchMode.TEXT)

        assert len(hits) == 1
        highlights = hits[0]["highlights"]
        assert "content" in highlights
        assert "<b>" in highlights["content"]

    def test_highlights_content_returns_html_string(
        self,
        backend: TantivyBackend,
    ) -> None:
        """highlight_hits must return HTML strings (from Snippet.to_html()), not Snippet objects."""
        doc = Document.objects.create(
            title="Highlight Test",
            content="The quick brown fox jumps over the lazy dog",
            checksum="HH1",
            pk=90,
        )
        backend.add_or_update(doc)

        hits = backend.highlight_hits("quick", [doc.pk])

        assert len(hits) == 1
        highlights = hits[0]["highlights"]
        assert "content" in highlights
        content_highlight = highlights["content"]
        assert isinstance(content_highlight, str), (
            f"Expected str, got {type(content_highlight)}: {content_highlight!r}"
        )
        # Tantivy wraps matched terms in <b> tags
        assert "<b>" in content_highlight, (
            f"Expected HTML with <b> tags, got: {content_highlight!r}"
        )

    def test_highlights_notes_returns_html_string(
        self,
        backend: TantivyBackend,
    ) -> None:
        """Note highlights must be HTML strings via notes_text companion field.

        The notes JSON field does not support tantivy SnippetGenerator; the
        notes_text plain-text field is used instead.  We use the full-text
        query "urgent" (not notes.note:) because notes_text IS in
        DEFAULT_SEARCH_FIELDS via the normal search path… actually, we use
        notes.note: prefix so the query targets notes content directly, but
        the snippet is generated from notes_text which stores the same text.
        """
        user = User.objects.create_user("hl_noteuser")
        doc = Document.objects.create(
            title="Doc with matching note",
            content="unrelated content",
            checksum="HH2",
            pk=91,
        )
        Note.objects.create(document=doc, note="urgent payment required", user=user)
        backend.add_or_update(doc)

        # Use notes.note: prefix so the document matches the query and the
        # notes_text snippet generator can produce highlights.
        hits = backend.highlight_hits("notes.note:urgent", [doc.pk])

        assert len(hits) == 1
        highlights = hits[0]["highlights"]
        assert "notes" in highlights
        note_highlight = highlights["notes"]
        assert isinstance(note_highlight, str), (
            f"Expected str, got {type(note_highlight)}: {note_highlight!r}"
        )
        assert "<b>" in note_highlight, (
            f"Expected HTML with <b> tags, got: {note_highlight!r}"
        )

    def test_empty_doc_list_returns_empty_hits(self, backend: TantivyBackend) -> None:
        """highlight_hits with no doc IDs must return an empty list."""
        hits = backend.highlight_hits("anything", [])
        assert hits == []

    def test_no_highlights_when_no_match(self, backend: TantivyBackend) -> None:
        """Documents not matching the query should not appear in results."""
        doc = Document.objects.create(
            title="Unrelated",
            content="completely different text",
            checksum="HH3",
            pk=92,
        )
        backend.add_or_update(doc)

        hits = backend.highlight_hits("quick", [doc.pk])

        assert len(hits) == 0


class TestVersionIndexing:
    """
    GIVEN:
        - A root document whose new version has just been consumed, e.g. by
          the password removal workflow action
    WHEN:
        - The consumption finished signal is handled
    THEN:
        - The root document is indexed with the new version's content, since
          versions are not searchable on their own
    """

    def test_consumed_version_updates_root_entry(
        self,
        backend: TantivyBackend,
        mocker: MockerFixture,
    ) -> None:
        root = Document.objects.create(
            title="Statement",
            content="",
            checksum="VER1",
            pk=90,
        )
        backend.add_or_update(root)
        version = Document.objects.create(
            title="Statement",
            content="unprotected statement text",
            checksum="VER2",
            pk=91,
            root_document=root,
            version_index=1,
        )
        mocker.patch("documents.search.get_backend", return_value=backend)

        add_to_index(sender=None, document=version)

        assert backend.search_ids("unprotected", user=None) == [root.pk]


class TestEffectiveContentIndexing:
    """
    GIVEN:
        - A root document with a newer version
    WHEN:
        - The root document is indexed
    THEN:
        - The newest version's content is indexed, never the root's own
          outdated text
    """

    def test_root_is_indexed_with_latest_version_content(
        self,
        backend: TantivyBackend,
    ) -> None:
        root = Document.objects.create(
            title="Statement",
            content="stale original text",
            checksum="EFF1",
            pk=95,
        )
        Document.objects.create(
            title="Statement",
            content="latest version text",
            checksum="EFF2",
            pk=96,
            root_document=root,
            version_index=1,
        )

        backend.add_or_update(root)

        assert backend.search_ids("latest", user=None) == [root.pk]
        assert backend.search_ids("stale", user=None) == []


class TestIndexDirectoryGarbageCollection:
    """Regression tests for Tantivy segment files leaking on disk when
    multiple long-lived worker processes (Granian/Celery) take turns writing
    to the same on-disk index (issue #13679)."""

    def test_no_permanently_orphaned_segment_files_across_worker_processes(
        self,
        tmp_path: Path,
    ) -> None:
        """Simulate two long-lived worker processes, each with its own
        process-local ``TantivyBackend``/``Index`` opened once at process
        start, alternating turns as the writer -- exactly how paperless runs
        in production (several Granian + Celery worker processes).

        Every segment file physically present on disk must still be tracked
        in Tantivy's ``.managed.json`` bookkeeping; otherwise it can never be
        garbage collected by anyone again and the index directory grows
        without bound.
        """
        index_dir = tmp_path / "index"
        index_dir.mkdir()

        worker_a = TantivyBackend(path=index_dir)
        worker_a.open()
        worker_b = TantivyBackend(path=index_dir)
        worker_b.open()
        workers = [worker_a, worker_b]

        docs = [
            DocumentFactory.create(checksum=f"GC{i}", title=f"gc doc {i}")
            for i in range(5)
        ]

        try:
            # Alternate writers across many commits, repeatedly upserting the
            # same documents so segments accumulate and get superseded,
            # forcing the delete+add upsert pattern and eventual merges.
            for i in range(30):
                worker = workers[i % len(workers)]
                doc = docs[i % len(docs)]
                worker.add_or_update(doc)
        finally:
            worker_a.close()
            worker_b.close()

        managed_path = index_dir / ".managed.json"
        managed = set(json.loads(managed_path.read_text()))
        on_disk = {
            p.name
            for p in index_dir.iterdir()
            if p.is_file() and p.suffix in _SEGMENT_FILE_EXTENSIONS
        }
        orphans = on_disk - managed

        assert not orphans, (
            "Segment files present on disk but absent from Tantivy's "
            f".managed.json bookkeeping (permanently un-collectible): {orphans}"
        )
