---

BEGIN READ ONLY;

SET LOCAL statement_timeout = '180s';

WITH
latest_death_info AS MATERIALIZED (
    SELECT DISTINCT ON (a.individual_id)
           a.individual_id,
           a.death_date
    FROM ourpension.application_death_info a
    WHERE a.accepted IS NOT NULL
      AND a.canceled IS NULL
      AND a.rejected IS NULL
    ORDER BY a.individual_id, a.version DESC
),

-- Общая контрольная выборка до исключения переводов.
sample_sharers AS MATERIALIZED (
    SELECT s.id
    FROM latest_death_info di
    JOIN ourpension.sharer s
      ON s.individual_id = di.individual_id
    JOIN ourpension.contract c
      ON c.id = s.contract_id
    WHERE c.service_type = 'OPS'
      AND EXISTS (
          SELECT 1
          FROM report.dia_param p
          WHERE di.death_date <= p.report_date
      )
    ORDER BY di.individual_id, s.id
    LIMIT 200
),

-- Отбор участников по исходной схеме.
old_ops_sharer AS MATERIALIZED (
    SELECT s.id,
           di.death_date
    FROM sample_sharers sample
    JOIN ourpension.sharer s
      ON s.id = sample.id
    JOIN ourpension.contract c
      ON c.id = s.contract_id
    JOIN report.dia_param p
      ON true
    JOIN LATERAL (
        SELECT a.death_date
        FROM ourpension.application_death_info a
        WHERE a.individual_id = s.individual_id
          AND a.accepted IS NOT NULL
          AND a.canceled IS NULL
          AND a.rejected IS NULL
        ORDER BY a.version DESC
        LIMIT 1
    ) di ON true
    LEFT JOIN LATERAL (
        SELECT d.sharer_id
        FROM back_office.ops_transfer_details d
        JOIN back_office.ops_transfer_register tr
          ON tr.id = d.ops_transfer_register_id
        JOIN back_office.operation o
          ON o.ops_transfer_register_id = tr.id
        WHERE d.sharer_id = s.id
          AND o.approved IS NOT NULL
          AND o.effective_date <= p.report_date
        LIMIT 1
    ) transfer ON true
    WHERE c.service_type = 'OPS'
      AND di.death_date <= p.report_date
      AND transfer.sharer_id IS NULL
),

-- Отбор участников по новой схеме.
new_ops_sharer AS MATERIALIZED (
    SELECT s.id,
           di.death_date
    FROM latest_death_info di
    JOIN report.dia_param p
      ON di.death_date <= p.report_date
    JOIN ourpension.sharer s
      ON s.individual_id = di.individual_id
    JOIN sample_sharers sample
      ON sample.id = s.id
    JOIN ourpension.contract c
      ON c.id = s.contract_id
    WHERE c.service_type = 'OPS'
      AND NOT EXISTS (
          SELECT 1
          FROM back_office.ops_transfer_details d
          JOIN back_office.ops_transfer_register tr
            ON tr.id = d.ops_transfer_register_id
          JOIN back_office.operation o
            ON o.ops_transfer_register_id = tr.id
          WHERE d.sharer_id = s.id
            AND o.approved IS NOT NULL
            AND o.effective_date <= p.report_date
      )
),

