"""Le réservoir par cible : `GET /keywords` (spec Imports §6, §11.1).

Quatre promesses y sont vérifiées, et rien d'autre n'a autant d'importance :

1. le **prédicat d'appartenance** (review I6) — membre d'une liste de la cible **OU** porteur
   d'un map sur la cible ; un mot-clé retiré de toute liste mais attaché reste visible, un
   mot-clé de l'organisation sans lien avec la cible est invisible ;
2. les **observations côte à côte** (spec 2026-09-01 §3.2) — chaque métrique porte une
   entrée par source qui l'a observée ; la source préférée passe en **tête** et n'en masque
   aucune ; un champ sans observation rend `[]`, un mot-clé sans aucune métrique rend six
   listes vides plutôt qu'un 500 ; la vue par source rend la **même** forme, à zéro ou une
   entrée ;
3. les **filtres et le curseur** — dont la règle « au moins une source satisfait » : un
   mot-clé à 5 000 chez l'une et 10 chez l'autre apparaît dans `volume_min=1000` **comme**
   dans `volume_max=100`, et c'est la régression du §1 qui a ouvert ce lot ;
4. le **plan d'exécution** — la fonction de consolidation est inlinée (aucun `Function Scan`)
   et `keyword_metrics` n'est jamais parcourue au complet. C'est le seul point du module que
   des assertions fonctionnelles ne peuvent pas attraper : la requête resterait *juste* en
   devenant lente, et le benchmark T2 a montré ce que « lente » veut dire ici.
"""

import uuid
from collections.abc import AsyncIterator
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any

import pytest
import pytest_asyncio
from httpx import AsyncClient
from sqlalchemy import delete, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.errors import AppError
from app.db.cursor import SEARCH_MAX_ROWS, decode_offset, encode_offset
from app.keywords import service
from app.keywords.models import (
    Keyword,
    KeywordList,
    KeywordListItem,
    KeywordMetric,
    KeywordPageMap,
    KeywordTag,
    OrgSettings,
    Tag,
)
from app.sites.models import Site, SiteTarget
from tests.conftest import Fixture

NOW = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
OLD = NOW - timedelta(days=90)
LOCALE = "fr-FR"
# Les six champs d'une ligne du réservoir, tous rendus sous la même forme : une liste
# d'observations. Même nom et même ordre que dans `test_keyword_metrics_latest.py`.
METRICS = ("volume", "kd", "cpc", "intent", "serp_features", "trend")


def iso(moment: datetime) -> str:
    """`Z` et non `+00:00` : c'est la forme que Pydantic sérialise pour un instant UTC."""
    return moment.isoformat().replace("+00:00", "Z")


Headers = dict[str, str]


@pytest.fixture
def headers(auth_headers: Any, sandbox: Fixture) -> Headers:
    return auth_headers(sandbox.org_id)


class Builder:
    """Construit un réservoir à la main : ce module teste des lectures, jamais le pipeline."""

    def __init__(self, db: AsyncSession, sandbox: Fixture) -> None:
        self.db = db
        self.org_id = sandbox.org_id
        self.target_id = sandbox.target_id

    async def keyword(self, text_: str) -> Keyword:
        row = Keyword(org_id=self.org_id, locale=LOCALE, keyword=text_)
        self.db.add(row)
        await self.db.flush()
        return row

    async def metric(
        self, keyword: Keyword, source: str, fetched_at: datetime, **values: Any
    ) -> None:
        self.db.add(
            KeywordMetric(
                org_id=self.org_id,
                keyword_id=keyword.id,
                source=source,
                fetched_at=fetched_at,
                raw={},
                **values,
            )
        )
        await self.db.flush()

    async def list(
        self, name: str, *keywords: Keyword, target: uuid.UUID | None = None
    ) -> KeywordList:
        row = KeywordList(org_id=self.org_id, site_target_id=target or self.target_id, name=name)
        self.db.add(row)
        await self.db.flush()
        for keyword in keywords:
            self.db.add(KeywordListItem(list_id=row.id, keyword_id=keyword.id, org_id=self.org_id))
        await self.db.flush()
        return row

    async def tag(self, name: str, *keywords: Keyword) -> Tag:
        row = Tag(org_id=self.org_id, name=name)
        self.db.add(row)
        await self.db.flush()
        for keyword in keywords:
            self.db.add(KeywordTag(keyword_id=keyword.id, tag_id=row.id, org_id=self.org_id))
        await self.db.flush()
        return row

    async def map(
        self,
        keyword: Keyword,
        url: str,
        *,
        source: str = "import_url",
        page_id: uuid.UUID | None = None,
    ) -> None:
        self.db.add(
            KeywordPageMap(
                org_id=self.org_id,
                site_target_id=self.target_id,
                keyword_id=keyword.id,
                url_normalized=url,
                page_id=page_id,
                source=source,
                seen_at=NOW,
            )
        )
        await self.db.flush()

    async def done(self) -> None:
        await self.db.commit()


