"""consolidation multi-sources

Revision ID: 542f48d3235f
Revises: 333652ce4b5e
Create Date: 2026-09-01 15:53:32.849525

"""

from collections.abc import Sequence

from alembic import op

# revision identifiers, used by Alembic.
revision: str = "542f48d3235f"
down_revision: str | Sequence[str] | None = "333652ce4b5e"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


# --- consolidation multi-sources : fonction + vue `keyword_metrics_latest` -------------
#
# **Aucune hiérarchie de sources** (décision A1, spec 2026-09-01 §2) : « personne ne prend
# le dessus sur personne ». La fonction n'élit plus une valeur par champ — elle rend, par
# champ, **la dernière observation de chaque source**, côte à côte, en jsonb.
#
# Règle, par mot-clé et **par champ** :
#   1. une entrée par source qui a une valeur non nulle pour ce champ — la plus récente de
#      cette source (`DISTINCT ON (m.source)`) ; l'historique complet reste dans
#      `GET /keywords/{id}/metrics` ;
#   2. une source sans valeur pour un champ n'apparaît pas dans le tableau de ce champ ;
#   3. un champ sans aucune observation rend `[]`, **jamais `null`** (d'où le `coalesce` :
#      un `jsonb_agg` sans GROUP BY rend une ligne, NULL sur ensemble vide) ;
#   4. `preferred_source` **ordonne** et ne masque plus rien : elle passe en tête du tri du
#      `jsonb_agg`, jamais dans celui du `DISTINCT ON` — là, elle changerait quelle
#      observation est retenue pour une source, ce qui n'a aucun sens (spec §3.4).
# Ordre des entrées, contractuel : la préférée d'abord, puis `fetched_at DESC`, puis le nom
# de la source. Le web n'a rien à retrier.
#
# `fetched_at` passe par `to_char(… AT TIME ZONE 'UTC', …)` et **jamais** par le timestamptz
# nu : `jsonb_build_object` sérialise un timestamptz dans le fuseau de **session** Postgres
# (mesuré : `+07:00` avec `TimeZone = Asia/Bangkok`), et rien ne fixe `TimeZone` côté api.
# La chaîne rendue est donc en UTC quel que soit le serveur, et Pydantic la parse en
# datetime aware.
#
# Forme : un seul SELECT, sans CTE ni DISTINCT au niveau supérieur, sans agrégat dans la
# SELECT list de tête — c'est ce qui garde la fonction **inlinable** par le planificateur,
# qui pousse alors le filtre sur `keyword_id` jusque dans chaque latéral (mesuré : aucun
# `Function Scan`, six `Index Scan Backward using uq_keyword_metrics_keyword_source_fetched`,
# aucun `Seq Scan on keyword_metrics`). Le `DISTINCT ON` ordonne `m.source DESC,
# m.fetched_at DESC` — même sens sur les deux colonnes : un ordre mixte imposerait un
# `Incremental Sort` au-dessus de l'index. DISTINCT ON n'exige que le préfixe, pas le sens,
# et l'ordre final est de toute façon refait par le `jsonb_agg`.
#
# **Aucun index à ajouter** : le btree de `uq_keyword_metrics_keyword_source_fetched` porte
# déjà exactement `(keyword_id, source, fetched_at)` — cf. le commentaire de
# `app/keywords/models.py`, qui interdit explicitement d'en poser un second.


def _pick(column: str) -> str:
    """Le latéral d'un champ : la dernière observation de **chaque** source, en jsonb.

    L'alias est suffixé (`volume_pick`, jamais `volume`) : un alias de relation homonyme
    d'un paramètre de sortie de la fonction rendrait la référence ambiguë.
    """
    return f"""
    LEFT JOIN LATERAL (
        SELECT coalesce(
                   jsonb_agg(
                       jsonb_build_object(
                           'value', x.{column},
                           'source', x.source,
                           'fetched_at', to_char(x.fetched_at AT TIME ZONE 'UTC',
                                                 'YYYY-MM-DD"T"HH24:MI:SS.USOF:00'))
                       ORDER BY (x.source IS NOT DISTINCT FROM preferred_source) DESC,
                                x.fetched_at DESC,
                                x.source),
                   '[]'::jsonb) AS value
        FROM (
            SELECT DISTINCT ON (m.source) m.source, m.{column}, m.fetched_at
            FROM keyword_metrics m
            WHERE m.keyword_id = b.keyword_id AND m.{column} IS NOT NULL
            ORDER BY m.source DESC, m.fetched_at DESC
        ) x
    ) {column}_pick ON true"""


KEYWORD_METRICS_LATEST_FUNCTION = f"""
CREATE FUNCTION keyword_metrics_latest(preferred_source text DEFAULT NULL)
RETURNS TABLE (
    keyword_id uuid,
    org_id text,
    volume jsonb,
    kd jsonb,
    cpc jsonb,
    intent jsonb,
    serp_features jsonb,
    trend jsonb,
    last_updated timestamptz
)
LANGUAGE sql
STABLE
AS $$
    SELECT b.keyword_id,
           b.org_id,
           volume_pick.value,
           kd_pick.value,
           cpc_pick.value,
           intent_pick.value,
           serp_features_pick.value,
           trend_pick.value,
           b.last_updated
    FROM (
        SELECT m.keyword_id, min(m.org_id) AS org_id, max(m.fetched_at) AS last_updated
        FROM keyword_metrics m
        GROUP BY m.keyword_id
    ) b{_pick("volume")}{_pick("kd")}{_pick("cpc")}{_pick("intent")}{_pick("serp_features")}{
    _pick("trend")
}
$$
"""

