"""Reprise d'un import tué à mi-course (spec Imports §4.4, §11 `test_import_resume.py`).

Le scénario est celui du `kill -9` : un worker meurt **entre l'écriture d'un lot et sa
publication**. Il est simulé au seul endroit où il compte, `save_progress` — l'instruction
qui commit le lot *et* son curseur. Une `BaseException` la traverse : le worker n'attrape
que `Exception`, donc rien n'est finalisé, exactement comme un process tué par le noyau. Le
lot en vol part avec la session fermée.

Ce que le test prouve, et qui ne va pas de soi :

1. **Aucun doublon** — le lot perdu est rejoué en entier, et l'état final de la base est
   celui d'un run ininterrompu. C'est la propriété que la spec exige (§4.4).
2. **Le rapport est correct après reprise** — il est *identique*, clé par clé, à celui du
   run ininterrompu. La spec §11 se contentait de compteurs « non comparés » ; ils le sont
   ici, parce que les compteurs de lecture sont recalculés et que seuls ceux d'écriture
   voyagent dans `progress` (voir l'en-tête de `pipeline.py`). C'est le minor différé de T0,
   soldé.

Les deux runs tournent sur **deux organisations** : comparer le même réservoir deux fois ne
prouverait rien (le rejeu y est idempotent par construction). La comparaison porte donc sur
le contenu canonique, pas sur les identifiants, qui diffèrent nécessairement.
"""

import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import pytest
import pytest_asyncio
from sqlalchemy import delete, select, text, update
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.db.engine import get_engine
from app.jobs import service
from app.jobs.models import Job
from app.jobs.worker import Worker
from app.keywords import pipeline
from app.keywords.models import Import, KeywordList
from app.sites.models import Site, SiteTarget

MAGIC_REAL = Path(__file__).parent / "fixtures" / "real" / "semrush_magic_us.csv"
FETCHED_AT = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
# Cinq lots sur les 500 lignes de l'export réel : assez pour tuer le run au milieu.
BATCH = 100
KILL_AT_BATCH = 3


class HardKill(BaseException):
    """Simule un `kill -9` : le worker n'attrape que `Exception`, rien n'est finalisé."""


