"""Integration: phase 106 (task 04) — the importer's date semantics on
real Postgres.

The backfill-correction case (D4, the owner's "on sync, update the
date"): a row first imported with a "today" mtime (the shape every
pre-phase-106 deployment has after the migration's
``server_default=now()`` backfill) gets its REAL — older — date on the
next sync even though the content did not change; the refresh is a
date-only ``unchanged``, so the ``sources_meta`` generation the sync
paths gate their bump on (``added + updated + pruned > 0``) stays put
(seeded before, read after). The D1 manual lock and the prune
interaction are pinned across the real DB boundary too.

Deterministic in-process :class:`~tests.fakes.FakeEmbedder` — no
network, no live model (the ``test_importer_e2e.py`` pattern).
"""
from __future__ import annotations

import asyncio
import os
from collections.abc import Iterator
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest
from sqlalchemy import select, text
from sqlalchemy.orm import Session

from app.models import Document
from app.rag.importer import import_sources
from app.rag.sources_meta import current_sources_version
from tests.fakes import FakeEmbedder

#: mtime granularity tolerance (os.utime + stat round-trip).
_TOL = timedelta(milliseconds=50)


@pytest.fixture(autouse=True)
def _clean_kb(db: Session) -> Iterator[None]:
    """Global KB state — truncated around every test (the
    ``test_importer_e2e.py`` shape)."""
    db.execute(text("TRUNCATE chunks, documents, query_log"))
    db.commit()
    yield
    db.execute(text("TRUNCATE chunks, documents, query_log"))
    db.commit()


@pytest.fixture(autouse=True)
def _pin_sources_version(db: Session) -> Iterator[None]:
    """The single-row ``sources_meta`` generation is global mutable
    state — pin it to a known, non-zero value around every test so the
    no-bump assertion proves the gate, not the seed."""
    db.execute(text("UPDATE sources_meta SET version = 7 WHERE id = 1"))
    db.commit()
    yield
    db.execute(text("UPDATE sources_meta SET version = 0 WHERE id = 1"))
    db.commit()


def _doc(db: Session, source: str, rel: str) -> Document:
    doc = db.scalar(select(Document).where(Document.source == source, Document.path == rel))
    assert doc is not None, f"no documents row for ({source!r}, {rel!r})"
    return doc


def test_unchanged_reimport_refreshes_backfilled_date_without_version_bump(
    db: Session, tmp_path: Path
) -> None:
    """The backfill-correction case (D4), end to end on real Postgres.

    Run 1: the file's mtime is "now" (the migration backfill shape) →
    the row stores a today-date. The file is then ``os.utime``'d back
    to 2019 with IDENTICAL content. Run 2: the row stores the 2019
    date — ``added/updated/pruned`` all 0 (it is still ``unchanged``)
    and ``dates_updated == 1``. Because the content counts did not
    move, the ``sources_meta`` generation the sync paths bump on a
    KB change stays exactly where it was seeded (7).
    """
    root = tmp_path / "Backfill"
    root.mkdir()
    file = root / "note.md"
    file.write_text("# Note\n\nthe content never changes\n", encoding="utf-8")
    assert current_sources_version(db) == 7  # the seeded generation

    llm = FakeEmbedder()
    first = asyncio.run(import_sources([root], llm, session=db))
    assert (first.added, first.unchanged, first.dates_updated) == (1, 0, 0)
    # The backfill shape: the stored date is the "today" mtime.
    stored = _doc(db, root.name, "note.md").created_at
    assert abs(stored - datetime.now(UTC)) < timedelta(seconds=60)

    real = datetime(2019, 3, 4, 8, 0, 0, tzinfo=UTC)
    os.utime(file, (real.timestamp(), real.timestamp()))  # content identical

    second = asyncio.run(import_sources([root], llm, session=db))
    # A date-only refresh: unchanged for every content gate.
    assert (second.added, second.updated, second.pruned, second.unchanged) == (0, 0, 0, 1)
    assert second.dates_updated == 1
    db.expire_all()
    doc = _doc(db, root.name, "note.md")
    assert abs(doc.created_at - real) <= _TOL  # the real (OLDER) date stored
    assert doc.created_at_manual is False
    # The date-only refresh left the generation untouched — the
    # ``added + updated + pruned > 0`` gate the sync paths use never
    # fired (D4: no sources_meta bump, no regeneration).
    assert current_sources_version(db) == 7