categories(mnemonics, category) AS (
    VALUES
        ('ОПС',             'ops'),
        ('ОПС ГВ',          'ops'),
        ('ОПС КП',          'ops'),
        ('Р ПП ОПС',        'ops'),
        ('ОПС ИД',          'ops_id'),
        ('ОПС ФР',          'ops_id'),
        ('Р ПП ОПС ИД',     'ops_id'),
        ('ДПС Л',           'dps_l'),
        ('ДПС Л ГВ',        'dps_l'),
        ('ДПС Л КП',        'dps_l'),
        ('Р ПП ДПС Л',      'dps_l'),
        ('ДПС Л ИД',        'dps_l_id'),
        ('ДПС Л ФР',        'dps_l_id'),
        ('Р ПП ДПС Л ИД',   'dps_l_id'),
        ('ДПС Г',           'dps_g'),
        ('ДПС Г ГВ',        'dps_g'),
        ('ДПС Г КП',        'dps_g'),
        ('Р ПП ДПС Г',      'dps_g'),
        ('ДПС Г ИД',        'dps_g_id'),
        ('ДПС Г ФР',        'dps_g_id'),
        ('Р ПП ДПС Г ИД',   'dps_g_id'),
        ('ДПС МК',          'dps_mk'),
        ('ДПС МК ГВ',       'dps_mk'),
        ('ДПС МК КП',       'dps_mk'),
        ('Р ПП ДПС МК',     'dps_mk'),
        ('ДПС МК ИД',       'dps_mk_id'),
        ('ДПС МК ФР',       'dps_mk_id'),
        ('Р ПП ДПС МК ИД',  'dps_mk_id')
),

-- Исходные соединения: от счёта участника к проводкам.
-- Агрегация вынесена ниже, чтобы сравнивать обе версии одинаково.
old_movements AS (
    SELECT s.id AS sharer_id,
           category.category,
           t.value AS signed_value
    FROM old_ops_sharer s
    JOIN back_office.account da
      ON da.owner_sharer_id = s.id
    JOIN back_office.account_type dat
      ON dat.id = da.account_type_id
    JOIN categories category
      ON category.mnemonics = dat.mnemonics
    JOIN back_office.transaction t
      ON t.debit_account_id = da.id
    JOIN back_office.account ca
      ON ca.id = t.credit_account_id
    JOIN back_office.account_type cat
      ON cat.id = ca.account_type_id
    WHERE cat.mnemonics IN ('Р ОПС', 'Р ОПС ИД')

    UNION ALL

    SELECT s.id AS sharer_id,
           category.category,
           -t.value AS signed_value
    FROM old_ops_sharer s
    JOIN back_office.account ca
      ON ca.owner_sharer_id = s.id
    JOIN back_office.account_type cat
      ON cat.id = ca.account_type_id
    JOIN categories category
      ON category.mnemonics = cat.mnemonics
    JOIN back_office.transaction t
      ON t.credit_account_id = ca.id
    JOIN back_office.account da
      ON da.id = t.debit_account_id
    JOIN back_office.account_type dat
      ON dat.id = da.account_type_id
    WHERE dat.mnemonics IN ('Р ОПС', 'Р ОПС ИД')
),

reserve_accounts AS MATERIALIZED (
    SELECT array_agg(a.id) AS ids
    FROM back_office.account a
    JOIN back_office.account_type at
      ON at.id = a.account_type_id
    WHERE at.mnemonics IN ('Р ОПС', 'Р ОПС ИД')
),

reserve_movements AS NOT MATERIALIZED (
    SELECT t.debit_account_id AS account_id,
           t.value AS signed_value
    FROM reserve_accounts r
    JOIN back_office.transaction t
      ON t.credit_account_id = ANY (r.ids)

    UNION ALL

    SELECT t.credit_account_id AS account_id,
           -t.value AS signed_value
    FROM reserve_accounts r
    JOIN back_office.transaction t
      ON t.debit_account_id = ANY (r.ids)
),

-- Новые соединения: от резервных счетов к счетам участников.
new_movements AS (
    SELECT s.id AS sharer_id,
           category.category,
           m.signed_value
    FROM reserve_movements m
    JOIN back_office.account a
      ON a.id = m.account_id
    JOIN back_office.account_type at
      ON at.id = a.account_type_id
    JOIN categories category
      ON category.mnemonics = at.mnemonics
    JOIN new_ops_sharer s
      ON s.id = a.owner_sharer_id
),

both_movements AS (
    SELECT 'old' AS source, m.*
    FROM old_movements m

    UNION ALL

    SELECT 'new' AS source, m.*
    FROM new_movements m
),