# Confort de lecture (Matrice, requêtes ad hoc) : la consolidation par défaut, sans source
# préférée. Une vue et une fonction peuvent porter le même nom — catalogues distincts.
KEYWORD_METRICS_LATEST_VIEW = """
CREATE VIEW keyword_metrics_latest AS SELECT * FROM keyword_metrics_latest(NULL::text)
"""


# --- l'ancienne règle, recopiée mot pour mot pour le `downgrade` -----------------------
#
# `alembic/versions/` n'a pas d'`__init__.py` : une révision n'est pas importable depuis une
# autre, ni en CLI ni dans le sous-process de `tests/test_migrations.py`. Le corps de
# `333652ce4b5e` est donc dupliqué ici — c'est le prix d'un downgrade réellement rejouable.
# Ne jamais éditer `333652ce4b5e_keywords.py`, déjà appliquée.
_LEGACY_PRIORITY_VOLUME_CPC = """CASE m.source
                WHEN 'dataforseo' THEN 1
                WHEN 'semrush_positions' THEN 2
                WHEN 'semrush_magic' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""
_LEGACY_PRIORITY_KD = """CASE m.source
                WHEN 'semrush_magic' THEN 1
                WHEN 'semrush_positions' THEN 1
                WHEN 'dataforseo' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""
_LEGACY_PRIORITY_QUALITATIVE = """CASE m.source
                WHEN 'semrush_magic' THEN 1
                WHEN 'semrush_positions' THEN 1
                WHEN 'se_ranking' THEN 2
                WHEN 'generic_csv' THEN 99
                ELSE 50 END"""


def _legacy_pick(column: str, priority: str) -> str:
    return f"""
    LEFT JOIN LATERAL (
        SELECT m.{column} AS value, m.source, m.fetched_at
        FROM keyword_metrics m
        WHERE m.keyword_id = b.keyword_id AND m.{column} IS NOT NULL
        ORDER BY (m.source IS NOT DISTINCT FROM preferred_source) DESC,
                 m.fetched_at DESC,
                 {priority},
                 m.source,
                 m.id DESC
        LIMIT 1
    ) {column}_pick ON true"""


LEGACY_KEYWORD_METRICS_LATEST_FUNCTION = f"""
CREATE FUNCTION keyword_metrics_latest(preferred_source text DEFAULT NULL)
RETURNS TABLE (
    keyword_id uuid,
    org_id text,
    volume integer,
    volume_source text,
    volume_fetched_at timestamptz,
    kd numeric,
    kd_source text,
    kd_fetched_at timestamptz,
    cpc numeric,
    cpc_source text,
    cpc_fetched_at timestamptz,
    intent text,
    intent_source text,
    intent_fetched_at timestamptz,
    serp_features jsonb,
    serp_features_source text,
    serp_features_fetched_at timestamptz,
    trend smallint[],
    trend_source text,
    trend_fetched_at timestamptz,
    last_updated timestamptz
)
LANGUAGE sql
STABLE
AS $$
    SELECT b.keyword_id,
           b.org_id,
           volume_pick.value, volume_pick.source, volume_pick.fetched_at,
           kd_pick.value, kd_pick.source, kd_pick.fetched_at,
           cpc_pick.value, cpc_pick.source, cpc_pick.fetched_at,
           intent_pick.value, intent_pick.source, intent_pick.fetched_at,
           serp_features_pick.value, serp_features_pick.source, serp_features_pick.fetched_at,
           trend_pick.value, trend_pick.source, trend_pick.fetched_at,
           b.last_updated
    FROM (
        SELECT m.keyword_id, min(m.org_id) AS org_id, max(m.fetched_at) AS last_updated
        FROM keyword_metrics m
        GROUP BY m.keyword_id
    ) b{_legacy_pick("volume", _LEGACY_PRIORITY_VOLUME_CPC)}{
    _legacy_pick("kd", _LEGACY_PRIORITY_KD)
}{_legacy_pick("cpc", _LEGACY_PRIORITY_VOLUME_CPC)}{
    _legacy_pick("intent", _LEGACY_PRIORITY_QUALITATIVE)
}{_legacy_pick("serp_features", _LEGACY_PRIORITY_QUALITATIVE)}{
    _legacy_pick("trend", _LEGACY_PRIORITY_QUALITATIVE)
}
$$
"""


def upgrade() -> None:
    """Upgrade schema."""
    # Aucun DDL sur une table, aucune donnée touchée : la révision ne fait que remplacer la
    # règle de lecture. L'ordre des quatre temps n'est pas négociable — un `DROP FUNCTION`
    # en premier échoue (« cannot drop function … because other objects depend on it :
    # view keyword_metrics_latest depends on function »).
    op.execute("DROP VIEW IF EXISTS keyword_metrics_latest")
    op.execute("DROP FUNCTION IF EXISTS keyword_metrics_latest(text)")
    op.execute(KEYWORD_METRICS_LATEST_FUNCTION)
    op.execute(KEYWORD_METRICS_LATEST_VIEW)


def downgrade() -> None:
    """Downgrade schema."""
    # La **même** séquence en quatre temps : « l'inverse » au pied de la lettre (recréer la
    # vue avant la fonction) est impossible, la vue dépend de la fonction. Seul le corps
    # recréé change — celui de `333652ce4b5e`, recopié ci-dessus.
    op.execute("DROP VIEW IF EXISTS keyword_metrics_latest")
    op.execute("DROP FUNCTION IF EXISTS keyword_metrics_latest(text)")
    op.execute(LEGACY_KEYWORD_METRICS_LATEST_FUNCTION)
    op.execute(KEYWORD_METRICS_LATEST_VIEW)