-- 99c_power_supply_centers_dedup.sql
-- Issue #3322 — power_supply_centers раздут ×10: 4880 строк на 481 уникальный ЦП.
--
-- Причина. rosseti_wfs_loader брал external_id из feature['id'] WFS-ответа, а
-- GeoServer отдаёт СЕССИОННЫЙ fid (новый на каждый GetFeature) → ON CONFLICT
-- (source, external_id) не срабатывал ни разу, каждый weekly-прогон добавлял
-- полный набор ~488 фич заново. Починка разбора сама старые строки не убирает
-- (ON CONFLICT ничего не перезапишет, ключи не совпадут) → нужен этот бэкфилл.
--
-- Что делает файл:
--   (а) пересчитывает external_id по НОВОЙ формуле (см. ниже) для всех строк
--       source='rosseti_wfs';
--   (б) схлопывает копии: победитель группы — свежайший снапшот
--       (fetched_at DESC NULLS LAST, id DESC — DESC в PG это NULLS FIRST,
--       поэтому NULLS LAST задан ЯВНО);
--   (в) печатает числа: строк до / после, удалено, переключено на новый ключ.
--       Ожидание после прогона — ~481-488 строк (столько ЦП отдаёт источник).
--   (г) идемпотентен: на повторном прогоне ключи уже совпадают → 0 удалений,
--       0 обновлений, «до» = «после».
--
-- ФОРМУЛА КЛЮЧА (дублирует rosseti_wfs_loader._stable_external_id — менять только
-- синхронно, иначе следующий weekly-прогон вставит второй комплект строк):
--     seed = sc_name_norm || '|' || voltage_class || '|' || lon_e5 || '|' || lat_e5
--     external_id = 'h:' || left(hex(sha256(utf8(seed))), 16)
-- где lon_e5/lat_e5 — координата в единицах 1e-5 градуса (~1 м), округление
-- floor(|v|*1e5 + 0.5) со знаком — ДВОИЧНОЕ, ровно как в питоне; пустая строка,
-- если geom отсутствует. Целые, а не форматированный float — текстовое
-- представление double в питоне и в PG различается.
--
-- ПОЧЕМУ НЕ round(...::numeric): каст float8→numeric берёт кратчайшее десятичное
-- представление, и округление идёт по нему, а не по двоичному double. На реальных
-- координатах расходится в 0.19% случаев (замер: 761 из 400000), напр. 64.423605
-- → питон 6442360 (двоичное 6442360.499999999), numeric-путь 6442361. Каждое
-- расхождение = вечный дубль ЦП, который сам не зарастёт: миграция применяется
-- один раз (_schema_migrations). Поэтому в SQL считаем ТЕМ ЖЕ double: floor/abs/
-- sign над float8 — это IEEE754, бит в бит как math.floor в питоне.
-- sha256, а не sha1: sha256 встроен в PG16, sha1 потребовал бы pgcrypto.
--
-- Байт-в-байт совпадение с питоном держится на том, что SQL НИЧЕГО не нормализует
-- сам: sc_name_norm и voltage_class — уже готовые колонки, их записал тот же
-- normalize_sc_name / parse_voltage_class. Если normalize_sc_name когда-нибудь
-- изменится, старые sc_name_norm разъедутся с новыми ключами — тогда нужен
-- повторный прогон логики этого файла (он идемпотентен, ре-apply безопасен).
--
-- Порядок: миграция ПЕРЕД деплоем кода (schema-first) — новый код после неё
-- попадает ON CONFLICT-ом в уже схлопнутые строки.
--
-- Naming: deploy.yml применяет файлы по `ls -1 data/sql/*.sql | sort`;
-- '99c_' идёт после '99b_grant_quarter_price_index_fdw.sql' ('b' < 'c').

BEGIN;

SET LOCAL lock_timeout = '5s';

DO $$
DECLARE
    rows_before  bigint;
    names_before bigint;
    rows_after   bigint;
    names_after  bigint;
    deleted      bigint;
    rekeyed      bigint;
BEGIN
    SELECT count(*), count(DISTINCT sc_name_norm)
      INTO rows_before, names_before
      FROM power_supply_centers
     WHERE source = 'rosseti_wfs';

    CREATE TEMP TABLE psc_new_key ON COMMIT DROP AS
    SELECT
        id,
        fetched_at,
        'h:' || substring(
            encode(
                sha256(convert_to(
                    sc_name_norm
                    || '|' || coalesce(voltage_class, '')
                    || '|' || CASE WHEN geom IS NULL THEN ''
                                   ELSE (sign(ST_X(geom))
                                         * floor(abs(ST_X(geom)) * 100000 + 0.5))::bigint::text END
                    || '|' || CASE WHEN geom IS NULL THEN ''
                                   ELSE (sign(ST_Y(geom))
                                         * floor(abs(ST_Y(geom)) * 100000 + 0.5))::bigint::text END,
                    'UTF8'
                )),
                'hex'
            ) FROM 1 FOR 16
        ) AS new_key
      FROM power_supply_centers
     WHERE source = 'rosseti_wfs';

    -- (б) схлопывание: оставляем свежайший снапшот каждой группы.
    -- Резервы (reserve_mva и пр.) не теряются: rosseti/eesk-лоадеры пишут их
    -- UPDATE-ом по sc_name_norm, т.е. во ВСЕ копии сразу, победитель их несёт.
    WITH ranked AS (
        SELECT
            id,
            row_number() OVER (
                PARTITION BY new_key
                ORDER BY fetched_at DESC NULLS LAST, id DESC
            ) AS rn
          FROM psc_new_key
    )
    DELETE FROM power_supply_centers p
     USING ranked r
     WHERE p.id = r.id
       AND r.rn > 1;
    GET DIAGNOSTICS deleted = ROW_COUNT;

    -- (а) пересчёт ключа у выживших. После DELETE каждый new_key принадлежит
    -- ровно одной строке → UNIQUE (source, external_id) не нарушается.
    -- IS DISTINCT FROM даёт идемпотентность: второй прогон обновит 0 строк.
    UPDATE power_supply_centers p
       SET external_id = k.new_key
      FROM psc_new_key k
     WHERE p.id = k.id
       AND p.external_id IS DISTINCT FROM k.new_key;
    GET DIAGNOSTICS rekeyed = ROW_COUNT;

    SELECT count(*), count(DISTINCT sc_name_norm)
      INTO rows_after, names_after
      FROM power_supply_centers
     WHERE source = 'rosseti_wfs';

    RAISE NOTICE '#3322 power_supply_centers: было % строк / % имён -> стало % строк / % имён (удалено %, переключено на стабильный ключ %)',
        rows_before, names_before, rows_after, names_after, deleted, rekeyed;

    -- 700 — потолок здравого смысла: источник отдаёт ~488 ЦП по области.
    -- Превышение = формула ключа не схлопнула дубли. EXCEPTION, а не WARNING:
    -- иначе файл пометится applied навсегда, а дубли останутся. Откат всей
    -- транзакции ничего не теряет и оставляет миграцию непринятой до разбора.
    IF rows_after > 700 THEN
        RAISE EXCEPTION '#3322: после дедупа осталось % строк (ожидалось ~481-488) — формула ключа не схлопнула дубли, транзакция откачена', rows_after;
    END IF;
END $$;

COMMIT;