balances AS MATERIALIZED (
    SELECT source,
           sharer_id,
           SUM(CASE WHEN category = 'ops'
                    THEN signed_value ELSE 0 END) AS ops,
           SUM(CASE WHEN category = 'ops_id'
                    THEN signed_value ELSE 0 END) AS ops_id,
           SUM(CASE WHEN category = 'dps_l'
                    THEN signed_value ELSE 0 END) AS dps_l,
           SUM(CASE WHEN category = 'dps_l_id'
                    THEN signed_value ELSE 0 END) AS dps_l_id,
           SUM(CASE WHEN category = 'dps_g'
                    THEN signed_value ELSE 0 END) AS dps_g,
           SUM(CASE WHEN category = 'dps_g_id'
                    THEN signed_value ELSE 0 END) AS dps_g_id,
           SUM(CASE WHEN category = 'dps_mk'
                    THEN signed_value ELSE 0 END) AS dps_mk,
           SUM(CASE WHEN category = 'dps_mk_id'
                    THEN signed_value ELSE 0 END) AS dps_mk_id,
           SUM(signed_value) AS total
    FROM both_movements
    GROUP BY source, sharer_id
    HAVING SUM(signed_value) <> 0
),

-- Учитываем финальные соединения исходного отчёта.
results AS MATERIALIZED (
    SELECT b.*,
           i.insurance_number,
           s.number AS sharer_number,
           di.death_date
    FROM balances b
    JOIN ourpension.sharer s
      ON s.id = b.sharer_id
    JOIN ourpension.individual i
      ON i.id = s.individual_id
    JOIN LATERAL (
        SELECT a.death_date
        FROM ourpension.application_death_info a
        WHERE a.individual_id = s.individual_id
          AND a.accepted IS NOT NULL
          AND a.canceled IS NULL
          AND a.rejected IS NULL
        ORDER BY a.version DESC
        LIMIT 1
    ) di ON true
),

old_only_selection AS (
    SELECT id, death_date FROM old_ops_sharer
    EXCEPT ALL
    SELECT id, death_date FROM new_ops_sharer
),

new_only_selection AS (
    SELECT id, death_date FROM new_ops_sharer
    EXCEPT ALL
    SELECT id, death_date FROM old_ops_sharer
),

old_only_results AS (
    SELECT to_jsonb(r) - 'source' AS row_data
    FROM results r
    WHERE source = 'old'

    EXCEPT ALL

    SELECT to_jsonb(r) - 'source'
    FROM results r
    WHERE source = 'new'
),

new_only_results AS (
    SELECT to_jsonb(r) - 'source' AS row_data
    FROM results r
    WHERE source = 'new'

    EXCEPT ALL

    SELECT to_jsonb(r) - 'source'
    FROM results r
    WHERE source = 'old'
)

SELECT
    (SELECT COUNT(*) FROM sample_sharers)
        AS sample_count,
    (SELECT COUNT(*) FROM old_ops_sharer)
        AS old_selected_count,
    (SELECT COUNT(*) FROM new_ops_sharer)
        AS new_selected_count,
    (SELECT COUNT(*) FROM old_only_selection)
        AS selection_only_old,
    (SELECT COUNT(*) FROM new_only_selection)
        AS selection_only_new,
    (SELECT COUNT(*) FROM results WHERE source = 'old')
        AS old_result_count,
    (SELECT COUNT(*) FROM results WHERE source = 'new')
        AS new_result_count,
    (SELECT COUNT(*) FROM old_only_results)
        AS result_only_old,
    (SELECT COUNT(*) FROM new_only_results)
        AS result_only_new,
    (SELECT jsonb_agg(x.row_data)
     FROM (SELECT row_data FROM old_only_results LIMIT 5) x)
        AS differences_old_examples,
    (SELECT jsonb_agg(x.row_data)
     FROM (SELECT row_data FROM new_only_results LIMIT 5) x)
        AS differences_new_examples;

ROLLBACK;