@pytest_asyncio.fixture
async def build(db: AsyncSession, sandbox: Fixture) -> Builder:
    return Builder(db, sandbox)


async def _keywords(client: AsyncClient, headers: Headers, sandbox: Fixture, **params: Any) -> Any:
    response = await client.get(
        "/keywords",
        headers=headers,
        params={"site_target_id": str(sandbox.target_id), **params},
    )
    return response


async def _texts(
    client: AsyncClient, headers: Headers, sandbox: Fixture, **params: Any
) -> list[str]:
    response = await _keywords(client, headers, sandbox, **params)
    assert response.status_code == 200, response.text
    return [row["keyword"] for row in response.json()["items"]]


# --- prédicat d'appartenance (review I6) ----------------------------------------------------


async def test_belongs_by_list_or_by_map(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder, db: AsyncSession
) -> None:
    listed = await build.keyword("dans une liste")
    mapped = await build.keyword("seulement attaché")
    stranger = await build.keyword("sans lien avec la cible")
    other_target = SiteTarget(
        site_id=sandbox.site_id, org_id=sandbox.org_id, country="BE", language="fr"
    )
    db.add(other_target)
    await db.flush()
    elsewhere = await build.keyword("liste d une autre cible")
    await build.list("Cible", listed)
    await build.list("Ailleurs", elsewhere, target=other_target.id)
    await build.map(mapped, "https://exemple.fr/p")
    await build.done()

    visible = await _texts(client, headers, sandbox)
    assert set(visible) == {"dans une liste", "seulement attaché"}
    assert stranger.keyword not in visible
    assert elsewhere.keyword not in visible


