gendesign/tradein-mvp/backend/data/sql/230_house_merge_log.sql

-- 230_house_merge_log.sql
-- Журнал слияний домов + обратная операция (#2690).
--
-- WHY:
--   `house_dedup_merge` — НЕ спящая идея, а живой деструктивный проход: расписание
--   `house_dedup_merge` на проде enabled=true, dry_run=false, такт 7 дней. Шесть прогонов
--   с 2026-06-27 уже удалили 119 строк `houses` (счётчики losers_deleted в scrape_runs:
--   2/39/31/9/6/32). Единственным следом «кто в кого» была строка `logger.info` в контейнере,
--   а логи ротируются быстрее суток. То есть **уже сегодня** нельзя назвать, какой дом в какой
--   свернули 1 августа, — не говоря о том, чтобы вернуть.
--
--   Пока этого журнала нет, любой разговор о расширении ключа схлопывания (#2690, #1772)
--   ведётся без права на ошибку: единственный откат — restore всей БД на момент до прогона,
--   т.е. выброс недели сбора. Журнал снимает это условие: слияние становится обратимым,
--   и вопрос о ключе можно пересматривать, а не решать «навсегда».
--
--   Правку НЕ следует читать как одобрение текущего ключа/победителя/гео-стража. Она к ним
--   НЕЙТРАЛЬНА: ни ключ, ни правило выбора победителя, ни гео-страж здесь не меняются.
--   Меняется только одно — теперь есть что откатить.
--
-- WHAT (одна строка = один проигравший дом):
--   merge_pass / cluster_key / geo_guard / distance_m — ОСНОВАНИЕ слияния. Это не косметика:
--       ровно этих полей не хватило в #2690, чтобы ответить на вопрос «сколько слияний прошло
--       на расстояниях, которые гео-страж заблокировал бы» по ДАННЫМ, а не по ревью. distance_m
--       пишется всегда, даже когда страж для прохода выключен (fias-проход) — тогда он и есть
--       единственная запись о том, насколько далеко разъехались объединённые дома.
--   loser_row     — ПОЛНЫЙ jsonb-снимок удаляемой строки (`to_jsonb(h.*)`, все 86 колонок).
--       Ссылка на удалённую строку бесполезна, поэтому хранится содержимое. Снимок целиком,
--       а не список полей: проверено, что `jsonb_populate_record(NULL::houses, loser_row)`
--       восстанавливает строку побайтово, включая PostGIS-geom (to_jsonb отдаёт её GeoJSON'ом,
--       populate_record разбирает обратно входной функцией типа). Побочная выгода: новая
--       колонка в `houses` попадает в снимок и в откат САМА, без правки этой миграции.
--   keeper_before — снимок ПОБЕДИТЕЛЯ до переноса метаданных. Нужен, потому что слияние не
--       только удаляет проигравшего: `_CARRY_OVER_IDENTITY_SQL` дозаполняет победителю NULL-поля
--       идентичности (fias/кадастр/ГАР/DaData) значениями проигравшего. Без этого снимка откат
--       вернул бы дом, но оставил бы его ФИАС на победителе — и следующий же fias-проход слил
--       бы их обратно.
--   children_repointed — {"таблица.колонка": [id, ...]}. Дочерние строки ПЕРЕЖИЛИ слияние,
--       у них сменилась только ссылка, поэтому хранятся id, а не содержимое (иначе одни
--       listings с их raw-payload'ом дали бы ~7 КБ на строку вместо ~8 байт на id).
--   children_deleted   — {"таблица": [{строка целиком}, ...]}. Дочерние строки, которые проход
--       УДАЛИЛ из-за коллизии по UNIQUE. Их содержимое уничтожено, id недостаточно — только
--       полный снимок. Таких таблиц шесть (см. _STEPS), строки мелкие.
--   batch_id — один вызов merge_duplicate_houses() (оба прохода). Единица отката.
--   run_id / initiator — кто инициировал: scrape_runs.id для расписания, NULL для ручного.
--
-- НАМЕРЕННО БЕЗ ВНЕШНИХ КЛЮЧЕЙ на houses(id) и scrape_runs(id):
--   журнал обязан ПЕРЕЖИВАТЬ строки, которые описывает. loser_id указывает на заведомо
--   удалённый дом. keeper_id — на дом, который сам может быть слит следующим прогоном; FK
--   с CASCADE стёр бы историю ровно тогда, когда она нужнее всего, а FK без CASCADE
--   заблокировал бы слияние. То же с run_id: чистка scrape_runs не должна трогать журнал.
--
-- ОБЪЁМ (замерено на проде 2026-08-06):
--   9 571 дом, средняя строка houses в jsonb 2 581 Б. Строка журнала ≈ loser_row 2.5 КБ +
--   keeper_before 2.5 КБ + списки id (в среднем 27.9 дочерних строк на дом × ~8 Б) ≈ 5.3 КБ.
--   Наблюдаемый темп — 20 слияний в неделю (119 за 6 прогонов) ≈ 106 КБ/нед ≈ 5.5 МБ/год.
--   Ближайший прогон (замер тем же выражением, что и код): 93 проигравших ≈ 0.5 МБ.
--   Абсолютный потолок, если схлопнуть вообще все дома: 9 571 × 5.3 КБ ≈ 50 МБ против 23 МБ
--   самой таблицы houses.
--
-- RETENTION: НЕ НУЖЕН, сознательно. Потолок роста — двузначные мегабайты, то есть дешевле
--   любой процедуры чистки; а журнал слияний — это ровно то, что удалять не хочется: его
--   ценность в том, что он отвечает на вопрос «что было год назад», когда логов давно нет.
--   Если объём когда-нибудь станет проблемой, удалять надо не строки, а тяжёлые снимки
--   (loser_row/keeper_before → NULL) у записей старше N лет, сохранив соответствие
--   loser→keeper: оно весит байты и именно оно нужно дольше всего.
--
-- Dependencies: 002_core_tables.sql (houses), 135_scrape_schedules_seed_house_dedup_merge.sql
-- Пишется в ТОЙ ЖЕ транзакции, что и слияние (см. house_dedup_merge._run_merge_pass) —
-- разрыв «слияние прошло, запись не легла» невозможен по построению; dry_run откатывает и то,
-- и другое вместе.

BEGIN;

CREATE TABLE IF NOT EXISTS house_merge_log (
    id                 bigserial PRIMARY KEY,
    merged_at          timestamptz NOT NULL DEFAULT now(),
    batch_id           uuid        NOT NULL,
    run_id             bigint,
    initiator          text        NOT NULL,
    merge_pass         text        NOT NULL,
    cluster_key        text        NOT NULL,
    geo_guard          boolean     NOT NULL,
    distance_m         double precision,
    norm_address       text,
    loser_id           bigint      NOT NULL,
    keeper_id          bigint      NOT NULL,
    loser_row          jsonb       NOT NULL,
    keeper_before      jsonb       NOT NULL,
    children_repointed jsonb       NOT NULL DEFAULT '{}'::jsonb,
    children_deleted   jsonb       NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX IF NOT EXISTS idx_house_merge_log_loser  ON house_merge_log (loser_id);
CREATE INDEX IF NOT EXISTS idx_house_merge_log_keeper ON house_merge_log (keeper_id);
CREATE INDEX IF NOT EXISTS idx_house_merge_log_batch  ON house_merge_log (batch_id);

COMMENT ON TABLE house_merge_log IS
    'Журнал слияний домов (#2690): одна строка = один проигравший дом, удалённый проходом '
    'house_dedup_merge. Пишется в ТОЙ ЖЕ транзакции, что и слияние. Содержит полный снимок '
    'удалённой строки и перечень перенесённых/удалённых дочерних строк — достаточно, чтобы '
    'назвать поимённо, что во что свернули, и вернуть обратно (house_merge_undo). Намеренно '
    'БЕЗ FK на houses/scrape_runs: журнал переживает строки, которые описывает. Retention нет.';

COMMENT ON COLUMN house_merge_log.cluster_key IS
    'ЗНАЧЕНИЕ ключа, по которому дома попали в один кластер («addr:вайнера66» / «fias:<uuid>»), '
    'а не имя ключа — по нему видно, какое именно совпадение сработало.';
COMMENT ON COLUMN house_merge_log.geo_guard IS
    'Был ли для этого прохода включён гео-страж 250 м. false = слияние разрешено БЕЗ проверки '
    'близости; вместе с distance_m это и есть аудит основания (#2690).';
COMMENT ON COLUMN house_merge_log.distance_m IS
    'ST_DistanceSphere между победителем и проигравшим на момент слияния; NULL = у одной из '
    'сторон не было geom. Пишется ВСЕГДА, в том числе когда гео-страж выключен.';
COMMENT ON COLUMN house_merge_log.loser_row IS
    'to_jsonb() удалённой строки houses целиком. Восстановление: '
    'INSERT INTO houses SELECT r.* FROM jsonb_populate_record(NULL::houses, loser_row) r.';
COMMENT ON COLUMN house_merge_log.keeper_before IS
    'Снимок победителя ДО переноса метаданных с проигравшего (COALESCE-дозаполнение полей '
    'идентичности). Без него откат вернул бы дом, но оставил его ФИАС/кадастр на победителе.';
COMMENT ON COLUMN house_merge_log.children_repointed IS
    '{"таблица.колонка": [id, ...]} — дочерние строки, у которых слияние сменило ссылку '
    'loser→keeper. Строки целы, поэтому хранятся id: откат возвращает ссылку обратно.';
COMMENT ON COLUMN house_merge_log.children_deleted IS
    '{"таблица": [{строка целиком}, ...]} — дочерние строки, УДАЛЁННЫЕ проходом из-за коллизии '
    'по UNIQUE с победителем. Содержимое уничтожено, поэтому хранится снимок, а не id.';

-- ── Обратная операция ────────────────────────────────────────────────────────
--
-- Откат одного батча (или его части) по журналу. Транзакционен: вызывающий сам решает
-- COMMIT/ROLLBACK, увидев отчёт. Возвращает СТРОКУ НА КАЖДУЮ запись журнала со статусом —
-- в том числе «не смог», потому что молчаливо-успешный откат хуже отсутствующего.
--
-- Порядок внутри одной записи важен: сначала воскресить дом (на него ссылаются дети), потом
-- вернуть ссылки детей, потом вернуть удалённых детей, потом снять перенос метаданных с
-- победителя. Записи батча обходятся в обратном порядке (id DESC) — если дом A слили в B,
-- а B потом в C, разматывать надо с конца.
--
-- ИЗВЕСТНЫЕ ГРАНИЦЫ (сознательные, отражены в статусе):
--   * дочерняя строка, удалённая по коллизии, может не вернуться: место в UNIQUE-ключе занято
--     строкой победителя. ON CONFLICT DO NOTHING + счётчик в статусе, а не тихая потеря;
--   * backfill-строки house_sources/house_address_aliases, которые проход дописал победителю,
--     НЕ удаляются: они собраны из собственных полей победителя и остались бы верны и без
--     слияния;
--   * если id проигравшего уже занят — запись пропускается со статусом, откат не гадает.
CREATE OR REPLACE FUNCTION house_merge_undo(
    p_batch       uuid,
    p_only_losers bigint[] DEFAULT NULL
)
RETURNS TABLE (
    out_log_id   bigint,
    out_loser_id bigint,
    out_keeper_id bigint,
    out_status   text
)
LANGUAGE plpgsql
AS $$
DECLARE
    rec           record;
    v_table       text;
    v_column      text;
    v_ids         bigint[];
    v_rows        jsonb;
    v_field       text;
    v_repointed   int;
    v_restored    int;
    v_lost        int;
    v_n           int;
    -- Список полей ДОЛЖЕН совпадать с SET в house_dedup_merge._CARRY_OVER_IDENTITY_SQL;
    -- за расхождением следит тест test_undo_carryover_fields_match_merge_carryover.
    c_carry_fields constant text[] := ARRAY[
        'house_fias_id', 'cadastral_number', 'gar_house_guid', 'gar_flat_count',
        'gar_matched_at', 'gar_match_method', 'dadata_qc_geo', 'dadata_qc_house',
        'dadata_enriched_at'
    ];
BEGIN
    FOR rec IN
        SELECT *
        FROM house_merge_log l
        WHERE l.batch_id = p_batch
          AND (p_only_losers IS NULL OR l.loser_id = ANY (p_only_losers))
        ORDER BY l.id DESC
    LOOP
        out_log_id := rec.id;
        out_loser_id := rec.loser_id;
        out_keeper_id := rec.keeper_id;

        IF EXISTS (SELECT 1 FROM houses h WHERE h.id = rec.loser_id) THEN
            out_status := 'skipped: houses.id ' || rec.loser_id || ' занят — уже откачено?';
            RETURN NEXT;
            CONTINUE;
        END IF;

        -- 1. Воскресить проигравшего целиком из снимка (все колонки, включая geom).
        INSERT INTO houses
        SELECT r.* FROM jsonb_populate_record(NULL::houses, rec.loser_row) r;

        -- 2. Вернуть ссылки уцелевших детей. Условие «сейчас указывает на победителя»
        --    защищает от затирания строк, которые после слияния перепривязали чем-то ещё.
        v_repointed := 0;
        FOR v_table, v_column, v_ids IN
            SELECT split_part(e.key, '.', 1),
                   split_part(e.key, '.', 2),
                   ARRAY(SELECT jsonb_array_elements_text(e.value)::bigint)
            FROM jsonb_each(rec.children_repointed) AS e
        LOOP
            EXECUTE format(
                'UPDATE %I SET %I = $1 WHERE id = ANY ($2) AND %I = $3',
                v_table, v_column, v_column
            ) USING rec.loser_id, v_ids, rec.keeper_id;
            GET DIAGNOSTICS v_n = ROW_COUNT;
            v_repointed := v_repointed + v_n;
        END LOOP;

        -- 3. Вернуть детей, удалённых по коллизии UNIQUE. Место могло остаться занятым
        --    строкой победителя — тогда DO NOTHING, и это попадёт в отчёт как «не вернулось».
        v_restored := 0;
        v_lost := 0;
        FOR v_table, v_rows IN
            SELECT e.key, e.value FROM jsonb_each(rec.children_deleted) AS e
        LOOP
            EXECUTE format(
                'INSERT INTO %I SELECT r.* FROM jsonb_array_elements($1) AS el, '
                'LATERAL jsonb_populate_record(NULL::%I, el) r ON CONFLICT DO NOTHING',
                v_table, v_table
            ) USING v_rows;
            GET DIAGNOSTICS v_n = ROW_COUNT;
            v_restored := v_restored + v_n;
            v_lost := v_lost + (jsonb_array_length(v_rows) - v_n);
        END LOOP;

        -- 4. Снять перенос метаданных с победителя. Только там, где до слияния было NULL И
        --    текущее значение всё ещё РОВНО то, что принёс этот проигравший: если поле успел
        --    заполнить загрузчик (или донором был другой проигравший кластера) — не трогаем.
        --    Сравнение в jsonb-пространстве, чтобы один цикл покрыл text/int/timestamptz.
        FOREACH v_field IN ARRAY c_carry_fields LOOP
            IF rec.keeper_before ->> v_field IS NULL THEN
                EXECUTE format(
                    'UPDATE houses SET %I = NULL WHERE id = $1 AND to_jsonb(%I) = $2',
                    v_field, v_field
                ) USING rec.keeper_id, rec.loser_row -> v_field;
            END IF;
        END LOOP;

        out_status := format(
            'restored: дом %s вернулся, ссылок возвращено %s, дочерних строк восстановлено %s'
            || CASE WHEN v_lost > 0 THEN ', НЕ ВЕРНУЛОСЬ ' || v_lost || ' (место занято)'
                    ELSE '' END,
            rec.loser_id, v_repointed, v_restored
        );
        RETURN NEXT;
    END LOOP;
END;
$$;

COMMENT ON FUNCTION house_merge_undo(uuid, bigint[]) IS
    'Откат слияния домов по журналу house_merge_log (#2690). Аргументы: batch_id (единица '
    'отката = один вызов merge_duplicate_houses) и опциональный список loser_id для частичного '
    'отката. Возвращает строку-статус на КАЖДУЮ запись журнала, включая неудачные. '
    'Транзакции не открывает и не закрывает — вызывающий смотрит отчёт и решает COMMIT/ROLLBACK: '
    '  BEGIN; SELECT * FROM house_merge_undo(''<batch_id>''); -- прочитать статусы -- COMMIT;';

COMMIT;