"""Benchmark T2 du pipeline d'import — **bloquant avant merge** (spec Imports §4.6, §11).

Rejouable :

    uv run python -m benchmarks.import_500k

Il génère lui-même sa fixture (graine fixe, donc chiffres reproductibles) par **amplification
de l'export Semrush réel** — les plafonds d'export Semrush sont à 50 000 lignes, il faut
concaténer pour atteindre les 500 000 exigées. Mots-clés, volumes, KD, CPC et tendances
viennent tous du fichier réel ; seuls le suffixe qui rend les mots-clés distincts, l'URL et la
position sont synthétisés — sans URL, `keyword_page_map`, la table la plus écrite du modèle,
ne serait jamais exercée (l'export Keyword Magic n'en porte aucune).

**Le script vérifie ses propres chiffres.** Chaque compteur d'écriture du rapport est confronté
à un `count(*)` exact filtré par organisation ; une divergence fait échouer le benchmark. Sans
cela, on publie des chiffres qui ne se réconcilient pas — c'est exactement ce qui s'est produit
à la première campagne, où les tailles étaient lues dans `pg_stat_user_tables.n_live_tup`, une
**estimation** du collecteur de statistiques, non filtrée par organisation et polluée par les
campagnes précédentes.

Tourne contre `data_test` par défaut ; nettoie derrière lui. Exclu de l'image Docker
(`.dockerignore`) et hors de `testpaths` : ce n'est pas un test unitaire, il dure ~5 minutes.

**Ne pas lancer `pytest` pendant qu'il tourne.** Les deux partagent `data_test`, et le worker
prend le plus vieux job `queued` **toutes organisations confondues** (socle §2.1) : un worker
de test réclame alors le job du benchmark, le balayage des zombies déposséde celui-ci en cours
de route, et le run échoue sur un rapport vide. Ce n'est pas une régression du pipeline — c'est
la file partagée qui fait son travail.
"""

import asyncio
import csv
import os
import random
import time
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit

os.environ.setdefault(
    "DATABASE_URL", "postgresql+asyncpg://seoinsight:seoinsight@localhost:5432/seoinsight_data_test"
)
os.environ.setdefault("AUTH_BASE_URL", "http://test-auth.local")

BENCH_DIR = Path(__file__).parent
os.environ.setdefault("IMPORT_FILES_DIR", str(BENCH_DIR / ".files"))

from sqlalchemy import delete, text  # noqa: E402
from sqlalchemy.exc import ProgrammingError  # noqa: E402
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker  # noqa: E402

from app.db.engine import get_engine  # noqa: E402
from app.db.guard import assert_local  # noqa: E402
from app.jobs import service  # noqa: E402
from app.jobs.models import Job  # noqa: E402
from app.jobs.worker import Worker  # noqa: E402
from app.keywords import pipeline  # noqa: E402
from app.keywords import service as keywords_service  # noqa: E402
from app.keywords.models import Import, Keyword, KeywordList  # noqa: E402
from app.sites.models import Site, SiteTarget  # noqa: E402

BENCH_DATABASE = "seoinsight_data_test"


def assert_disposable_database(url: str) -> None:
    """Même garde que `app.db_reset`, pour la même raison : ce script est **destructeur**.

    Il supprime toutes les lignes `org_bench_%`, puis `VACUUM` et `ANALYZE` des tables
    partagées — des opérations qu'aucune base autre qu'une base de test locale ne doit subir.
    Le contrôle porte sur l'URL, avant toute connexion : l'hôte doit être local (`assert_local`
    couvre aussi le `host=` en query string, qui l'emporterait sur l'hôte de l'URL), et la base
    doit être celle des tests. Une `DATABASE_URL` de dev qui traîne dans l'environnement ne doit
    pas suffire à faire tourner un benchmark sur les données de travail de l'utilisateur.
    """
    assert_local(url)
    database = urlsplit(url).path.lstrip("/")
    if database != BENCH_DATABASE:
        raise RuntimeError(
            f"Refus : le benchmark écrit et VACUUM des tables partagées, il n'accepte que la "
            f"base {BENCH_DATABASE!r} (DATABASE_URL pointe sur {database!r})"
        )