async def test_attach_status(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    orphan = await build.keyword("orphelin")
    pending = await build.keyword("en attente")
    attached = await build.keyword("attaché")
    multi = await build.keyword("cannibalisation")
    await build.list("Toutes", orphan, pending, attached, multi)
    await build.map(pending, "https://exemple.fr/a")
    await build.map(attached, "https://exemple.fr/b", page_id=uuid.uuid4())
    await build.map(multi, "https://exemple.fr/c")
    await build.map(multi, "https://exemple.fr/d")
    await build.done()

    rows = (await _keywords(client, headers, sandbox)).json()["items"]
    statuses = {row["keyword"]: row["attach_status"] for row in rows}
    assert statuses == {
        "orphelin": "orphan",
        "en attente": "pending",
        "attaché": "attached",
        "cannibalisation": "multi",
    }
    assert await _texts(client, headers, sandbox, attach_status="orphan") == ["orphelin"]
    assert await _texts(client, headers, sandbox, attach_status="multi") == ["cannibalisation"]


async def test_unknown_attach_status_is_422(
    client: AsyncClient, headers: Headers, sandbox: Fixture
) -> None:
    assert (await _keywords(client, headers, sandbox, attach_status="perdu")).status_code == 422


async def test_unknown_target_is_404(client: AsyncClient, headers: Headers) -> None:
    response = await client.get(
        "/keywords", headers=headers, params={"site_target_id": str(uuid.uuid4())}
    )
    assert response.status_code == 404


# --- consolidation (décision n°3) -----------------------------------------------------------


async def test_preferred_source_orders_without_masking(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder, db: AsyncSession
) -> None:
    """La source préférée passe en **tête** de la liste — elle n'en fait disparaître aucune.

    C'est le renversement de la spec 2026-09-01 §3.4 : elle n'élit plus une valeur par champ,
    elle ordonne des observations qui restent toutes rendues.
    """
    keyword = await build.keyword("chaussures rouges")
    # La préférée est **vieille** et n'a pas de KD : elle doit passer devant sur le volume
    # malgré son âge, sans effacer l'autre observation ni s'approprier le KD.
    await build.metric(keyword, "generic_csv", OLD, volume=100)
    await build.metric(keyword, "semrush_magic", NOW, volume=999, kd=Decimal(42))
    await build.list("L", keyword)
    await build.done()

    default = (await _keywords(client, headers, sandbox)).json()
    assert default["preferred_source"] is None
    # Sans préférence, l'ordre est la fraîcheur — et les deux sources sont là.
    assert default["items"][0]["volume"] == [
        {"value": 999, "source": "semrush_magic", "fetched_at": iso(NOW)},
        {"value": 100, "source": "generic_csv", "fetched_at": iso(OLD)},
    ]

    db.add(OrgSettings(org_id=sandbox.org_id, preferred_source="generic_csv"))
    await db.commit()
    preferred = (await _keywords(client, headers, sandbox)).json()
    assert preferred["preferred_source"] == "generic_csv"
    row = preferred["items"][0]
    # La préférée en tête bien qu'elle soit la plus ancienne, l'autre juste derrière.
    assert row["volume"] == [
        {"value": 100, "source": "generic_csv", "fetched_at": iso(OLD)},
        {"value": 999, "source": "semrush_magic", "fetched_at": iso(NOW)},
    ]
    # Trouée sur le KD, la source préférée ne prive pas l'autre de le porter.
    assert row["kd"] == [{"value": 42.0, "source": "semrush_magic", "fetched_at": iso(NOW)}]
    # Un champ qu'aucune source n'a observé rend une liste **vide**, jamais une entrée creuse.
    assert row["cpc"] == []
    assert row["last_updated"] == iso(NOW)


async def test_a_keyword_without_any_metric_renders_empty_lists(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Un mot-clé attaché à la main n'a aucune observation : six listes vides, pas un 500.

    Il n'apparaît pas du tout dans `keyword_metrics_latest` (`GROUP BY keyword_id`), donc le
    `LEFT JOIN LATERAL` rend NULL sur les six colonnes — un cas que le `coalesce` de la
    fonction SQL, qui ne protège qu'un champ vide d'un mot-clé qui a d'autres métriques, ne
    couvre pas. Sans le `or []` de `_row`, Pydantic lèverait sur `list[Observation[T]]`.
    """
    bare = await build.keyword("attaché à la main")
    await build.map(bare, "https://exemple.fr/p")
    await build.done()

    row = (await _keywords(client, headers, sandbox)).json()["items"][0]
    assert row["keyword"] == "attaché à la main"
    assert [row[field] for field in METRICS] == [[]] * len(METRICS)
    assert row["last_updated"] is None
    # Même forme en vue par source : des listes vides, pas des `null`.
    other = (await _keywords(client, headers, sandbox, view="generic_csv")).json()["items"][0]
    assert other["volume"] == []


async def test_source_view_shows_raw_values_of_that_source_only(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    keyword = await build.keyword("bottines")
    await build.metric(
        keyword,
        "generic_csv",
        OLD,
        volume=100,
        kd=Decimal("12.50"),
        cpc=Decimal("0.75"),
        intent="commercial",
        serp_features={"local_pack": True},
        trend=[1, 2, 3],
    )
    await build.metric(keyword, "semrush_magic", NOW, volume=999)
    orphan_of_the_source = await build.keyword("sandales")
    await build.metric(orphan_of_the_source, "semrush_magic", NOW, volume=1)
    await build.list("L", keyword, orphan_of_the_source)
    await build.done()

    page = (await _keywords(client, headers, sandbox, view="generic_csv")).json()
    assert page["view"] == "generic_csv"
    rows = {row["keyword"]: row for row in page["items"]}
    # **La même forme que la vue consolidée** (décision Q13) : un tableau, ici d'une seule
    # entrée, sur les **six** champs. C'est toute la promesse « une seule branche à écrire
    # côté web » — une valeur nue sur un seul champ suffirait à la rompre.
    row = rows["bottines"]
    assert {field: [entry["value"] for entry in row[field]] for field in METRICS} == {
        "volume": [100],
        "kd": [12.5],
        "cpc": [0.75],
        "intent": ["commercial"],
        "serp_features": [{"local_pack": True}],
        "trend": [[1, 2, 3]],
    }
    # La provenance est celle de la source demandée, pas de la plus fraîche du mot-clé.
    assert {
        (entry["source"], entry["fetched_at"]) for field in METRICS for entry in row[field]
    } == {("generic_csv", iso(OLD))}
    # Le mot-clé sans observation de cette source reste **visible** (il appartient au
    # réservoir de la cible), avec six listes vides.
    assert [rows["sandales"][field] for field in METRICS] == [[]] * len(METRICS)
    assert rows["sandales"]["last_updated"] is None


async def test_unknown_view_is_422(client: AsyncClient, headers: Headers, sandbox: Fixture) -> None:
    assert (await _keywords(client, headers, sandbox, view="ahrefs")).status_code == 422


# --- filtres --------------------------------------------------------------------------------


async def test_full_text_and_fuzzy_search(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    a = await build.keyword("chaussures rouges")
    b = await build.keyword("bottines noires")
    await build.list("L", a, b)
    await build.done()
    assert await _texts(client, headers, sandbox, q="rouges") == ["chaussures rouges"]
    # Repli trigramme : la faute de frappe ne passe pas par le tsvector.
    assert await _texts(client, headers, sandbox, q="chaussure rouge") == ["chaussures rouges"]
    assert await _texts(client, headers, sandbox, q="introuvable") == []


# --- recherche : le correctif T6 ---------------------------------------------------------------
#
# Le retour de test réel qui a ouvert ce lot : `q = "jewelry manufacturing"` rendait « zbl
# jewelry » et « wire wrapping… » en tête. Deux fautes cumulées — un `OR` trigramme qui
# élargissait au lieu de tolérer, et aucun classement par pertinence. Les tests ci-dessous
# tiennent les deux moitiés du correctif ; ils sont écrits sur le cas réel, pas sur un cas
# d'école, parce que c'est ce cas-là qui a échoué.


async def test_every_word_is_required(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """« jewelry manufacturing » ne rend que ce qui porte **les deux** mots."""
    both = await build.keyword("jewelry manufacturing")
    also_both = await build.keyword("jewelry manufacturing process")
    one_word = await build.keyword("zbl jewelry")
    other_word = await build.keyword("contract manufacturing china")
    neither = await build.keyword("wire wrapping tutorial")
    await build.list("L", both, also_both, one_word, other_word, neither)
    await build.done()

    found = await _texts(client, headers, sandbox, q="jewelry manufacturing")
    assert set(found) == {"jewelry manufacturing", "jewelry manufacturing process"}
    # Le bruit qui avait motivé le lot : partager **un** mot ne suffit plus.
    assert "zbl jewelry" not in found
    assert "contract manufacturing china" not in found


async def test_last_word_is_a_prefix(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Le mot en cours de frappe est un préfixe — sans quoi la liste se viderait à chaque lettre."""
    match = await build.keyword("jewelry manufacturing")
    await build.list("L", match)
    await build.done()
    page = (await _keywords(client, headers, sandbox, q="jewelry manufac")).json()
    assert [row["keyword"] for row in page["items"]] == ["jewelry manufacturing"]
    #  Trouvé par la tsquery, donc **pas** par le repli : c'est bien le préfixe qui a joué.
    assert page["fuzzy"] is False
    #  Le préfixe ne porte que sur le **dernier** mot : un premier mot tronqué ne passe pas la
    #  tsquery. La ligne revient quand même, mais par le repli trigramme, qui l'annonce.
    truncated = (await _keywords(client, headers, sandbox, q="jewel manufacturing")).json()
    assert [row["keyword"] for row in truncated["items"]] == ["jewelry manufacturing"]
    assert truncated["fuzzy"] is True


async def test_results_are_ordered_by_relevance(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Égalité exacte, puis préfixe de la chaîne entière, puis le reste."""
    #  Créés dans un ordre qui contredit le classement attendu : sous l'ancien `ORDER BY id
    #  DESC`, ce test rendrait exactement l'inverse.
    exact = await build.keyword("jewelry manufacturing")
    prefixed = await build.keyword("jewelry manufacturing process")
    scattered = await build.keyword("manufacturing of fine jewelry")
    await build.list("L", exact, prefixed, scattered)
    await build.done()

    found = await _texts(client, headers, sandbox, q="jewelry manufacturing")
    assert found[0] == "jewelry manufacturing"
    assert found[1] == "jewelry manufacturing process"
    assert found[2] == "manufacturing of fine jewelry"


async def test_trigram_is_a_fallback_and_says_so(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Le trigramme ne s'exécute que si la tsquery ne rend rien, et la réponse le déclare."""
    exact = await build.keyword("chaussures rouges")
    await build.list("L", exact)
    await build.done()

    page = (await _keywords(client, headers, sandbox, q="rouges")).json()
    assert [row["keyword"] for row in page["items"]] == ["chaussures rouges"]
    assert page["fuzzy"] is False

    #  Faute de frappe : la tsquery ne rend rien, le repli prend le relais **et le dit**.
    approximate = (await _keywords(client, headers, sandbox, q="chaussure rouge")).json()
    assert [row["keyword"] for row in approximate["items"]] == ["chaussures rouges"]
    assert approximate["fuzzy"] is True

    #  Hors recherche, le drapeau est toujours faux.
    assert (await _keywords(client, headers, sandbox)).json()["fuzzy"] is False


async def test_search_paginates_by_rank(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """En recherche l'ordre est la pertinence : le curseur reprend au rang, pas à la clé."""
    keywords = [await build.keyword(f"jewelry manufacturing {index:02d}") for index in range(7)]
    await build.list("L", *keywords)
    await build.done()

    seen: list[str] = []
    cursor: str | None = None
    for _ in range(10):
        params: dict[str, Any] = {"limit": 3, "q": "jewelry manufacturing"}
        if cursor is not None:
            params["cursor"] = cursor
        page = (await _keywords(client, headers, sandbox, **params)).json()
        seen.extend(row["keyword"] for row in page["items"])
        cursor = page["next_cursor"]
        if cursor is None:
            break
    assert cursor is None
    assert len(seen) == len(set(seen)) == 7

    #  Les deux dialectes de curseur ne se mélangent pas : un curseur de keyset présenté à une
    #  recherche est une entrée invalide, pas un rang lu au hasard.
    keyset = (await _keywords(client, headers, sandbox, limit=1)).json()["next_cursor"]
    assert (
        await _keywords(client, headers, sandbox, q="jewelry manufacturing", cursor=keyset)
    ).status_code == 422


async def test_fuzzy_fallback_never_offers_a_next_page(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Le repli rend **une** page (review F1) : il n'est atteint que sur la première.

    Émettre un `next_cursor` mènerait à une page 2 qui rejouerait la recherche exacte — celle
    qui n'a précisément rien rendu —, donc à une page vide. Le filet à fautes de frappe n'est
    pas un mode de navigation.
    """
    keywords = [await build.keyword(f"chaussures rouges {index:02d}") for index in range(7)]
    await build.list("L", *keywords)
    await build.done()

    page = (await _keywords(client, headers, sandbox, q="chaussure rouge", limit=3)).json()
    assert page["fuzzy"] is True
    assert len(page["items"]) == 3
    #  Sept lignes correspondent, trois sont rendues, et pourtant aucune suite n'est promise :
    #  c'est le contrat du repli, et `fuzzy` est ce qui le dit au client.
    assert page["next_cursor"] is None


async def test_a_forged_offset_cursor_is_422_not_500(
    client: AsyncClient, headers: Headers, sandbox: Fixture
) -> None:
    """Un rang forgé au-delà de la borne est une entrée invalide (review F2).

    Vingt-cinq chiffres décodent en un entier Python parfaitement valide, que le driver refuse
    ensuite de lier à un `bigint` : sans la validation au décodage, c'était une `DataError`
    remontée en 500 — alors que rien n'est cassé côté serveur.
    """
    forged = encode_offset(10**24)
    response = await _keywords(client, headers, sandbox, q="jewelry", cursor=forged)
    assert response.status_code == 422, response.text
    assert "Curseur hors limite" in response.json()["detail"]
    #  La borne elle-même reste lisible : c'est un rang atteignable, pas une valeur refusée.
    assert decode_offset(encode_offset(SEARCH_MAX_ROWS)) == SEARCH_MAX_ROWS
    with pytest.raises(AppError):
        decode_offset(encode_offset(SEARCH_MAX_ROWS + 1))


async def test_search_without_any_word_finds_nothing(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    """Une saisie sans lexème n'a pas de tsquery : elle part au repli, qui ne trouve rien."""
    keyword = await build.keyword("jewelry manufacturing")
    await build.list("L", keyword)
    await build.done()
    assert await _texts(client, headers, sandbox, q="???") == []


def test_build_tsquery_joins_with_and_and_prefixes_the_last_word() -> None:
    assert service.build_tsquery("jewelry manufacturing") == "jewelry & manufacturing:*"
    assert service.build_tsquery("  Jewelry   ") == "jewelry:*"
    #  Les opérateurs de `to_tsquery` ne survivent pas à l'extraction des lexèmes : il n'y a
    #  donc rien à échapper, et aucune saisie ne peut produire d'erreur de syntaxe tsquery.
    assert service.build_tsquery("a & b | !c:*") == "a & b & c:*"
    assert service.build_tsquery("e-commerce") == "e & commerce:*"
    assert service.build_tsquery("???") is None
    assert service.build_tsquery("") is None


# --- suggestions au typing ----------------------------------------------------------------------


async def _suggest(client: AsyncClient, headers: Headers, sandbox: Fixture, **params: Any) -> Any:
    return await client.get(
        "/keywords/suggest",
        headers=headers,
        params={"site_target_id": str(sandbox.target_id), **params},
    )


async def test_suggest_ranks_the_prefix_first(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    exact = await build.keyword("jewelry manufacturing")
    prefixed = await build.keyword("jewelry manufacturing process")
    scattered = await build.keyword("manufacturing of fine jewelry")
    unrelated = await build.keyword("zbl jewelry")
    await build.list("L", exact, prefixed, scattered, unrelated)
    await build.metric(exact, "semrush_magic", NOW, volume=1200)
    #  Une **seconde** source, plus récente et plus basse : sans elle, le maximum et la
    #  dernière observation coïncideraient et ce test passerait par hasard.
    await build.metric(exact, "generic_csv", NOW + timedelta(days=1), volume=800)
    await build.done()

    items = (await _suggest(client, headers, sandbox, q="jewelry manufacturing")).json()["items"]
    assert [row["keyword"] for row in items] == [
        "jewelry manufacturing",
        "jewelry manufacturing process",
        "manufacturing of fine jewelry",
    ]
    #  Un repère de taille accompagne la suggestion — sans lui, deux propositions voisines ne
    #  se départagent pas. C'est le **maximum toutes sources** (§3.5), pas la plus fraîche : la
    #  complétion n'a la place que d'un chiffre, et il ne doit dépendre d'aucun réglage.
    assert items[0]["volume"] == 1200
    assert items[1]["volume"] is None
    assert items[0]["keyword_id"] == str(exact.id)


async def test_suggest_is_bounded_to_ten(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    keywords = [await build.keyword(f"jewelry manufacturing {index:02d}") for index in range(14)]
    await build.list("L", *keywords)
    await build.done()
    items = (await _suggest(client, headers, sandbox, q="jewelry")).json()["items"]
    assert len(items) == 10


async def test_suggest_respects_the_belonging_predicate_and_the_org(
    client: AsyncClient,
    auth_headers: Any,
    headers: Headers,
    sandbox: Fixture,
    build: Builder,
    db: AsyncSession,
) -> None:
    """Même prédicat d'appartenance que le réservoir : ce qu'on propose est ce qu'on rendra."""
    listed = await build.keyword("jewelry manufacturing")
    stranger = await build.keyword("jewelry manufacturing ailleurs")
    other_target = SiteTarget(
        site_id=sandbox.site_id, org_id=sandbox.org_id, country="BE", language="fr"
    )
    db.add(other_target)
    await db.flush()
    elsewhere = await build.keyword("jewelry manufacturing autre cible")
    await build.list("L", listed)
    await build.list("Ailleurs", elsewhere, target=other_target.id)
    await build.done()

    items = (await _suggest(client, headers, sandbox, q="jewelry")).json()["items"]
    assert [row["keyword"] for row in items] == ["jewelry manufacturing"]
    assert stranger.keyword not in {row["keyword"] for row in items}

    #  Une autre organisation ne voit rien — et reçoit un 404 sur la cible, jamais un 403.
    stranger_headers = auth_headers(f"org_{uuid.uuid4().hex}")
    assert (
        await client.get(
            "/keywords/suggest",
            headers=stranger_headers,
            params={"site_target_id": str(sandbox.target_id), "q": "jewelry"},
        )
    ).status_code == 404


async def test_suggest_falls_back_to_trigram(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    keyword = await build.keyword("chaussures rouges")
    await build.list("L", keyword)
    await build.done()
    items = (await _suggest(client, headers, sandbox, q="chaussure rouge")).json()["items"]
    assert [row["keyword"] for row in items] == ["chaussures rouges"]


async def test_suggest_needs_two_characters(
    client: AsyncClient, headers: Headers, sandbox: Fixture
) -> None:
    assert (await _suggest(client, headers, sandbox, q="j")).status_code == 422
    assert (await _suggest(client, headers, sandbox, q="")).status_code == 422
    assert (await _suggest(client, headers, sandbox)).status_code == 422


async def test_filters_by_list_tag_source_and_intent(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    a = await build.keyword("a mot")
    b = await build.keyword("b mot")
    listed = await build.list("Ciblée", a)
    await build.list("Autre", b)
    tag = await build.tag("prio", b)
    await build.metric(a, "semrush_magic", NOW, volume=10, intent="commercial")
    await build.metric(b, "generic_csv", NOW, volume=20, intent="informational")
    await build.done()

    assert await _texts(client, headers, sandbox, list_id=str(listed.id)) == ["a mot"]
    assert await _texts(client, headers, sandbox, tag_id=str(tag.id)) == ["b mot"]
    assert await _texts(client, headers, sandbox, source="semrush_magic") == ["a mot"]
    assert await _texts(client, headers, sandbox, intent="informational") == ["b mot"]
    rows = (await _keywords(client, headers, sandbox, tag_id=str(tag.id))).json()["items"]
    assert [t["name"] for t in rows[0]["tags"]] == ["prio"]
    assert [entry["name"] for entry in rows[0]["lists"]] == ["Autre"]


async def test_volume_filters_accept_any_source(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder, db: AsyncSession
) -> None:
    """« Au moins une source satisfait » — la régression du §1, retournée en test.

    Le mot-clé ci-dessous porte 5 000 chez une source et 10 chez l'autre. L'ancienne
    consolidation en **élisait** une (10, la plus fraîche) : le mot-clé disparaissait de
    « volume ≥ 1000 » alors qu'une de ses sources l'y plaçait, et c'est exactement ce qui a été
    constaté sur `jewelry manufacturer`. Il apparaît désormais dans les **deux** filtres — ce
    qui est exact, puisque la ligne affiche les deux valeurs côte à côte.
    """
    keyword = await build.keyword("piège")
    await build.metric(keyword, "generic_csv", OLD, volume=5000)
    await build.metric(keyword, "semrush_magic", NOW, volume=10)
    big = await build.keyword("vrai gros volume")
    await build.metric(big, "semrush_magic", NOW, volume=5000)
    await build.list("L", keyword, big)
    await build.done()

    assert set(await _texts(client, headers, sandbox, volume_min=1000)) == {
        "piège",
        "vrai gros volume",
    }
    assert set(await _texts(client, headers, sandbox, volume_max=100)) == {"piège"}

    # La source préférée **ordonne** les observations, elle n'en masque plus : elle ne peut
    # donc plus déplacer le résultat d'un filtre. Les deux ensembles sont inchangés.
    db.add(OrgSettings(org_id=sandbox.org_id, preferred_source="generic_csv"))
    await db.commit()
    assert set(await _texts(client, headers, sandbox, volume_min=1000)) == {
        "piège",
        "vrai gros volume",
    }
    assert set(await _texts(client, headers, sandbox, volume_max=100)) == {"piège"}


async def test_updated_since(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    fresh = await build.keyword("fraîche")
    stale = await build.keyword("périmée")
    await build.metric(fresh, "semrush_magic", NOW, volume=1)
    await build.metric(stale, "semrush_magic", OLD, volume=1)
    await build.list("L", fresh, stale)
    await build.done()
    since = (NOW - timedelta(days=7)).isoformat()
    assert await _texts(client, headers, sandbox, updated_since=since) == ["fraîche"]


async def test_cursor_walks_the_whole_reservoir_without_holes(
    client: AsyncClient, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    keywords = [await build.keyword(f"mot {index:02d}") for index in range(7)]
    await build.list("L", *keywords)
    await build.done()

    seen: list[str] = []
    cursor: str | None = None
    for _ in range(10):
        params: dict[str, Any] = {"limit": 3}
        if cursor is not None:
            params["cursor"] = cursor
        page = (await _keywords(client, headers, sandbox, **params)).json()
        seen.extend(row["keyword"] for row in page["items"])
        cursor = page["next_cursor"]
        if cursor is None:
            break
    assert cursor is None
    assert len(seen) == len(set(seen)) == 7
    # L'ordre est celui des identifiants UUIDv7 décroissants : le plus récent d'abord.
    assert seen == [f"mot {index:02d}" for index in reversed(range(7))]


async def test_broken_cursor_is_422(
    client: AsyncClient, headers: Headers, sandbox: Fixture
) -> None:
    assert (await _keywords(client, headers, sandbox, cursor="!!")).status_code == 422


async def test_another_org_sees_an_empty_reservoir(
    client: AsyncClient, auth_headers: Any, headers: Headers, sandbox: Fixture, build: Builder
) -> None:
    keyword = await build.keyword("secret")
    await build.list("L", keyword)
    await build.done()
    stranger = auth_headers(f"org_{uuid.uuid4().hex}")
    response = await client.get(
        "/keywords", headers=stranger, params={"site_target_id": str(sandbox.target_id)}
    )
    # La cible appartient à une autre organisation : elle n'existe pas (404, jamais 403).
    assert response.status_code == 404


# --- plan d'exécution -------------------------------------------------------------------------


@pytest_asyncio.fixture
async def bulk(db: AsyncSession) -> AsyncIterator[tuple[str, uuid.UUID]]:
    """Assez de volume pour que le planificateur ait un choix à faire (≈3 000 mots-clés).

    Sur une poignée de lignes, Postgres parcourt tout séquentiellement et le test de plan ne
    prouverait rien. `ANALYZE` est indispensable : sans statistiques, le planificateur travaille
    sur des estimations par défaut et le verdict serait un artefact.
    """
    org_id = f"org_{uuid.uuid4().hex}"
    site = Site(org_id=org_id, domain=f"{uuid.uuid4().hex}.example", name="Bulk")
    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.flush()
    keyword_list = KeywordList(org_id=org_id, site_target_id=target.id, name="Bulk")
    db.add(keyword_list)
    await db.flush()

    ids = list(
        (
            await db.scalars(
                Keyword.__table__.insert()
                .values(
                    [
                        {"org_id": org_id, "locale": LOCALE, "keyword": f"bulk {index:05d}"}
                        for index in range(3000)
                    ]
                )
                .returning(Keyword.id)
            )
        ).all()
    )
    await db.execute(
        KeywordListItem.__table__.insert(),
        [{"list_id": keyword_list.id, "keyword_id": kid, "org_id": org_id} for kid in ids],
    )
    await db.execute(
        KeywordMetric.__table__.insert(),
        [
            {
                "org_id": org_id,
                "keyword_id": kid,
                "source": source,
                "fetched_at": NOW,
                "volume": rank * 10,
                "raw": {},
            }
            for rank, kid in enumerate(ids)
            for source in ("semrush_magic", "generic_csv")
        ],
    )
    await db.commit()
    await db.execute(
        text("ANALYZE keywords, keyword_metrics, keyword_list_items, keyword_page_map")
    )
    await db.commit()
    yield org_id, target.id
    await db.execute(delete(Site).where(Site.org_id == org_id))
    await db.execute(delete(Keyword).where(Keyword.org_id == org_id))
    await db.commit()


async def test_reservoir_plan_inlines_the_function_and_never_scans_all_metrics(
    db: AsyncSession, bulk: tuple[str, uuid.UUID]
) -> None:
    org_id, target_id = bulk
    sql, names = service.build_query(
        service.CONSOLIDATED, service.Filters(volume_min=1000), cursor=None
    )
    plan = "\n".join(
        (
            await db.execute(
                text(f"EXPLAIN (ANALYZE, BUFFERS, COSTS OFF) {sql}"),
                {
                    "org": org_id,
                    "target": target_id,
                    "limit": 51,
                    "preferred": None,
                    "view": service.CONSOLIDATED,
                    **{name: 1000 for name in names},
                },
            )
        )
        .scalars()
        .all()
    )
    # La fonction est **inlinée** : un `Function Scan` signifierait que Postgres l'exécute
    # comme une boîte noire, sans pouvoir y pousser le filtre sur `keyword_id` (§3.3).
    assert "Function Scan" not in plan, plan
    # `keyword_metrics` n'est jamais parcourue au complet : chaque accès passe par le btree
    # `uq_keyword_metrics_keyword_source_fetched`, dont `keyword_id` est la colonne de tête.
    assert "Seq Scan on keyword_metrics" not in plan, plan


async def _plan(db: AsyncSession, sql: str, params: dict[str, Any]) -> str:
    rows = await db.execute(text(f"EXPLAIN (ANALYZE, BUFFERS, COSTS OFF) {sql}"), params)
    return "\n".join(rows.scalars().all())


async def test_search_plan_uses_the_gin_index_on_the_tsvector(
    db: AsyncSession, bulk: tuple[str, uuid.UUID]
) -> None:
    """La recherche entre par `ix_keywords_tsv`, jamais par un balayage de `keywords`.

    C'est l'assertion qui manquait : le correctif T6 remplace un prédicat qui *pouvait* se
    résoudre en trigramme par une tsquery à préfixe, et une tsquery à préfixe mal indexée
    resterait juste en devenant un balayage complet du réservoir.
    """
    org_id, target_id = bulk
    sql, _ = service.build_query(service.CONSOLIDATED, service.Filters(q="bulk 00042"), cursor=None)
    plan = await _plan(
        db,
        sql,
        {
            "org": org_id,
            "target": target_id,
            "limit": 51,
            "offset": 0,
            "preferred": None,
            "view": service.CONSOLIDATED,
            **service.search_params("bulk 00042"),
        },
    )
    assert "ix_keywords_tsv" in plan, plan
    assert "Seq Scan on keywords" not in plan, plan


async def test_suggest_plan_ranks_before_consolidating(
    db: AsyncSession, bulk: tuple[str, uuid.UUID]
) -> None:
    """Le classement et la coupe à dix passent avant la consolidation, pas après.

    Un `Seq Scan on keywords` ou une consolidation évaluée sur tous les candidats rendrait la
    complétion inutilisable à la frappe — c'est le point unique de latence de l'endpoint.
    """
    org_id, target_id = bulk
    plan = await _plan(
        db,
        service._SUGGEST_SQL.format(
            match=service._TS_MATCH,
            order=f"{service._TIER} DESC, {service._RANK} DESC, k.id DESC",
        ),
        {
            "org": org_id,
            "target": target_id,
            "limit": service.SUGGEST_LIMIT,
            "preferred": None,
            **service.search_params("bulk 000"),
        },
    )
    assert "ix_keywords_tsv" in plan, plan
    assert "Seq Scan on keywords" not in plan, plan
    assert "Seq Scan on keyword_metrics" not in plan, plan