"""Raw chunks: building them from dropped messages, storing them once, finding them."""

from datetime import datetime, timezone

from navi.llm.base import Message
from navi.memory._chunks import _lexical_terms, _matched_terms, build_chunks
from tests.conftest_factory import FakeConnection, make_store_with_pool


def _msg(role: str, content: str, seq: int = 0, **kwargs) -> Message:
    return Message(role=role, content=content, sequence_number=seq, **kwargs)


class TestBuildChunks:
    def test_marks_summary_and_compression_rows_are_not_indexed(self):
        messages = [
            _msg("user", "hello", 1),
            _msg("user", "old summary", 2, is_summary=True),
            _msg("system", "compression marker", 3, is_compression=True),
            _msg("system", "plain system", 4),
            _msg("assistant", "hi", 5),
        ]
        chunks = build_chunks(messages)
        assert len(chunks) == 1
        assert "old summary" not in chunks[0]["content"]
        assert "compression marker" not in chunks[0]["content"]
        assert "hello" in chunks[0]["content"]
        assert "hi" in chunks[0]["content"]

    def test_turns_are_packed_until_the_size_cap(self):
        messages = [_msg("user", "x" * 60, index) for index in range(10)]
        chunks = build_chunks(messages, max_chars=150)
        assert len(chunks) == 5          # two messages per chunk
        assert all(len(c["content"]) <= 150 for c in chunks)
        assert [c["start_seq"] for c in chunks] == [0, 2, 4, 6, 8]

    def test_the_turn_range_tracks_the_packed_messages(self):
        messages = [_msg("user", "a", 7), _msg("assistant", "b", 8), _msg("tool", "c", 9, name="grep")]
        chunk = build_chunks(messages)[0]
        assert chunk["start_seq"] == 7
        assert chunk["end_seq"] == 9
        assert "Tool result (grep): c" in chunk["content"]

    def test_a_huge_message_is_elided_in_the_middle(self):
        body = "HEAD " + "x" * 20_000 + " TAIL"
        chunk = build_chunks([_msg("tool", body, 1)])[0]
        assert len(chunk["content"]) < 13_000
        assert chunk["content"].startswith("Tool result: HEAD")
        assert chunk["content"].endswith("TAIL")
        assert "middle elided" in chunk["content"]

    def test_timestamps_come_from_the_messages(self):
        early = datetime(2026, 1, 1, tzinfo=timezone.utc)
        late = datetime(2026, 1, 2, tzinfo=timezone.utc)
        chunk = build_chunks([
            _msg("user", "a", 1, created_at=early),
            _msg("assistant", "b", 2, created_at=late),
        ])[0]
        assert chunk["started_at"] == early
        assert chunk["ended_at"] == late

    def test_the_hash_is_stable_and_content_addressed(self):
        first = build_chunks([_msg("user", "same", 1)])[0]
        second = build_chunks([_msg("user", "same", 99)])[0]
        assert first["content_hash"] == second["content_hash"]

    def test_an_assistant_tool_request_without_text_is_recorded(self):
        from navi.llm.base import ToolCallRequest

        msg = Message(
            role="assistant",
            content="",
            sequence_number=1,
            tool_calls=[ToolCallRequest(id="1", name="grep", arguments={})],
        )
        chunks = build_chunks([msg])
        assert chunks and "grep" in chunks[0]["content"]