assert_disposable_database(os.environ["DATABASE_URL"])

API_DIR = BENCH_DIR.parent
# L'export complet (50 000 lignes) donne des mots-clés tous distincts ; à défaut,
# l'échantillon de 500 lignes suffit à faire tourner le script, avec moins de variété.
REAL_FULL = API_DIR / "tests" / "fixtures" / "real" / "semrush_magic_us_full.csv"
REAL_SAMPLE = API_DIR / "tests" / "fixtures" / "real" / "semrush_magic_us.csv"
FIXTURE = BENCH_DIR / ".files" / "bench_positions_500k.csv"

TARGET_ROWS = 500_000
BLOCKS = 10
SEED = 20260831
DUPLICATE_RATE = 0.05  # lignes strictement répétées (vrais doublons intra-fichier)
MULTI_URL_RATE = 0.15  # mots-clés classés sur une seconde URL (cannibalisation)
FETCHED_AT = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
SPEC_BUDGET_SECONDS = 600  # §4.6 : au-delà, paralléliser par fichier

HEADER = (
    "Keyword",
    "Intent",
    "Position",
    "Search Volume",
    "Keyword Difficulty",
    "CPC",
    "URL",
    "Trends",
    "Number of Results",
)
# Les compteurs d'écriture du rapport, et la table qui doit les confirmer exactement.
RECONCILED = {
    "keywords_created": "keywords",
    "metrics_written": "keyword_metrics",
    "list_items_added": "keyword_list_items",
    "maps_written": "keyword_page_map",
}


def build_fixture(path: Path) -> dict[str, int]:
    source = REAL_FULL if REAL_FULL.exists() else REAL_SAMPLE
    with source.open(encoding="utf-8", newline="") as handle:
        base = [row for row in csv.DictReader(handle) if row.get("Volume") is not None]
    print(f"  base réelle : {len(base)} lignes ({source.name})")

    random.seed(SEED)
    path.parent.mkdir(parents=True, exist_ok=True)
    rows = duplicates = multi = 0
    per_block = TARGET_ROWS // BLOCKS
    with path.open("w", encoding="utf-8", newline="") as handle:
        writer = csv.writer(handle)
        writer.writerow(HEADER)
        for block in range(BLOCKS):
            written = 0
            for index, row in enumerate(base):
                if written >= per_block:
                    break
                keyword = row["Keyword"] if block == 0 else f"{row['Keyword']} v{block}"
                slug = f"/p/{block}-{index}"
                line = [
                    keyword,
                    row["Intent"],
                    str(1 + (index % 100)),
                    row["Volume"],
                    row["Keyword Difficulty"],
                    row["CPC (USD)"],
                    f"https://boutique-{block % 3}.example{slug}",
                    row["Trend"],
                    row["Number of Results"],
                ]
                writer.writerow(line)
                written += 1
                rows += 1
                if written < per_block and random.random() < MULTI_URL_RATE:
                    second = list(line)
                    second[2] = str(1 + (index % 100) + 30)
                    second[6] = f"https://boutique-{block % 3}.example{slug}/bis"
                    writer.writerow(second)
                    written += 1
                    rows += 1
                    multi += 1
                if written < per_block and random.random() < DUPLICATE_RATE:
                    writer.writerow(line)
                    written += 1
                    rows += 1
                    duplicates += 1
    return {"rows": rows, "duplicates": duplicates, "multi_url": multi}


async def exact_counts(session: AsyncSession, org_id: str) -> dict[str, int]:
    """`count(*)` **filtré par organisation** — la seule mesure qui se confronte au rapport.

    Surtout pas `pg_stat_user_tables.n_live_tup` : c'est une estimation, à l'échelle de la
    table entière, qui compte les lignes des autres organisations et retarde sur les écritures
    en masse.
    """
    return {
        key: int(
            await session.scalar(
                text(f"SELECT count(*) FROM {table} WHERE org_id = :org"), {"org": org_id}
            )
            or 0
        )
        for key, table in RECONCILED.items()
    }


