---

-- СНВ: серверная обработка ВСЕХ строк одного уже загруженного XML.
-- Версия 1 для установки и проверки в тестовой базе НПФ 360.
-- Загрузчик a33r8zRX сохраняет XML в ourpension.load_snv до вызова функции.
-- Существующие загрузчик, функции и ручные конструкторы этот скрипт не меняет.
-- Новые объекты: служебная таблица успешных обработок и f_process_snv_file.
-- Модель E57LIUhi / SN0s1Sgg отдельно вызывать не требуется.
--
-- УСТАНОВКА: выполнить весь скрипт. В конце нет вызова обработки данных.
-- Планируемый вызов из операции регламента:
-- SELECT ourpension.f_process_snv_file(:fileId, :username, :user_id);
-- Параметры без значений по умолчанию. Пользователь запуска должен быть
-- отдельным от пользователей ручного прикрепления файлов snv_card.
--
-- УСПЕХ: JSON с success=true. ОШИБКА: исключение SQL (не JSON успеха).
-- Регламент должен считать исключение ошибкой всего файла, а перемещать
-- файл в успешные только ПОСЛЕ фиксации транзакции обработки в БД.
--
-- Повторный вызов того же file_id возвращает сохранённый результат без
-- повторного создания справок. Новый file_id для того же XML здесь НЕ
-- распознаётся как дубликат: это задача учёта файлов в регламенте.
--
-- Дополнительные условия автоматического режима:
-- * ровно одно соответствие исходной строки принятому заявлению;
-- * нет нескольких справок одному клиенту за один год внутри файла:
--   существующая f_merge_year_tax_deduction аннулирует их по клиенту/году;
-- * для каждой строки найден отправленный запрос в журнале ФНС;
-- * отсутствует ручное вложение snv_card у пользователя запуска.
-- Ошибки этих условий останавливают весь файл ДО регистрации.
--
-- Особенность исходного валидатора: проверка с названием «ИНН» фактически
-- проверяет только наличие номера договора. Новое требование совпадения
-- ИНН здесь не вводится. Проверки имени/отчества сохранены как в валидаторе.
-- Сопоставление фамилии/имени и расчёт взносов взяты из модели пользователя.
--
-- Изменения регистрации и журнала откатываются вместе при SQL-ошибке.
-- Значения nextval, как обычно, могут иметь пропуски после отката.
-- Транзакции/перенос сетевых файлов этим скриптом не управляются.
-- Регламенту следует сохранять MESSAGE и DETAIL исключения в журнале ошибок.
--
-- ЛОКАЛЬНАЯ ПРОВЕРКА 26.10.2026: разбор SQL + выполнение в изолированной
-- PostgreSQL/PGlite с тестовыми таблицами, функцией регистрации из присланного
-- кода и тестовыми зависимостями. Пройдены 24 сценария, включая откат при
-- ошибке второй строки, тихом пропуске регистрации и сбое отметки журнала,
-- а также повтор file_id. Это не проверка на базе/в интеграциях заказчика.

BEGIN;

CREATE TABLE ourpension.snv_auto_file_processing (
    file_id integer PRIMARY KEY,
    batch_id integer NOT NULL,
    registered_count integer NOT NULL CHECK (registered_count > 0),
    journal_updated_count integer NOT NULL CHECK (journal_updated_count > 0),
    version_datetime timestamp without time zone NOT NULL,
    processed_at timestamp with time zone NOT NULL DEFAULT clock_timestamp(),
    username text NOT NULL,
    user_id integer NOT NULL,
    result jsonb NOT NULL
);

COMMENT ON TABLE ourpension.snv_auto_file_processing IS
    'Успешная автоматическая регистрация СНВ; защита от повтора одного file_id';

CREATE FUNCTION ourpension.f_process_snv_file(
    p_file_id integer,
    p_username text,
    p_user_id integer
)
RETURNS jsonb
LANGUAGE plpgsql
AS $function$
DECLARE
    v_source jsonb;
    v_mapped jsonb;
    v_ready jsonb;
    v_errors jsonb := '[]'::jsonb;
    v_result jsonb;
    v_saved jsonb;
    v_row record;
    v_bad record;
    v_row_errors text[];
    v_source_count integer;
    v_count integer;
    v_batch_id integer;
    v_registered_count integer := 0;
    v_journal_updated_count integer;
    v_version_datetime timestamp without time zone;