@pytest_asyncio.fixture(autouse=True)
async def _small_batches(db: AsyncSession, monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
    monkeypatch.setattr(pipeline, "BATCH_SIZE", BATCH)
    await _purge(db)
    yield
    await _purge(db)


async def _purge(db: AsyncSession) -> None:
    leftovers = select(Job.id).where(Job.type == pipeline.JOB_TYPE)
    await db.execute(delete(Import).where(Import.job_id.in_(leftovers)))
    await db.execute(delete(Job).where(Job.type == pipeline.JOB_TYPE))
    await db.commit()


async def _fresh_org(db: AsyncSession) -> tuple[str, uuid.UUID]:
    org_id = f"org_{uuid.uuid4().hex}"
    site = Site(org_id=org_id, domain=f"{uuid.uuid4().hex}.example", name="Reprise")
    db.add(site)
    await db.flush()
    target = SiteTarget(site_id=site.id, org_id=org_id, country="FR", language="fr")
    db.add(target)
    await db.commit()
    return org_id, target.id


async def _stage(
    db: AsyncSession,
    org_id: str,
    target_id: uuid.UUID,
    import_dir: Path,
    *,
    source: str,
    content: bytes,
) -> uuid.UUID:
    """Rend l'**identifiant** du job, pas l'instance : la session est expirée plus loin."""
    sha = uuid.uuid4().hex * 2
    _place(import_dir / org_id / f"{sha}.csv", content)
    keyword_list = KeywordList(org_id=org_id, site_target_id=target_id, name="Liste", source=source)
    db.add(keyword_list)
    await db.commit()
    import_id = uuid.uuid4()
    job = await service.enqueue(
        db,
        org_id,
        pipeline.JOB_TYPE,
        payload={
            "import_id": str(import_id),
            "file_sha256": sha,
            "source": source,
            "site_target_id": str(target_id),
            "locale": "fr-FR",
            "fetched_at": FETCHED_AT.isoformat(),
            "keyword_list_id": str(keyword_list.id),
        },
    )
    db.add(
        Import(
            id=import_id,
            org_id=org_id,
            site_target_id=target_id,
            source=source,
            kind="csv",
            file_sha256=sha,
            fetched_at=FETCHED_AT,
            keyword_list_id=keyword_list.id,
            job_id=job.id,
            status="queued",
        )
    )
    await db.commit()
    return job.id


def _place(path: Path, content: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(content)


async def _run_worker() -> None:
    worker = Worker(async_sessionmaker(get_engine(), expire_on_commit=False))
    assert await worker.run_once(), "aucun job pris : la file était vide"


async def _content(db: AsyncSession, org_id: str) -> dict[str, Any]:
    """Le contenu canonique du réservoir d'une org, **sans les identifiants**.

    C'est ce qui doit être identique entre un run ininterrompu et un run repris ; les `id`
    et les `xmin`, eux, diffèrent forcément d'une organisation à l'autre.
    """
    keywords = (
        await db.execute(
            text("SELECT keyword, locale FROM keywords WHERE org_id = :org ORDER BY keyword"),
            {"org": org_id},
        )
    ).all()
    metrics = (
        await db.execute(
            text(
                "SELECT k.keyword, m.source, m.fetched_at, m.volume, m.kd, m.cpc, m.intent, "
                "m.trend, m.raw FROM keyword_metrics m JOIN keywords k ON k.id = m.keyword_id "
                "WHERE m.org_id = :org ORDER BY k.keyword, m.source"
            ),
            {"org": org_id},
        )
    ).all()
    items = (
        await db.execute(
            text(
                "SELECT k.keyword FROM keyword_list_items i JOIN keywords k ON k.id = i.keyword_id "
                "WHERE i.org_id = :org ORDER BY k.keyword"
            ),
            {"org": org_id},
        )
    ).all()
    maps = (
        await db.execute(
            text(
                "SELECT k.keyword, m.url_normalized, m.source, m.position, m.seen_at, m.page_id "
                "FROM keyword_page_map m JOIN keywords k ON k.id = m.keyword_id "
                "WHERE m.org_id = :org ORDER BY k.keyword, m.url_normalized"
            ),
            {"org": org_id},
        )
    ).all()
    return {"keywords": keywords, "metrics": metrics, "items": items, "maps": maps}


async def _counts(db: AsyncSession, org_id: str) -> dict[str, int]:
    """Décompte exact par table — pour confronter le rapport à l'état réel de la base."""
    tables = {
        "keywords_created": "keywords",
        "metrics_written": "keyword_metrics",
        "list_items_added": "keyword_list_items",
        "maps_written": "keyword_page_map",
    }
    return {
        key: int(
            await db.scalar(
                text(f"SELECT count(*) FROM {table} WHERE org_id = :org"), {"org": org_id}
            )
            or 0
        )
        for key, table in tables.items()
    }


def _positions_fixture(rows: int = 500) -> str:
    """Un export Positions synthétique : des URL, donc des `keyword_page_map` à écrire.

    L'export Keyword Magic réel n'en porte aucune — c'est précisément l'angle mort que ce
    fichier vient couvrir : la table la plus écrite du modèle, et son upsert gardé par
    `seen_at`, doivent traverser une reprise sans doublon ni position incohérente.
    Un mot-clé sur sept classe sur une seconde URL (cannibalisation), et une ligne sur
    treize est strictement répétée (vrai doublon intra-fichier).
    """
    header = "Keyword,Intent,Position,Search Volume,Keyword Difficulty,CPC,URL,Trends"
    lines = [header]
    written = 0
    index = 0
    while written < rows:
        keyword = f"mot cle {index}"
        url = f"https://exemple.fr/p/{index}"
        line = (
            f"{keyword},Commercial,{1 + index % 90},{100 + index},"
            f"{index % 100},0.5{index % 10},{url},"
        )
        lines.append(line)
        written += 1
        if written < rows and index % 7 == 0:
            lines.append(line.replace(url, f"{url}/bis").replace(f",{1 + index % 90},", ",42,"))
            written += 1
        if written < rows and index % 13 == 0:
            lines.append(line)
            written += 1
        index += 1
    return "\n".join(lines) + "\n"


@pytest.mark.parametrize(
    ("source", "expect_maps"),
    [
        # L'export réel — mais il ne porte aucune URL, donc aucun `keyword_page_map`.
        pytest.param("semrush_magic", False, id="magic-sans-url"),
        # La table la plus écrite du modèle, et son upsert gardé par `seen_at`.
        pytest.param("semrush_positions", True, id="positions-avec-maps"),
    ],
)
async def test_resumed_import_matches_an_uninterrupted_one(
    db: AsyncSession,
    import_dir: Path,
    monkeypatch: pytest.MonkeyPatch,
    source: str,
    expect_maps: bool,
) -> None:
    content = (
        MAGIC_REAL.read_bytes()
        if source == "semrush_magic"
        else _positions_fixture().encode("utf-8")
    )

    # --- run témoin, ininterrompu -------------------------------------------------------
    control_org, control_target = await _fresh_org(db)
    control_job_id = await _stage(
        db, control_org, control_target, import_dir, source=source, content=content
    )
    await _run_worker()
    db.expire_all()
    control = await db.get(Job, control_job_id)
    assert control is not None and control.status == "done"
    control_report = dict(control.report or {})
    control_content = await _content(db, control_org)
    assert control_report["rows_read"] >= 500
    # Le scénario n'a d'intérêt que s'il y a bien plusieurs lots à traverser.
    assert control_report["rows_read"] > BATCH * KILL_AT_BATCH
    assert bool(control_content["maps"]) is expect_maps
    # Le rapport doit se réconcilier avec l'état réel de la base, table par table.
    assert await _counts(db, control_org) == {
        key: control_report[key]
        for key in ("keywords_created", "metrics_written", "list_items_added", "maps_written")
    }

    # --- run tué entre deux lots, puis repris --------------------------------------------
    resumed_org, resumed_target = await _fresh_org(db)
    resumed_job_id = await _stage(
        db, resumed_org, resumed_target, import_dir, source=source, content=content
    )

    real_save = service.save_progress
    calls = {"n": 0}

    async def killing_save(session: AsyncSession, job: Job, progress: dict[str, Any]) -> bool:
        calls["n"] += 1
        if calls["n"] == KILL_AT_BATCH:
            # Le lot est écrit mais **pas encore publié** : la session se ferme sans commit.
            raise HardKill
        return await real_save(session, job, progress)

    monkeypatch.setattr(service, "save_progress", killing_save)
    with pytest.raises(HardKill):
        await _run_worker()
    monkeypatch.setattr(service, "save_progress", real_save)

    db.expire_all()
    killed = await db.get(Job, resumed_job_id)
    assert killed is not None
    # Le curseur s'est arrêté au dernier lot **publié** : le lot en vol n'a rien laissé.
    assert killed.progress["batch"] == KILL_AT_BATCH - 1
    assert killed.status == "running"  # jamais finalisé — c'est un kill, pas une exception

    # Ce que fait le balayage des zombies au-delà du seuil (socle §2.5).
    await db.execute(
        update(Job)
        .where(Job.id == resumed_job_id)
        .values(status="queued", started_at=None)
        .execution_options(synchronize_session=False)
    )
    await db.commit()

    await _run_worker()
    db.expire_all()
    finished = await db.get(Job, resumed_job_id)
    assert finished is not None and finished.status == "done"

    # 1. L'état final de la base est celui d'un run ininterrompu — aucun doublon, et pour
    #    Positions cela inclut chaque attache, sa position et sa date d'observation.
    assert await _content(db, resumed_org) == control_content
    # 1 bis. Et le rapport du run repris se réconcilie lui aussi avec la base.
    resumed_counts = await _counts(db, resumed_org)
    assert resumed_counts == await _counts(db, control_org)
    # 2. Et le rapport aussi, clé par clé — compteurs de lecture **et** d'écriture.
    #    `errors_csv` est écarté : il porte le sha du fichier, qui diffère forcément entre
    #    deux imports mis en scène séparément (et vaut `None` quand le fichier est sain).
    resumed_report = dict(finished.report or {})
    resumed_report.pop("errors_csv")
    control_report.pop("errors_csv")
    assert resumed_report == control_report

    for org_id in (control_org, resumed_org):
        await db.execute(delete(Site).where(Site.org_id == org_id))
        await db.execute(text("DELETE FROM keywords WHERE org_id = :org"), {"org": org_id})
        await db.execute(delete(Job).where(Job.org_id == org_id))
    await db.commit()