def reconcile(label: str, report: dict[str, Any], counts: dict[str, int]) -> None:
    expected = {key: report[key] for key in RECONCILED}
    if expected != counts:
        raise AssertionError(
            f"{label} : le rapport ne se réconcilie pas avec la base.\n"
            f"  rapport : {expected}\n  base    : {counts}"
        )
    print(f"  réconciliation     : OK {counts}")


async def measure(
    factory: async_sessionmaker[AsyncSession],
    org_id: str,
    label: str,
    *,
    check_reconciliation: bool,
) -> dict[str, Any]:
    """Chronomètre un run et confronte son rapport à la base.

    `check_reconciliation` n'a de sens que sur le **premier** import d'une organisation :
    les compteurs d'écriture décrivent ce que *ce run* a changé, alors que `count(*)` décrit
    le cumul. Au re-import, l'égalité attendue est ailleurs — 0 écriture, base inchangée.
    """
    async with factory() as probe:
        wal_before = await probe.scalar(text("SELECT pg_current_wal_lsn()::text"))
    started = time.perf_counter()
    assert await Worker(factory).run_once(), "file vide : aucun job à exécuter"
    elapsed = time.perf_counter() - started
    async with factory() as probe:
        wal_after = await probe.scalar(text("SELECT pg_current_wal_lsn()::text"))
        wal = float(
            await probe.scalar(
                text(f"SELECT pg_wal_lsn_diff('{wal_after}'::pg_lsn, '{wal_before}'::pg_lsn)")
            )
            or 0
        )
        report = dict(
            await probe.scalar(
                text(
                    "SELECT report FROM jobs WHERE org_id = :org AND type = 'import_csv' "
                    "ORDER BY created_at DESC LIMIT 1"
                ),
                {"org": org_id},
            )
            or {}
        )
        counts = await exact_counts(probe, org_id)
    batches = -(-report["rows_read"] // pipeline.BATCH_SIZE)
    print(f"\n### {label}")
    rate = report["rows_read"] / elapsed
    print(f"  durée              : {elapsed:8.2f} s ({rate:,.0f} lignes/s)")
    print(f"  WAL généré         : {wal / 1e6:8.1f} Mo")
    print(f"  transactions       : {batches} lots de {pipeline.BATCH_SIZE} lignes (+2)")
    print(f"  rapport            : {report}")
    if check_reconciliation:
        reconcile(label, report, counts)
    assert report["rows_ignored"] == (
        report["rows_read"] - report["rows_error"] - report["rows_inserted"]
    ), "les compteurs de lignes ne s'additionnent pas"
    return {"elapsed": elapsed, "wal": wal, "report": report, "counts": counts}


async def measure_reservoir(
    factory: async_sessionmaker[AsyncSession], org_id: str, target_id: uuid.UUID
) -> None:
    """Chronomètre `GET /keywords` **tel qu'il tourne en production**, à l'échelle 500 k.

    La requête n'est pas écrite ici : elle est construite par `keywords.service.build_query`,
    la seule implémentation du réservoir. Deux pages sont mesurées — sans filtre (le coût
    plancher : appartenance + consolidation de 50 mots-clés) et avec `volume_min`, qui exerce
    le pré-filtre EXISTS puis le re-filtrage sur la valeur consolidée (§3.3, review T2).

    Le plan est publié pour les deux : ce qu'il ne doit **jamais** contenir, c'est un
    `Function Scan` (la consolidation ne serait plus inlinée) ni un `Seq Scan on
    keyword_metrics` (la table la plus grosse du modèle parcourue en entier pour 50 lignes) —
    exactement les deux propriétés que garde `test_keywords_api.py`, ici confrontées au volume.
    """
    cases: tuple[tuple[str, keywords_service.Filters], ...] = (
        ("page de 50, sans filtre", keywords_service.Filters()),
        ("page de 50, volume consolidé >= 1000", keywords_service.Filters(volume_min=1000)),
    )
    async with factory() as session:
        for label, filters in cases:
            sql, names = keywords_service.build_query(
                keywords_service.CONSOLIDATED, filters, cursor=None
            )
            params: dict[str, Any] = {
                "org": org_id,
                "target": target_id,
                "limit": 51,
                "preferred": "semrush_positions",
                "view": keywords_service.CONSOLIDATED,
                **{name: getattr(filters, name) for name in names},
            }
            for _ in range(2):  # une passe à froid, une à chaud
                started = time.perf_counter()
                rows = (await session.execute(text(sql), params)).all()
                page = time.perf_counter() - started
            plan = (
                await session.execute(text(f"EXPLAIN (ANALYZE, BUFFERS, COSTS OFF) {sql}"), params)
            ).scalars()
            lines = list(plan)
            #  La première ligne « Buffers » est celle du nœud racine : elle cumule déjà tout
            #  l'arbre. Les additionner toutes compterait chaque bloc autant de fois qu'il y a
            #  de niveaux au-dessus de lui.
            root_buffers = next(
                (line.strip() for line in lines if line.strip().startswith("Buffers:")), "n/a"
            )
            print(f"\n### Requête réservoir — {label}")
            print(f"  durée              : {page * 1000:8.1f} ms ({len(rows)} lignes)")
            print(f"  {root_buffers} (nœud racine)")
            for guard in ("Function Scan", "Seq Scan on keyword_metrics"):
                verdict = "PRÉSENT ⛔" if any(guard in line for line in lines) else "absent ✅"
                print(f"  {guard:26} : {verdict}")
            print("  EXPLAIN :")
            for line in lines:
                print("   ", line)


async def main() -> None:
    if not FIXTURE.exists():
        print("Construction de la fixture 500 k…")
        print(f"  {build_fixture(FIXTURE)}")
    print(f"Fixture : {FIXTURE} ({FIXTURE.stat().st_size / 1e6:.1f} Mo)")

    engine = get_engine()
    factory = async_sessionmaker(engine, expire_on_commit=False)
    org_id = f"org_bench_{uuid.uuid4().hex[:8]}"

    async with factory() as session:
        # Les campagnes précédentes fausseraient les tailles et le décompte de lignes mortes.
        await session.execute(text("DELETE FROM imports WHERE org_id LIKE 'org_bench_%'"))
        await session.execute(text("DELETE FROM sites WHERE org_id LIKE 'org_bench_%'"))
        await session.execute(text("DELETE FROM keywords WHERE org_id LIKE 'org_bench_%'"))
        await session.execute(text("DELETE FROM jobs WHERE org_id LIKE 'org_bench_%'"))
        await session.commit()
    async with engine.connect() as connection:
        await connection.execution_options(isolation_level="AUTOCOMMIT")
        await connection.execute(
            text("VACUUM keywords, keyword_metrics, keyword_page_map, keyword_list_items")
        )
        try:
            # Remise à zéro des compteurs de statistiques : agréable mais réservée au
            # superutilisateur. Le `VACUUM` ci-dessus suffit à ramener `n_dead_tup` à 0, qui
            # est la seule statistique que le benchmark publie.
            await connection.execute(
                text(
                    "SELECT pg_stat_reset_single_table_counters(relid) FROM pg_stat_user_tables "
                    "WHERE relname LIKE 'keyword%'"
                )
            )
        except ProgrammingError:
            print("  (compteurs de stats non réinitialisés : rôle non superutilisateur)")

    async with factory() as session:
        site = Site(org_id=org_id, domain=f"{uuid.uuid4().hex}.example", name="Bench")
        session.add(site)
        await session.flush()
        target = SiteTarget(site_id=site.id, org_id=org_id, country="FR", language="fr")
        session.add(target)
        await session.flush()
        keyword_list = KeywordList(org_id=org_id, site_target_id=target.id, name="Bench")
        session.add(keyword_list)
        await session.commit()
        target_id, list_id = target.id, keyword_list.id

    async def stage() -> None:
        sha = uuid.uuid4().hex * 2
        path = Path(os.environ["IMPORT_FILES_DIR"]) / org_id / f"{sha}.csv"
        path.parent.mkdir(parents=True, exist_ok=True)
        path.symlink_to(FIXTURE)
        async with factory() as session:
            import_id = uuid.uuid4()
            job = await service.enqueue(
                session,
                org_id,
                pipeline.JOB_TYPE,
                payload={
                    "import_id": str(import_id),
                    "file_sha256": sha,
                    "source": "semrush_positions",
                    "site_target_id": str(target_id),
                    "locale": "fr-FR",
                    "fetched_at": FETCHED_AT.isoformat(),
                    "keyword_list_id": str(list_id),
                },
            )
            session.add(
                Import(
                    id=import_id,
                    org_id=org_id,
                    site_target_id=target_id,
                    source="semrush_positions",
                    kind="csv",
                    file_sha256=sha,
                    fetched_at=FETCHED_AT,
                    keyword_list_id=list_id,
                    job_id=job.id,
                    status="queued",
                )
            )
            await session.commit()

    await stage()
    first = await measure(
        factory, org_id, "Import initial (500 k lignes)", check_reconciliation=True
    )

    async with factory() as session:
        print(
            "\n  tables après import (taille totale, puis n_dead_tup — **estimation** du "
            "collecteur pg_stat, jamais un count : base vidée et VACUUM au départ) :"
        )
        for name, total, dead in (
            await session.execute(
                text(
                    "SELECT relname, pg_size_pretty(pg_total_relation_size(relid)), n_dead_tup "
                    "FROM pg_stat_user_tables WHERE relname IN "
                    "('keywords','keyword_metrics','keyword_page_map','keyword_list_items') "
                    "ORDER BY relname"
                )
            )
        ).all():
            print(f"    {name:22} {total:>10}   mortes≈{dead} (estimation)")
        await session.execute(text("ANALYZE keywords, keyword_metrics, keyword_page_map"))
        await session.commit()

    await stage()
    second = await measure(
        factory,
        org_id,
        "Re-import du même fichier (chemin « ignorées »)",
        check_reconciliation=False,
    )
    assert second["report"]["rows_inserted"] == 0, "le rejeu a écrit quelque chose"
    assert second["counts"] == first["counts"], "le rejeu a changé le contenu de la base"

    async with factory() as session:
        dead = (
            await session.execute(
                text(
                    "SELECT relname, n_dead_tup FROM pg_stat_user_tables WHERE relname IN "
                    "('keywords','keyword_metrics','keyword_page_map') ORDER BY relname"
                )
            )
        ).all()
        print(f"\n  lignes mortes après re-import (n_dead_tup, estimation pg_stat) : {dict(dead)}")

    # --- requête réservoir consolidée : filtre sur valeur calculée + curseur (§11) --------
    #
    # **La requête mesurée est celle que la production exécute**, construite par
    # `keywords.service.build_query` — jamais une copie écrite ici. Une seconde rédaction du
    # réservoir dans le benchmark divergerait au premier changement et fabriquerait de la
    # fausse confiance : c'est exactement ce qui s'est produit à la campagne précédente, où
    # le benchmark mesurait encore le `JOIN` naïf que T3 avait déjà remplacé.
    await measure_reservoir(factory, org_id, target_id)

    print("\n### Résumé")
    print(f"  import 500 k : {first['elapsed']:.1f} s (budget spec : {SPEC_BUDGET_SECONDS} s)")
    print(f"  re-import    : {second['elapsed']:.1f} s, rows_inserted=0, base inchangée")
    assert first["elapsed"] < SPEC_BUDGET_SECONDS, (
        "au-delà du budget §4.6 : paralléliser l'import par fichier"
    )

    async with factory() as session:
        await session.execute(delete(Site).where(Site.org_id == org_id))
        await session.execute(delete(Keyword).where(Keyword.org_id == org_id))
        await session.execute(delete(Job).where(Job.org_id == org_id))
        await session.commit()
    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())