BEGIN
    IF p_file_id IS NULL OR p_file_id <= 0 THEN
        RAISE EXCEPTION 'СНВ: требуется положительный fileId';
    END IF;
    IF p_username IS NULL OR btrim(p_username) = ''
       OR p_user_id IS NULL OR p_user_id <= 0 THEN
        RAISE EXCEPTION 'СНВ: требуются username и положительный user_id регламента';
    END IF;

    -- Два одновременных вызова для одного файла выполняются последовательно.
    PERFORM pg_advisory_xact_lock(hashtext('ourpension.snv_auto.file'), p_file_id);
    SELECT h.result INTO v_saved
    FROM ourpension.snv_auto_file_processing h
    WHERE h.file_id = p_file_id;
    IF FOUND THEN
        RETURN v_saved || jsonb_build_object('already_processed', true);
    END IF;

    -- Не позволяем функции создания справки забрать чужое ручное вложение.
    PERFORM pg_advisory_xact_lock(
        hashtext('ourpension.snv_auto.username'), hashtext(p_username)
    );
    IF EXISTS (
        SELECT 1 FROM dev.upload_file f
        WHERE f.code = 'snv_card' AND f.username = p_username
    ) THEN
        RAISE EXCEPTION 'СНВ: у пользователя запуска есть ручное вложение snv_card; fileId=%', p_file_id;
    END IF;

    -- Снимок уже загруженных строк. Во всех дальнейших шагах используется
    -- этот же набор, а не пользовательский выбор tools.selection.
    PERFORM l.id FROM ourpension.load_snv l
    WHERE l.file_id = p_file_id ORDER BY l.id FOR UPDATE;
    SELECT jsonb_agg(to_jsonb(l) ORDER BY l.id) INTO v_source
    FROM ourpension.load_snv l WHERE l.file_id = p_file_id;
    IF v_source IS NULL THEN
        RAISE EXCEPTION 'СНВ: файл % не содержит загруженных строк', p_file_id;
    END IF;
    v_source_count := jsonb_array_length(v_source);

    -- Проверки исходного валидатора без зависимости от tools.selection.
    FOR v_row IN
        SELECT l.* FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) l
    LOOP
        v_row_errors := ARRAY[]::text[];
        IF NOT EXISTS (
            SELECT 1 FROM ourpension.contract c
            WHERE c.number = v_row.contract_number
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не найден номер договора');
        END IF;
        IF NOT EXISTS (
            SELECT 1 FROM ourpension.contract c
            WHERE c.number = v_row.contract_number AND c.date = v_row.contract_date
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не совпадает дата договора');
        END IF;
        IF EXISTS (
            SELECT 1 FROM ourpension.contract c
            JOIN ourpension.individual i ON i.id = c.individual_id
            WHERE c.number = v_row.contract_number
              AND i.inn = v_row.person_inn
              AND c.service_type <> v_row.service_type
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не совпадает вид деятельности');
        END IF;
        IF NOT EXISTS (
            SELECT 1 FROM ourpension.contract c
            JOIN ourpension.individual i ON i.id = c.individual_id
            WHERE c.number = v_row.contract_number
              AND i.first_name = v_row.person_first_name
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не совпадает имя клиента');
        END IF;
        IF NOT EXISTS (
            SELECT 1 FROM ourpension.contract c
            JOIN ourpension.individual i ON i.id = c.individual_id
            WHERE c.number = v_row.contract_number
              AND i.middle_name = v_row.person_middle_name
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не совпадает отчество клиента');
        END IF;

        -- Исключаем тихий пропуск создания и ошибки преобразования года.
        IF v_row.deduction_sign IS NULL
           OR v_row.deduction_sign::text NOT IN ('RECEIVED', 'NOT_RECEIVED') THEN
            v_row_errors := array_append(v_row_errors, 'Не определён признак получения вычета');
        END IF;
        IF v_row.service_type IS NULL
           OR v_row.service_type::text NOT IN ('NPO', 'PDS') THEN
            v_row_errors := array_append(v_row_errors, 'Не определён вид деятельности НПО/ПДС');
        END IF;
        IF v_row.year IS NULL OR btrim(v_row.year::text) !~ '^[0-9]{1,4}$' THEN
            v_row_errors := array_append(v_row_errors, 'Некорректный год справки');
        ELSIF v_row.year::integer < 1 THEN
            v_row_errors := array_append(v_row_errors, 'Некорректный год справки');
        END IF;

        -- Без отправленного запроса проставить обязательную отметку невозможно.
        IF NOT EXISTS (
            SELECT 1
            FROM report.fns_reference_request_journal fns
            JOIN dev.model_file_to_verification mfc
              ON mfc.id = fns.model_file_to_verification_id
            JOIN report.smev_register sr ON sr.id = mfc.register_id
            WHERE fns.contract_number = v_row.contract_number
              AND sr.number = v_row.request_number::text
              AND fns.is_sent_to_fns = true
        ) THEN
            v_row_errors := array_append(v_row_errors, 'Не найден отправленный запрос в журнале ФНС');
        END IF;
        IF cardinality(v_row_errors) > 0 THEN
            v_errors := v_errors || jsonb_build_array(jsonb_build_object(
                'load_snv_id', v_row.id, 'errors', to_jsonb(v_row_errors)
            ));
        END IF;
    END LOOP;
    IF jsonb_array_length(v_errors) > 0 THEN
        RAISE EXCEPTION USING
            MESSAGE = format('СНВ: проверка файла %s завершилась ошибками; регистрация не начата', p_file_id),
            DETAIL = v_errors::text;
    END IF;

    -- Сопоставление из модели. Номер, дата и вид деятельности должны
    -- относиться к одному договору; иначе строка не готова к регистрации.
    WITH base AS (
        SELECT l.id AS load_snv_id,
               ct.id AS application_termination_contract_id,
               c.individual_id, l.sum_tax_deduction,
               l.year::integer AS year, l.deduction_sign::text AS type,
               c.service_type::text AS service_type,
               sh.sharer_id_1c::text AS sharer_id_1c
        FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) l
        JOIN ourpension.contract c
          ON c.number = l.contract_number AND c.date = l.contract_date
         AND c.service_type = l.service_type
        JOIN ourpension.individual i ON i.id = c.individual_id
         AND lower(btrim(i.last_name)) = lower(btrim(l.person_last_name))
         AND lower(btrim(i.first_name)) = lower(btrim(l.person_first_name))
        JOIN ourpension.sharer sh ON sh.contract_id = c.id
        LEFT JOIN ourpension.application_termination_contract_data ctd
          ON ctd.sharer_id = sh.id
        JOIN ourpension.application_termination_contract ct
          ON CASE WHEN ctd.sharer_id IS NULL THEN ct.sharer_id = sh.id
                  ELSE ct.id = ctd.application_termination_contract_id END
         AND ct.accepted IS NOT NULL AND ct.rejected IS NULL AND ct.canceled IS NULL
         AND ct.termination_type IN ('TRANSFER', 'PAYMENT', 'FULL', 'LIFE_SITUATION')
    )
    SELECT coalesce(jsonb_agg(to_jsonb(b) ORDER BY b.load_snv_id), '[]'::jsonb)
    INTO v_mapped FROM base b;

    FOR v_row IN
        SELECT l.id FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) l
    LOOP
        SELECT count(*) INTO v_count
        FROM jsonb_to_recordset(v_mapped) b(load_snv_id integer)
        WHERE b.load_snv_id = v_row.id;
        IF v_count <> 1 THEN
            v_errors := v_errors || jsonb_build_array(jsonb_build_object(
                'load_snv_id', v_row.id, 'matches', v_count,
                'error', 'Требуется ровно одно соответствие клиенту, участнику и принятому заявлению'
            ));
        END IF;
    END LOOP;
    FOR v_bad IN
        SELECT b.individual_id, b.year, jsonb_agg(b.load_snv_id) AS row_ids
        FROM jsonb_to_recordset(v_mapped)
             b(load_snv_id integer, individual_id integer, year integer)
        GROUP BY b.individual_id, b.year HAVING count(*) > 1
    LOOP
        v_errors := v_errors || jsonb_build_array(jsonb_build_object(
            'load_snv_ids', v_bad.row_ids,
            'error', 'Несколько справок одного клиента за один год: они аннулируют друг друга'
        ));
    END LOOP;
    FOR v_bad IN
        SELECT b.load_snv_id FROM jsonb_to_recordset(v_mapped)
             b(load_snv_id integer, sharer_id_1c text)
        WHERE b.sharer_id_1c IS NULL OR btrim(b.sharer_id_1c) = ''
    LOOP
        v_errors := v_errors || jsonb_build_array(jsonb_build_object(
            'load_snv_id', v_bad.load_snv_id, 'error', 'Не определён счёт 1С для расчёта взносов'
        ));
    END LOOP;
    IF jsonb_array_length(v_errors) > 0 THEN
        RAISE EXCEPTION USING
            MESSAGE = format('СНВ: неполный или неоднозначный набор файла %s; регистрация не начата', p_file_id),
            DETAIL = v_errors::text;
    END IF;

    -- Расчёт из модели SN0s1Sgg: один запрос операций на пару счёт/вид.
    WITH base AS (
        SELECT * FROM jsonb_to_recordset(v_mapped) b(
            load_snv_id integer, application_termination_contract_id integer,
            individual_id integer, sum_tax_deduction numeric, year integer,
            type text, service_type text, sharer_id_1c text
        )
    ), account_pairs AS (
        SELECT DISTINCT b.sharer_id_1c, b.service_type FROM base b
    ), operations AS (
        SELECT p.sharer_id_1c, p.service_type,
               date_part('year', t.period_date)::integer AS year, t.amount
        FROM account_pairs p
        JOIN mgr.ft_dwh_get__pension_account_operation(jsonb_build_object(
            'p_sPensionAccountId', p.sharer_id_1c, 'p_sServiceType', p.service_type
        )) t ON true
        WHERE CASE WHEN p.service_type = 'NPO'
                        THEN t.operation_type_name IN ('Ч/з банк', 'От работодателя')
                   WHEN p.service_type = 'PDS'
                        THEN t.operation_type_name = 'Сберегательные взносы'
                   ELSE false END
    ), sum_by_year AS (
        SELECT o.sharer_id_1c, o.service_type, o.year, sum(o.amount) AS value
        FROM operations o GROUP BY o.sharer_id_1c, o.service_type, o.year
    ), prepared AS (
        SELECT b.*, coalesce(s.value, 0) AS sum_income,
               m.value AS max_tax_deduction
        FROM base b
        LEFT JOIN sum_by_year s ON s.sharer_id_1c = b.sharer_id_1c
             AND s.service_type = b.service_type AND s.year = b.year
        LEFT JOIN ourpension.max_year_tax_deduction m
             ON m.year = b.year AND coalesce(s.value, 0) >= 0
    )
    SELECT jsonb_agg(to_jsonb(p) ORDER BY p.load_snv_id)
    INTO v_ready FROM prepared p;

    -- Проверяем также возможное размножение строк справочником лимитов.
    FOR v_row IN
        SELECT l.id FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) l
    LOOP
        SELECT count(*) INTO v_count FROM jsonb_to_recordset(v_ready)
             b(load_snv_id integer) WHERE b.load_snv_id = v_row.id;
        IF v_count <> 1 THEN
            RAISE EXCEPTION 'СНВ: для строки % файла % получено % наборов параметров',
                v_row.id, p_file_id, v_count;
        END IF;
    END LOOP;

    -- Упорядоченные блокировки сериализуют автоматическую регистрацию
    -- пересекающихся клиент/год из разных файлов. Ручной код эти блокировки
    -- не использует; совместный запуск с ручным режимом требует отдельной QA.
    FOR v_row IN
        SELECT b.individual_id, b.year FROM jsonb_to_recordset(v_ready)
             b(individual_id integer, year integer)
        ORDER BY b.individual_id, b.year
    LOOP
        PERFORM pg_advisory_xact_lock(hashtext('ourpension.snv_auto.person_year'),
            hashtext(v_row.individual_id::text || ':' || v_row.year::text));
    END LOOP;

    v_batch_id := nextval('ourpension.year_tax_deduction_batch_sequence')::integer;
    v_version_datetime := clock_timestamp() AT TIME ZONE 'Europe/Moscow';
    FOR v_row IN
        SELECT * FROM jsonb_to_recordset(v_ready) b(
            load_snv_id integer, application_termination_contract_id integer,
            individual_id integer, sum_tax_deduction numeric, year integer,
            type text, sum_income numeric, max_tax_deduction numeric
        ) ORDER BY b.load_snv_id
    LOOP
        -- p_date NULL воспроизводит отсутствующую row.date в ручной модели.
        -- Текущее время создания задаётся явно последним аргументом.
        -- Сумма -1, как и в операции qu84AJSQB, означает NULL.
        PERFORM ourpension.f_merge_year_tax_deduction(
            v_row.application_termination_contract_id::integer, NULL::date,
            v_row.individual_id::integer, v_row.max_tax_deduction::integer,
            v_row.sum_income::numeric, NULLIF(v_row.sum_tax_deduction, -1)::numeric,
            v_row.type::text, v_row.year::integer, p_username::text,
            p_user_id::integer, v_batch_id::integer, NULL::integer,
            v_version_datetime::timestamp
        );
        -- Возврат функции всегда 0; проверяем реальный результат записи.
        SELECT count(*) INTO v_count FROM ourpension.year_tax_deduction y
        WHERE y.batch_id = v_batch_id AND y.individual_id = v_row.individual_id
          AND y.year = v_row.year
          AND y.application_termination_contract_id = v_row.application_termination_contract_id
          AND y.type::text = v_row.type AND y.canceled IS NULL
          AND y.accepted IS NOT NULL;
        IF v_count <> 1 THEN
            RAISE EXCEPTION 'СНВ: не подтверждена регистрация строки % файла %',
                v_row.load_snv_id, p_file_id;
        END IF;
        v_registered_count := v_registered_count + 1;
    END LOOP;

    -- Исходный UPDATE из TMbTo05q выполняется только после ВСЕХ справок.
    UPDATE report.fns_reference_request_journal fns
    SET response_received = true
    FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) snv,
         dev.model_file_to_verification mfc, report.smev_register sr
    WHERE fns.model_file_to_verification_id = mfc.id
      AND mfc.register_id = sr.id
      AND fns.contract_number = snv.contract_number
      AND sr.number = snv.request_number::text
      AND fns.is_sent_to_fns = true
      AND snv.file_id = p_file_id;
    GET DIAGNOSTICS v_journal_updated_count = ROW_COUNT;

    -- Повторяем контроль сопоставления: нулевой UPDATE не является успехом.
    FOR v_row IN
        SELECT l.* FROM jsonb_populate_recordset(NULL::ourpension.load_snv, v_source) l
    LOOP
        IF NOT EXISTS (
            SELECT 1 FROM report.fns_reference_request_journal fns
            JOIN dev.model_file_to_verification mfc
              ON mfc.id = fns.model_file_to_verification_id
            JOIN report.smev_register sr ON sr.id = mfc.register_id
            WHERE fns.contract_number = v_row.contract_number
              AND sr.number = v_row.request_number::text
              AND fns.is_sent_to_fns = true AND fns.response_received = true
        ) THEN
            RAISE EXCEPTION 'СНВ: отметка ответа не подтверждена для строки % файла %',
                v_row.id, p_file_id;
        END IF;
    END LOOP;
    IF v_registered_count <> v_source_count OR v_journal_updated_count < 1 THEN
        RAISE EXCEPTION 'СНВ: неполный результат регистрации файла %', p_file_id;
    END IF;

    v_result := jsonb_build_object(
        'success', true, 'already_processed', false, 'file_id', p_file_id,
        'batch_id', v_batch_id, 'source_count', v_source_count,
        'registered_count', v_registered_count,
        'journal_updated_count', v_journal_updated_count,
        'version_datetime', v_version_datetime
    );
    INSERT INTO ourpension.snv_auto_file_processing(
        file_id, batch_id, registered_count, journal_updated_count,
        version_datetime, username, user_id, result
    ) VALUES (
        p_file_id, v_batch_id, v_registered_count, v_journal_updated_count,
        v_version_datetime, p_username, p_user_id, v_result
    );
    RETURN v_result;
    -- Не перехватываем ошибку как успех: исключение откатывает весь вызов.
END;
$function$;

COMMENT ON FUNCTION ourpension.f_process_snv_file(integer, text, integer) IS
    'Все строки загруженного СНВ: проверка, расчёт, регистрация и отметка ответа; SQL-ошибка откатывает вызов';

COMMIT;

-- ПРОВЕРКИ В ТЕСТОВОЙ БАЗЕ ПОСЛЕ УСТАНОВКИ (выполняются отдельно):
-- 1. Валидный файл с несколькими разными клиент/год: все справки в одном batch.
-- 2. Ошибка хотя бы одной строки: нет новых справок, аннулирования и отметок.
-- 3. Нулевой набор, неизвестный договор, несколько заявлений, неверный год,
--    неопределённый тип вычета, нет записи журнала: весь файл отклонён.
-- 4. Сбой на второй регистрации: первая запись и аннулирование откатились.
-- 5. Повтор ТОГО ЖЕ file_id: already_processed=true, batch и количество прежние.
-- 6. Прежняя справка того же клиента/года: аннулируется по действующим правилам.
-- 7. Проверить фактическую сигнатуру mgr.ft_dwh_get__pension_account_operation,
--    типы полей и права отдельного пользователя регламента в целевой базе.
-- 8. После успешного COMMIT проверить перенос файла отдельно: этот SQL
--    не подключает расписание, сетевую папку или обработку результата роботом.