class TestAddChunks:
    async def test_chunks_are_inserted_with_scope_and_breadcrumbs(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        store._pgvector_checked = True
        store._pgvector_available = True
        conn.enqueue("INSERT 0 1")
        conn.enqueue("INSERT 0 1")
        chunks = build_chunks(
            [_msg("user", "first", 1), _msg("user", "second", 2)], max_chars=10
        )
        for chunk in chunks:
            chunk["breadcrumbs"] = ["port 8099"]
        written = await store.add_chunks(chunks, user_id="u1", session_id="s1")
        assert written == 2
        assert all("memory_chunks" in call[1] for call in conn.calls)
        assert "ON CONFLICT (session_id, content_hash) DO NOTHING" in conn.calls[0][1]
        assert conn.calls[0][2][1] == "u1"
        assert conn.calls[0][2][2] == "s1"
        assert conn.calls[0][2][8] == ["port 8099"]

    async def test_without_pgvector_the_write_is_text_only(self):
        # `::vector` would fail on a plain Postgres; the row must not name the column.
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        store._pgvector_checked = True
        store._pgvector_available = False
        conn.enqueue("INSERT 0 1")
        written = await store.add_chunks(
            build_chunks([_msg("user", "no vectors here", 1)]), session_id="s1"
        )
        assert written == 1
        assert "embedding" not in conn.calls[0][1]

    async def test_a_conflict_is_not_counted_as_written(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        store._pgvector_checked = True
        store._pgvector_available = True
        conn.enqueue("INSERT 0 0")
        written = await store.add_chunks(
            build_chunks([_msg("user", "again", 1)]), session_id="s1"
        )
        assert written == 0

    async def test_no_session_means_nothing_is_written(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        assert await store.add_chunks(build_chunks([_msg("user", "x", 1)])) == 0
        assert conn.calls == []


class TestLexicalTerms:
    """The chunk query is scored, so its terms must be content words only."""

    def test_function_words_are_not_terms(self):
        assert _lexical_terms("the port was on the other host") == ["port", "other", "host"]

    def test_matching_is_word_boundary_not_substring(self):
        # "the" inside "other" used to count as a matched query term.
        assert _matched_terms("other things", {"the"}) == 0
        assert _matched_terms("the port", {"the", "port"}) == 2

    def test_terms_are_capped_and_deduplicated(self):
        query = "port " * 5 + " ".join(f"word{i}" for i in range(30))
        terms = _lexical_terms(query)
        assert len(terms) == 12
        assert terms[0] == "port"
        assert terms.count("port") == 1


class TestSearchChunks:
    async def test_lexical_fallback_ranks_by_matched_terms(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        conn.enqueue([
            {
                "rank": 0.5, "id": "a", "session_id": "s1", "start_seq": 1, "end_seq": 2,
                "started_at": None, "ended_at": datetime(2026, 1, 1),
                "content": "the port is 8099 and the host is localhost",
                "breadcrumbs": ["port"], "created_at": datetime(2026, 1, 1),
            },
            {
                "rank": 0.1, "id": "b", "session_id": "s1", "start_seq": 3, "end_seq": 4,
                "started_at": None, "ended_at": datetime(2026, 1, 1),
                "content": "some port talk without the value",
                "breadcrumbs": [], "created_at": datetime(2026, 1, 1),
            },
        ])
        hits = await store.search_chunks("port host localhost", user_id="u1", limit=5)
        assert [h["id"] for h in hits] == ["a", "b"]
        assert hits[0]["terms_hit"] == 3
        assert hits[1]["terms_hit"] == 1
        assert hits[0]["score"] is None

    async def test_the_scope_filter_is_applied(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        conn.enqueue([])
        await store.search_chunks("q", user_id=None, session_id="s1", limit=3)
        query, args = conn.calls[0][1], conn.calls[0][2]
        assert "user_id IS NULL" in query
        assert "session_id = $1" in query
        assert args[0] == "s1"

    async def test_an_empty_term_list_falls_back_to_recent(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        conn.enqueue([
            {
                "id": "x", "session_id": "s1", "start_seq": 1, "end_seq": 1,
                "started_at": None, "ended_at": None, "content": "body",
                "breadcrumbs": [], "created_at": None,
            },
        ])
        hits = await store.search_chunks("!!!", user_id="u1", limit=2)
        assert hits[0]["id"] == "x"
        # "Most recent" is not "most similar": a fabricated 1.0 used to pass
        # recall's similarity threshold and inject an unrelated turn.
        assert hits[0]["score"] is None

    async def test_a_short_vector_result_is_topped_up_lexically(self):
        # A chunk with a NULL embedding is invisible to `ORDER BY embedding <=>`
        # — it must still be reachable through the lexical path.
        class Backend:
            async def embed(self, texts, model=None, route=None):
                return [[1.0, 0.0]]

        conn = FakeConnection()
        store = make_store_with_pool(conn)
        store._pgvector_checked = True
        store._pgvector_available = True
        store._embedding_backend = Backend()
        conn.enqueue([{
            "distance": 0.1, "id": "vec", "session_id": "s1", "start_seq": 1,
            "end_seq": 2, "started_at": None, "ended_at": None,
            "content": "vector match", "breadcrumbs": [], "created_at": None,
        }])
        conn.enqueue([{
            "rank": 0.2, "id": "lex", "session_id": "s1", "start_seq": 3,
            "end_seq": 4, "started_at": None, "ended_at": None,
            "content": "lexical match port", "breadcrumbs": [], "created_at": None,
        }])

        hits = await store.search_chunks("port", user_id="u1", limit=2)
        assert [h["id"] for h in hits] == ["vec", "lex"]

    async def test_a_full_vector_result_skips_the_lexical_query(self):
        class Backend:
            async def embed(self, texts, model=None, route=None):
                return [[1.0, 0.0]]

        conn = FakeConnection()
        store = make_store_with_pool(conn)
        store._pgvector_checked = True
        store._pgvector_available = True
        store._embedding_backend = Backend()
        conn.enqueue([{
            "distance": 0.1, "id": "vec", "session_id": "s1", "start_seq": 1,
            "end_seq": 2, "started_at": None, "ended_at": None,
            "content": "vector match", "breadcrumbs": [], "created_at": None,
        }])

        hits = await store.search_chunks("port", user_id="u1", limit=1)
        assert [h["id"] for h in hits] == ["vec"]
        assert len(conn.calls) == 1


class TestLatestSummaryAt:
    async def test_returns_the_timestamp(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        when = datetime(2026, 2, 3, tzinfo=timezone.utc)
        conn.enqueue(when)
        assert await store.latest_summary_at("s1") == when
        assert "is_summary = TRUE" in conn.calls[0][1]

    async def test_a_failure_is_swallowed(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        conn.enqueue(RuntimeError("boom"))
        assert await store.latest_summary_at("s1") is None

    async def test_the_archive_is_consulted_too(self):
        # Long sessions archive old rows, summaries included; the hot table then
        # has no summary at all and `before_summary` would be lost.
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        hot = datetime(2026, 2, 1, tzinfo=timezone.utc)
        archived = datetime(2026, 2, 3, tzinfo=timezone.utc)
        conn.enqueue(hot)
        conn.enqueue(archived)
        assert await store.latest_summary_at("s1") == archived
        assert "session_messages_archive" in conn.calls[1][1]

    async def test_a_missing_archive_table_keeps_the_hot_value(self):
        conn = FakeConnection()
        store = make_store_with_pool(conn)
        hot = datetime(2026, 2, 1, tzinfo=timezone.utc)
        conn.enqueue(hot)
        conn.enqueue(RuntimeError("no such table"))
        assert await store.latest_summary_at("s1") == hot