-- 286_dryrun.sql — прогон миграции 286 на проде БЕЗ записи (#3376).
--
-- ЗАЧЕМ. Миграция 286 удаляет строки из offer_price_history. Прежде чем этому
-- уехать на деплой, надо увидеть на БОЕВЫХ данных: сколько строк она заберёт по
-- источникам, сколько соседей пересчитает, сколько процентов обнулит — и что
-- ВТОРОЙ прогон подряд не делает уже ничего (идемпотентность).
--
-- КАК ЗАПУСКАТЬ (ничего не остаётся, в конце ROLLBACK):
--   ssh poincare
--   docker exec -i tradein-postgres psql -U <user> -d tradein -v ON_ERROR_STOP=1 \
--       -f - < 286_dryrun.sql
--   (или скопировать файл в контейнер и psql -f). Смотреть строки NOTICE.
--
-- ЧЕМ ОТЛИЧАЕТСЯ ОТ 286 — ровно тремя вещами, всё остальное скопировано дословно:
--   1. вся работа обёрнута в FOR pass_no IN 1..2 — два прохода подряд в ОДНОЙ
--      транзакции; второй обязан дать deleted=0 recomputed=0 nulled=0;
--   2. порог 200 даёт WARNING, а не EXCEPTION: на dry-run важнее увидеть число и
--      добежать до второго прохода, чем оборвать транзакцию;
--   3. ROLLBACK вместо COMMIT.
-- ФАЙЛ РАЗОВЫЙ и живёт вне data/sql (автоприменение его не подхватывает). Правишь
-- 286 — правь и здесь, иначе сверять будет нечего.
BEGIN;
SET LOCAL lock_timeout = '5s';

-- ── ПРЕДЗАМЕР ОДНИМ SELECT (для сверки с числами в шапке 286) ────────────────
-- Правило внутри серии пошаговое (база = предыдущая ОСТАВЛЕННАЯ точка) и одним
-- SELECT не выражается — его считает DO-блок ниже. А эти два выражаются.

-- (а) кандидаты правила ПЕРВОЙ точки по СЫРЫМ строкам, до удалений фазы 1.
--     Ожидание из шапки 286: 4, все domklik (yandex 0 — их свидетелем была бы
--     listings.price_rub, а откат на неё снят).
SELECT 'предзамер: первая точка (сырые строки)' AS metric, k.source, count(*)
  FROM (
      SELECT oph.source,
             oph.price_rub,
             row_number()           OVER w AS rn,
             lead(oph.price_rub)    OVER w AS second_price,
             lead(oph.price_rub, 2) OVER w AS third_price
        FROM offer_price_history oph
       WHERE oph.change_time <> oph.recorded_at
      WINDOW w AS (PARTITION BY oph.listing_id ORDER BY oph.change_time, oph.id)
  ) k
 WHERE k.rn = 1
   AND k.price_rub > 0
   AND k.second_price > 0
   AND (k.second_price / k.price_rub BETWEEN 9.5 AND 10.5
        OR k.price_rub / k.second_price BETWEEN 9.5 AND 10.5)
   AND k.third_price > 0
   AND k.third_price / k.second_price BETWEEN 0.9 AND 1.1
 GROUP BY k.source
 ORDER BY k.source;

-- (б) сколько строк заберёт финальный шаг |diff_percent| > 100 → NULL.
SELECT 'предзамер: |diff_percent| > 100' AS metric, source, count(*)
  FROM offer_price_history
 WHERE change_time <> recorded_at
   AND abs(diff_percent) > 100
 GROUP BY source
 ORDER BY source;

-- ── ДВА ПРОХОДА ЛОГИКИ 286 ───────────────────────────────────────────────────
CREATE TEMP TABLE oph_decimal_slips (
    id         bigint PRIMARY KEY,
    listing_id bigint NOT NULL,
    next_id    bigint,
    source     text NOT NULL
) ON COMMIT DROP;

DO $$
DECLARE
    pass_no      int;
    r            record;
    cur_listing  bigint;
    base         numeric;
    witness      numeric;
    confirmed    boolean;
    to_delete    bigint;
    per_source   text;
    deleted_rows bigint;
    updated_rows bigint;
    nulled_rows  bigint;
    nulled_src   text;
BEGIN
FOR pass_no IN 1..2 LOOP
    DELETE FROM oph_decimal_slips;
    cur_listing := NULL;
    base := NULL;

    -- ФАЗА 1 — правило внутри серии, пошагово как в гейте.
    FOR r IN
        SELECT oph.id,
               oph.listing_id,
               oph.source,
               oph.price_rub,
               lead(oph.price_rub) OVER w AS next_price,
               lead(oph.id)        OVER w AS next_id,
               l.price_rub                AS listing_price
          FROM offer_price_history oph
          LEFT JOIN listings l ON l.id = oph.listing_id
         WHERE oph.change_time <> oph.recorded_at
           AND oph.listing_id IN (
                   SELECT listing_id
                     FROM offer_price_history
                    WHERE change_time <> recorded_at
                    GROUP BY listing_id
                   HAVING count(*) >= 2
               )
        WINDOW w AS (PARTITION BY oph.listing_id ORDER BY oph.change_time, oph.id)
         ORDER BY oph.listing_id, oph.change_time, oph.id
    LOOP
        IF cur_listing IS DISTINCT FROM r.listing_id THEN
            cur_listing := r.listing_id;
            base := NULL;
        END IF;

        witness := COALESCE(r.next_price, r.listing_price);

        IF base IS NOT NULL
           AND base > 0 AND r.price_rub > 0 AND witness > 0
           AND (r.price_rub / base BETWEEN 9.5 AND 10.5
                OR base / r.price_rub BETWEEN 9.5 AND 10.5)
           AND witness / base BETWEEN 0.9 AND 1.1
        THEN
            SELECT EXISTS (
                       SELECT 1
                         FROM offer_price_history t
                        WHERE t.listing_id = r.listing_id
                          AND t.change_time = t.recorded_at
                          AND t.price_rub = r.price_rub
                   )
              INTO confirmed;

            IF NOT confirmed THEN
                INSERT INTO oph_decimal_slips (id, listing_id, next_id, source)
                VALUES (r.id, r.listing_id, r.next_id, r.source);
                CONTINUE;
            END IF;
        END IF;

        IF r.price_rub > 0 THEN
            base := r.price_rub;
        END IF;
    END LOOP;

    -- ФАЗА 2 — правило первой точки, по оставшимся строкам.
    INSERT INTO oph_decimal_slips (id, listing_id, next_id, source)
    WITH kept AS (
        SELECT oph.id,
               oph.listing_id,
               oph.source,
               oph.price_rub,
               row_number()           OVER w AS rn,
               lead(oph.id)           OVER w AS second_id,
               lead(oph.price_rub)    OVER w AS second_price,
               lead(oph.price_rub, 2) OVER w AS third_price
          FROM offer_price_history oph
         WHERE oph.change_time <> oph.recorded_at
           AND NOT EXISTS (SELECT 1 FROM oph_decimal_slips s WHERE s.id = oph.id)
        WINDOW w AS (PARTITION BY oph.listing_id ORDER BY oph.change_time, oph.id)
    )
    SELECT k.id, k.listing_id, k.second_id, k.source
      FROM kept k
     WHERE k.rn = 1
       AND k.price_rub > 0
       AND k.second_price > 0
       AND (k.second_price / k.price_rub BETWEEN 9.5 AND 10.5
            OR k.price_rub / k.second_price BETWEEN 9.5 AND 10.5)
       -- Свидетель — только третья точка истории (см. шапку 286).
       AND k.third_price > 0
       AND k.third_price / k.second_price BETWEEN 0.9 AND 1.1
       AND NOT EXISTS (
               SELECT 1
                 FROM offer_price_history t
                WHERE t.listing_id = k.listing_id
                  AND t.change_time = t.recorded_at
                  AND t.price_rub = k.price_rub
           );

    SELECT count(*) INTO to_delete FROM oph_decimal_slips;

    SELECT string_agg(source || '=' || cnt, ', ' ORDER BY source)
      INTO per_source
      FROM (
          SELECT source, count(*) AS cnt
            FROM oph_decimal_slips
           GROUP BY source
      ) s;

    -- Отличие 2 от 286: здесь WARNING, а не EXCEPTION.
    IF to_delete > 200 THEN
        RAISE WARNING 'pass%: кандидатов % > порога 200 — на проде 286 здесь УПАЛА БЫ',
            pass_no, to_delete;
    END IF;

    DELETE FROM offer_price_history oph
     USING oph_decimal_slips s
     WHERE oph.id = s.id;
    GET DIAGNOSTICS deleted_rows = ROW_COUNT;

    WITH neighbours AS (
        SELECT id,
               price_rub,
               lag(price_rub) OVER (PARTITION BY listing_id ORDER BY change_time, id)
                   AS prev_price
          FROM offer_price_history
         WHERE change_time <> recorded_at
           AND listing_id IN (SELECT DISTINCT listing_id FROM oph_decimal_slips)
    ),
    recomputed AS (
        SELECT n.id,
               CASE
                   WHEN n.prev_price > 0
                    AND abs((n.price_rub - n.prev_price) / n.prev_price * 100) <= 100
                   THEN round((n.price_rub - n.prev_price) / n.prev_price * 100, 2)
               END AS new_diff
          FROM neighbours n
         WHERE n.id IN (SELECT next_id FROM oph_decimal_slips WHERE next_id IS NOT NULL)
    )
    UPDATE offer_price_history oph
       SET diff_percent = rc.new_diff
      FROM recomputed rc
     WHERE oph.id = rc.id
       AND oph.diff_percent IS DISTINCT FROM rc.new_diff;
    GET DIAGNOSTICS updated_rows = ROW_COUNT;

    SELECT string_agg(source || '=' || cnt, ', ' ORDER BY source)
      INTO nulled_src
      FROM (
          SELECT source, count(*) AS cnt
            FROM offer_price_history
           WHERE change_time <> recorded_at
             AND abs(diff_percent) > 100
           GROUP BY source
      ) s;

    UPDATE offer_price_history
       SET diff_percent = NULL
     WHERE change_time <> recorded_at
       AND abs(diff_percent) > 100;
    GET DIAGNOSTICS nulled_rows = ROW_COUNT;

    RAISE NOTICE 'pass%: deleted=% (%) recomputed=% nulled=% (%)',
        pass_no,
        deleted_rows, COALESCE(per_source, 'ни одного'),
        updated_rows,
        nulled_rows, COALESCE(nulled_src, 'ни одной');
END LOOP;
END $$;

ROLLBACK;