def test_manual_date_survives_unchanged_reimport_on_real_db(
    db: Session, tmp_path: Path
) -> None:
    """D1 across the DB boundary: the owner's correction
    (``created_at_manual`` — task 05's API writes it) survives an
    unchanged re-sync whose source date moved; the generation stays
    put too (no write happened at all on that row)."""
    root = tmp_path / "ManualKeep"
    root.mkdir()
    file = root / "note.md"
    file.write_text("# Note\n\ncorrected by the owner\n", encoding="utf-8")
    llm = FakeEmbedder()
    first = asyncio.run(import_sources([root], llm, session=db))
    assert first.added == 1
    correction = datetime(2024, 11, 30, 15, 45, 0, tzinfo=UTC)
    doc = _doc(db, root.name, "note.md")
    doc.created_at = correction
    doc.created_at_manual = True
    db.commit()
    ts = datetime(2018, 1, 1, 0, 0, 0, tzinfo=UTC).timestamp()
    os.utime(file, (ts, ts))

    second = asyncio.run(import_sources([root], llm, session=db))
    assert (second.added, second.updated, second.unchanged) == (0, 0, 1)
    assert second.dates_updated == 0  # the correction was NOT refreshed
    db.expire_all()
    doc = _doc(db, root.name, "note.md")
    assert doc.created_at == correction  # byte-identical (no rewrite)
    assert doc.created_at_manual is True
    assert current_sources_version(db) == 7


def test_date_only_refresh_coexists_with_prune_on_real_db(
    db: Session, tmp_path: Path
) -> None:
    """The matrix in one ``prune=True`` run (the sync button's shape):
    a manual row survives untouched, a non-manual unchanged row gets
    its date refreshed (counted in ``dates_updated`` only), and a
    deleted file is still pruned — the content gates and the date
    refresh compose without interfering."""
    root = tmp_path / "Matrix"
    root.mkdir()
    kept_manual = root / "manual.md"
    kept_manual.write_text("# Manual\n\nowner-corrected\n", encoding="utf-8")
    kept_plain = root / "plain.md"
    kept_plain.write_text("# Plain\n\nrefreshes\n", encoding="utf-8")
    gone = root / "gone.md"
    gone.write_text("# Gone\n\ndeleted upstream\n", encoding="utf-8")
    llm = FakeEmbedder()
    first = asyncio.run(import_sources([root], llm, session=db, prune=True))
    assert first.added == 3

    correction = datetime(2022, 7, 1, 10, 0, 0, tzinfo=UTC)
    doc = _doc(db, root.name, "manual.md")
    doc.created_at = correction
    doc.created_at_manual = True
    db.commit()
    moved = datetime(2017, 9, 9, 9, 9, 9, tzinfo=UTC)
    os.utime(kept_plain, (moved.timestamp(), moved.timestamp()))
    gone.unlink()  # deleted upstream

    second = asyncio.run(import_sources([root], llm, session=db, prune=True))
    # The refresh counts ONLY in dates_updated; the prune is a content
    # count (so this run DOES advance the generation — the gate is on
    # pruned, not on dates_updated).
    assert (second.added, second.updated, second.unchanged, second.pruned) == (0, 0, 2, 1)
    assert second.dates_updated == 1
    db.expire_all()
    manual = _doc(db, root.name, "manual.md")
    assert manual.created_at == correction and manual.created_at_manual is True
    plain = _doc(db, root.name, "plain.md")
    assert abs(plain.created_at - moved) <= _TOL and plain.created_at_manual is False
    # The pruned row is gone (the content gate did its job alongside the
    # date refresh).
    assert (
        db.scalar(select(Document).where(Document.source == root.name, Document.path == "gone.md"))
        is None
    )
    # The version the sync paths gate on was seeded, not advanced — this
    # suite only runs ``import_sources`` (the bump lives in the entry
    # points, which this run's ``pruned=1`` would trigger).
    assert current_sources_version(db) == 7