-- 149_developer_attribution.sql
-- #1088 (GG-форсайт) — детерминированный geo-join «пятно земли ↔ застройщик» для
-- Site Finder analyze/competitors. Связывает РНС-сторону (ekburg_construction_permits)
-- с домрф-стороной (domrf_kn_objects) через НОРМАЛИЗОВАННЫЙ ИНН — единственный
-- стабильный кросс-источниковый ключ застройщика — и резолвит участок → застройщик
-- (имя+ИНН), его проект/статус, и track-record застройщика по району/кварталу.
--
-- ── ЗАЧЕМ ────────────────────────────────────────────────────────────────────────
-- До #1088 analyze/competitors «слепы» к РНС: domrf_kn_objects.dev_inn даёт застройщика
-- для домрф-объектов, но НЕТ связки с РНС-разрешениями и НЕТ канон-имени застройщика
-- (одно ИНН пишется по-разному: prod — 163 ИНН в permits имеют >1 написания имени;
-- «ИП Хабибуллин О.В.» vs «ИП Хабибуллин Олег Вахалиевич»). Этот слой даёт:
--   1. fn_normalize_inn      — детерминированная нормализация ИНН (10/12-знач, иначе NULL)
--   2. developer_registry    — MATERIALIZED реестр застройщиков по норм-ИНН: канон-имя
--      + track-record (refresh через fn_refresh_developer_registry, unique-idx по ИНН)
--   3. developer_attribution — алиас-VIEW с denormalized полями для analyze-блока
--   4. fn_developer_for_parcel(cad_num) — резолвер участок→застройщик (3-ступенчатый
--      fallback: точный кадастр → spatial 150 м → кадастровый квартал)
--
-- ── PROD-ФАКТЫ (подтверждено EXPLAIN/COUNT, 2026-06-13) ───────────────────────────
-- ekburg_construction_permits: 4059 строк; developer_inn TEXT not-null=3564 (87.8%),
--   все pure-digit; 3401×10-знач (юрлица), 154×12-знач (ИП), 9×11-знач (СЛИПШИЙСЯ
--   мусор → норм. отбрасывает в NULL). cadastral_number not-null=4054 (99.9%); 3439
--   full-cad, 57 multi-value (запятая/«;» — несколько ЗУ в одной РНС), 2 quarter-only.
--   geom GEOMETRY(Point,4326) not-null=3994 (98.4%), GIST gist_ekburg_permits_geom.
--   НЕТ btree на cadastral_number → exact-lookup = Seq Scan 33мс (фикс ниже).
-- domrf_kn_objects: 10699 строк = 1543 distinct obj_id × снапшоты (2026-04-27..06-10),
--   ВСЕ region_cd=66. dev_inn TEXT, на latest snapshot not-null=1543 (100%), все
--   10-знач. ИНН СТАБИЛЕН между снапшотами (0 объектов меняют ИНН). cad_quarter —
--   формат «66:41:NNNNNNN», заполнен на 1131/1543 объектах (73%), НА LATEST snapshot
--   часто NULL → берём с ЛЮБОГО снапшота где NOT NULL (тоже стабилен, 0 изменений).
--   cad_building_num — ВЕЗДЕ NULL (не используем). НЕТ колонки geom (только
--   latitude/longitude NUMERIC, у всех 1543) → domrf spatial идёт через MakePoint.
--   district_name заполнен на 1315/1543 (85%), 8 админ-районов ЕКБ.
-- Кросс-матч норм-ИНН: 888 distinct в permits, 428 в domrf, СОВПАДАЮТ 292 (68.2%
--   застройщиков domrf имеют РНС в ЕКБ). 1740 permit-строк (42.9%) принадлежат
--   застройщику из domrf; 1240 domrf-объектов (80.4%) — застройщику из permits.
--
-- ── ИНН-НОРМАЛИЗАЦИЯ (детерминированная, edge-cases) ──────────────────────────────
-- fn_normalize_inn: btrim → если результат matches '^[0-9]{10}$' ИЛИ '^[0-9]{12}$'
-- → вернуть как есть; ИНАЧЕ NULL. Покрывает edge-cases:
--   NULL / '' / '0'         → NULL  (495 permit-строк, 0 domrf)
--   11-знач '66711382990'   → NULL  (9 permit-строк — два слипшихся ИНН/ИП; чинить
--                                    эвристикой НЕ беремся — детерминированно мусор)
--   ' 6671138299 ' (пробелы)→ '6671138299'  (btrim)
--   буквы/дефисы/запятые    → NULL  (не pure-digit)
-- Ведущие нули НЕ режем: российский ИНН — фикс-длины строка-идентификатор, '00…' —
-- значимые цифры (срезка сломала бы матч). IMMUTABLE — годен для functional index.
--
-- ── КАНОН-ИМЯ (dedupe, детерминированный) ─────────────────────────────────────────
-- Один ИНН → много написаний. Канон выбираем так (ORDER BY, без LLM/рандома):
--   1) приоритет источника: domrf-имя > permits-имя (domrf — юрлица из реестра дом.рф,
--      чище и полнее, чем ручной ввод РНС);
--   2) внутри источника — самое ЧАСТОЕ написание (по числу строк);
--   3) tie-break — лексикографически наименьшее (btrim'нутое) → стабильно при ре-apply.
--
-- ── РЕЗОЛВЕР УЧАСТОК→ЗАСТРОЙЩИК (3 ступени, детерминированный приоритет) ───────────
-- fn_developer_for_parcel(cad_num) возвращает атрибуцию с match_method:
--   'exact_cadastral' — permits.cadastral_number = cad_num (точное, max precision);
--   'spatial_150m'    — permit.geom в 150 м от geom участка (GIST envelope + geography
--                        refine); берём ближайший permit;
--   'quarter'         — по кадастровому кварталу: permits и domrf того же квартала.
-- Ступени НЕ взаимоисключающи на уровне данных, но метод проставляется по ЛУЧШЕЙ
-- сработавшей ступени (exact > spatial > quarter). Возвращаем ВСЕХ кандидатов с
-- match_method и distance_m — caller (backend) решает, брать топ-1 или ранжировать.
--
-- ── SPATIAL GIST-TRAP (важно для перфа) ───────────────────────────────────────────
-- ST_DWithin(geom::geography, …) НЕ использует GIST (индекс на geometry, не geography)
-- → Seq Scan 91мс. Правильно: ST_DWithin(geom, geom, degrees) для GIST-envelope +
-- ST_DWithin(::geography, ::geography, meters) для точной метрики. EXPLAIN: 91→12мс,
-- 1689→132 буфера, Index Scan using gist_ekburg_permits_geom. 0.0025° envelope на
-- широте ЕКБ (~56.8°N) безопасно перекрывает 150 м (1°lat≈111 км, 1°lon≈61 км).
--
-- ── DEPLOY ────────────────────────────────────────────────────────────────────────
-- Auto-applied deploy.yml через _schema_migrations (ровно один раз, NN=149).
-- Idempotent: CREATE OR REPLACE FUNCTION/VIEW + CREATE INDEX IF NOT EXISTS +
-- DROP MATERIALIZED VIEW IF EXISTS ... CASCADE перед CREATE MATERIALIZED VIEW (matview
-- не поддерживает CREATE OR REPLACE). CREATE MATERIALIZED VIEW наполняется данными
-- СРАЗУ (без WITH NO DATA) → реестр готов к чтению по завершении миграции; первый
-- плановый fn_refresh_developer_registry() просто обновит его при след. sync.
-- Migration order: эта SQL-миграция ПЕРВОЙ; backend-сервис
-- (app/services/site_finder/developer_attribution.py — рекомендация в отчёте) — следом.
-- ЧИСТО ADDITIVE: ни одна существующая таблица/вью/колонка не меняется и не дропается.

BEGIN;

-- ──────────────────────────────────────────────────────────────────────────────────
-- 1. fn_normalize_inn — детерминированная нормализация ИНН
-- ──────────────────────────────────────────────────────────────────────────────────
-- btrim + валидация длины/цифр. 10 (юрлицо) или 12 (физлицо/ИП) → как есть, иначе NULL.
-- IMMUTABLE STRICT: STRICT → NULL-вход даёт NULL-выход без вызова тела; IMMUTABLE →
-- можно использовать в functional index и планировщик кэширует результат.
CREATE OR REPLACE FUNCTION fn_normalize_inn(raw text)
RETURNS text
LANGUAGE sql
IMMUTABLE
STRICT
AS $$
    SELECT CASE
        WHEN btrim(raw) ~ '^[0-9]{10}$' OR btrim(raw) ~ '^[0-9]{12}$'
        THEN btrim(raw)
        ELSE NULL
    END;
$$;

COMMENT ON FUNCTION fn_normalize_inn(text) IS
    '#1088: детерминированная нормализация ИНН → 10/12-значная pure-digit строка, '
    'иначе NULL. NULL/пусто/0/11-знач-мусор/буквы → NULL. Ведущие нули сохраняются '
    '(фикс-длины идентификатор). IMMUTABLE — годна для functional index.';

-- ──────────────────────────────────────────────────────────────────────────────────
-- 2. Индексы для горячего пути резолвера (закрывают Seq Scan / ускоряют join по ИНН)
-- ──────────────────────────────────────────────────────────────────────────────────
-- 2a. btree на cadastral_number — exact-lookup (= cad_num) делал Seq Scan (4059 строк,
--     33мс). С индексом — Bitmap Index Scan (0.09мс). Equality collation-safe.
CREATE INDEX IF NOT EXISTS idx_ekburg_permits_cadastral
    ON ekburg_construction_permits (cadastral_number);

-- 2a-bis. text_pattern_ops btree — для quarter-ступени LIKE 'QUARTER:%'. БД collation
--     en_US.utf8 (НЕ C) → обычный btree НЕ обслуживает ни LIKE-prefix, ни range
--     ['X:','X;') (collation сортирует ':'/';'/цифры не по байтам → range вернул 0
--     строк, см. dry-run-баг). text_pattern_ops индексирует по C-семантике байтов →
--     LIKE 'prefix%' даёт Bitmap Index Scan (0.74мс vs Seq Scan 6мс от regex '^').
CREATE INDEX IF NOT EXISTS idx_ekburg_permits_cad_cpat
    ON ekburg_construction_permits (cadastral_number text_pattern_ops);

-- 2b. functional index на нормализованный ИНН — ускоряет join permits⋈domrf по
--     fn_normalize_inn(developer_inn). Partial (WHERE norm IS NOT NULL) — мусорные
--     ИНН (NULL после норм.) в индекс не попадают, держим его компактным.
CREATE INDEX IF NOT EXISTS idx_ekburg_permits_inn_norm
    ON ekburg_construction_permits (fn_normalize_inn(developer_inn))
    WHERE fn_normalize_inn(developer_inn) IS NOT NULL;

-- 2c. functional index на норм-ИНН для domrf (join вторая сторона). domrf
--     версионируется по snapshot_date → индекс по (norm_inn) хватает для матча
--     застройщика (ИНН стабилен между снапшотами).
CREATE INDEX IF NOT EXISTS idx_domrf_kn_objects_inn_norm
    ON domrf_kn_objects (fn_normalize_inn(dev_inn))
    WHERE fn_normalize_inn(dev_inn) IS NOT NULL;

-- ──────────────────────────────────────────────────────────────────────────────────
-- 3. developer_registry — реестр застройщиков по нормализованному ИНН + track-record
-- ──────────────────────────────────────────────────────────────────────────────────
-- Канон-имя застройщика (dedupe вариантов) + агрегаты по ОБЕИМ сторонам:
--   permits-сторона: число РНС/РВЭ, суммарные площади, диапазон лет, набор кварталов;
--   domrf-сторона:   число ЖК (latest snapshot per obj_id), суммарно квартир/в продаже,
--                    набор районов, набор классов.
--
-- MATERIALIZED (не plain VIEW): как VIEW point-lookup по ИНН стоил 52мс, а резолвер
-- 142-184мс — потому что план КАЖДЫЙ РАЗ пересчитывал весь реестр (DISTINCT ON 10699
-- domrf-строк + два FULL JOIN + dedupe-имя) ради джойна track-record к ~14-460
-- кандидатам участка. Реестр мал (1024 застройщика) и меняется РЕДКО (permits —
-- monthly sync #105; domrf — снапшоты ~раз в сутки), поэтому материализуем: резолвер
-- и прямой lookup читают готовые 1024 строки через unique-индекс за микросекунды.
-- Refresh — fn_refresh_developer_registry() по расписанию (см. ниже + рекомендация
-- backend). DROP ... CASCADE для идемпотентности (matview не поддерживает CREATE OR
-- REPLACE): уронит зависимый developer_attribution VIEW — он пересоздаётся ниже в §4.
DROP MATERIALIZED VIEW IF EXISTS developer_registry CASCADE;
CREATE MATERIALIZED VIEW developer_registry AS
WITH
-- 3.1 domrf latest snapshot per obj_id (ИНН/имя стабильны между снапшотами).
--      Только поля, нужные registry-агрегатам (canonical_name + domrf_agg). cad_quarter
--      здесь НЕ тянем (registry агрегирует по району/классу, не по кварталу — а резолвер
--      quarter_domrf читает domrf_kn_objects напрямую). Это снимает O(N) коррелированный
--      subplan (был 93k буферов / 10699 loops в EXPLAIN) — один проход DISTINCT ON.
domrf_latest AS (
    SELECT DISTINCT ON (o.obj_id)
        o.obj_id,
        fn_normalize_inn(o.dev_inn)                              AS inn,
        btrim(o.dev_name)                                        AS dev_name,
        o.obj_class,
        o.site_status,
        o.flat_count,
        o.free_flats,
        o.district_name
    FROM domrf_kn_objects o
    WHERE fn_normalize_inn(o.dev_inn) IS NOT NULL
    ORDER BY o.obj_id, o.snapshot_date DESC
),
-- 3.2 Канон-имя: domrf-имя приоритетнее permits-имени; внутри — самое частое;
--      tie → min lexical. Собираем кандидаты обеих сторон с приоритетом источника.
name_candidates AS (
    SELECT inn, dev_name AS name, 1 AS source_priority   -- domrf > permits
    FROM domrf_latest
    WHERE dev_name IS NOT NULL AND dev_name <> ''
    UNION ALL
    SELECT fn_normalize_inn(developer_inn) AS inn, btrim(developer_name) AS name, 2 AS source_priority
    FROM ekburg_construction_permits
    WHERE fn_normalize_inn(developer_inn) IS NOT NULL
      AND developer_name IS NOT NULL AND btrim(developer_name) <> ''
),
canonical_name AS (
    SELECT DISTINCT ON (inn)
        inn,
        name AS developer_name
    FROM (
        SELECT inn, name, source_priority, count(*) AS freq
        FROM name_candidates
        GROUP BY inn, name, source_priority
    ) ranked
    ORDER BY inn, source_priority ASC, freq DESC, name ASC
),
-- 3.3 permits track-record по застройщику (норм-ИНН)
permits_agg AS (
    SELECT
        fn_normalize_inn(developer_inn)                                       AS inn,
        count(*)                                                              AS permits_total,
        count(*) FILTER (WHERE permit_type = 'RNS')                           AS permits_rns,
        count(*) FILTER (WHERE permit_type = 'RVE')                           AS permits_rve,
        sum(total_area_sqm)                                                   AS permits_total_area_sqm,
        sum(living_area_sqm)                                                  AS permits_living_area_sqm,
        min(issue_date)                                                       AS permits_first_issue_date,
        max(issue_date)                                                       AS permits_last_issue_date,
        array_agg(DISTINCT substring(cadastral_number from '^(\d{2}:\d{2}:\d{6,7})'))
            FILTER (WHERE cadastral_number ~ '^\d{2}:\d{2}:\d{6,7}')          AS permits_quarters
    FROM ekburg_construction_permits
    WHERE fn_normalize_inn(developer_inn) IS NOT NULL
    GROUP BY fn_normalize_inn(developer_inn)
),
-- 3.4 domrf track-record по застройщику (по latest-snapshot объектам)
domrf_agg AS (
    SELECT
        inn,
        count(*)                                                  AS domrf_objects_total,
        sum(flat_count)                                           AS domrf_flats_total,
        sum(free_flats)                                           AS domrf_free_flats_total,
        count(*) FILTER (WHERE site_status = 'Строящиеся')        AS domrf_active_objects,
        array_agg(DISTINCT district_name)
            FILTER (WHERE district_name IS NOT NULL)              AS domrf_districts,
        array_agg(DISTINCT obj_class)
            FILTER (WHERE obj_class IS NOT NULL)                  AS domrf_classes
    FROM domrf_latest
    GROUP BY inn
)
SELECT
    COALESCE(p.inn, d.inn)                          AS developer_inn,
    cn.developer_name                               AS developer_name,
    -- track-record permits-сторона
    COALESCE(p.permits_total, 0)                    AS permits_total,
    COALESCE(p.permits_rns, 0)                      AS permits_rns,
    COALESCE(p.permits_rve, 0)                      AS permits_rve,
    p.permits_total_area_sqm,
    p.permits_living_area_sqm,
    p.permits_first_issue_date,
    p.permits_last_issue_date,
    p.permits_quarters,
    -- track-record domrf-сторона
    COALESCE(d.domrf_objects_total, 0)              AS domrf_objects_total,
    COALESCE(d.domrf_active_objects, 0)             AS domrf_active_objects,
    d.domrf_flats_total,
    d.domrf_free_flats_total,
    d.domrf_districts,
    d.domrf_classes,
    -- известен ли застройщик обеим сторонам (кросс-валидация)
    (p.inn IS NOT NULL AND d.inn IS NOT NULL)       AS in_both_sources
FROM permits_agg p
FULL OUTER JOIN domrf_agg d ON p.inn = d.inn
LEFT JOIN canonical_name cn ON cn.inn = COALESCE(p.inn, d.inn);

COMMENT ON MATERIALIZED VIEW developer_registry IS
    '#1088: реестр застройщиков по НОРМАЛИЗОВАННОМУ ИНН (fn_normalize_inn) — канон-имя '
    '(domrf>permits, most-frequent, tie→min-lexical) + track-record обеих сторон '
    '(permits РНС/РВЭ/площади/кварталы, domrf ЖК/квартиры/районы/классы). '
    'in_both_sources — кросс-валидация (prod: 292 ИНН). FULL OUTER JOIN — застройщик '
    'может быть только в одной стороне. MATERIALIZED — refresh через '
    'fn_refresh_developer_registry() (permits monthly / domrf daily).';

-- 3.5 UNIQUE индекс на developer_inn — обязателен для REFRESH ... CONCURRENTLY
-- (иначе refresh блокирует читателей). Одновременно обслуживает point-lookup по ИНН
-- (резолвер джойнит сюда, analyze тянет track-record по конкретному застройщику).
-- developer_inn здесь уже НОРМАЛИЗОВАН и NOT NULL (COALESCE двух not-null норм-ИНН).
CREATE UNIQUE INDEX IF NOT EXISTS uq_developer_registry_inn
    ON developer_registry (developer_inn);

-- 3.6 fn_refresh_developer_registry — обёртка REFRESH CONCURRENTLY (вызывает Celery-
-- beat / sync-задачи permits+domrf после обновления источников). CONCURRENTLY — не
-- блокирует SELECT-читателей (analyze) во время refresh; требует unique-индекс выше.
CREATE OR REPLACE FUNCTION fn_refresh_developer_registry()
RETURNS void
LANGUAGE plpgsql
AS $$
BEGIN
    REFRESH MATERIALIZED VIEW CONCURRENTLY developer_registry;
END;
$$;

COMMENT ON FUNCTION fn_refresh_developer_registry() IS
    '#1088: REFRESH MATERIALIZED VIEW CONCURRENTLY developer_registry. Вызывать после '
    'обновления источников (ekburg_permits_sync monthly / domrf_kn daily snapshot).';

-- ──────────────────────────────────────────────────────────────────────────────────
-- 4. developer_attribution — алиас-VIEW для analyze-блока (стабильное имя контракта)
-- ──────────────────────────────────────────────────────────────────────────────────
-- ТЗ #1088 называет целевую сущность `developer_attribution`. Держим её тонким VIEW
-- поверх materialized developer_registry: тот же набор строк, имя зафиксировано в
-- контракте (backend ходит сюда). VIEW поверх matview читается так же дёшево (matview
-- — физическая таблица с unique-индексом). Пересоздаётся ЗДЕСЬ после DROP ... CASCADE
-- в §3 (CASCADE роняет этот зависимый VIEW при ре-apply).
CREATE OR REPLACE VIEW developer_attribution AS
    SELECT * FROM developer_registry;

COMMENT ON VIEW developer_attribution IS
    '#1088: стабильное имя контракта для analyze-блока. Алиас materialized '
    'developer_registry (канон застройщик + track-record по норм-ИНН).';

-- ──────────────────────────────────────────────────────────────────────────────────
-- 5. fn_developer_for_parcel — резолвер участок → застройщик (3-ступенчатый fallback)
-- ──────────────────────────────────────────────────────────────────────────────────
-- Вход: cad_num участка. Геометрию/квартал берём ВНУТРИ из cad_parcels_geom (view над
-- cad_parcels, GIST cad_parcels_geom_gist) + _quarter_from_cad по строке номера.
-- Выход: SETOF строк-кандидатов с источником/методом/дистанцией + denormalized
-- track-record (join к developer_registry по ИНН). Caller (backend) ранжирует/берёт
-- топ-1. p_radius_m — радиус spatial-ступени (default 150).
--
-- STABLE (не IMMUTABLE) — читает таблицы. Параметр geom не принимаем: cad_num —
-- единственный вход (геометрия резолвится внутри), чтобы контракт был как у
-- competitors.get_competitors(cad_num). Если участка нет в cad_parcels_geom —
-- spatial/quarter-domrf по geom не сработают, но exact-cadastral и quarter-permits
-- (по строке cad_num) всё равно отработают.
CREATE OR REPLACE FUNCTION fn_developer_for_parcel(
    p_cad_num  text,
    p_radius_m double precision DEFAULT 150.0
)
RETURNS TABLE (
    developer_inn   text,
    developer_name  text,
    source          text,    -- 'permits' | 'domrf'
    match_method    text,    -- 'exact_cadastral' | 'spatial_150m' | 'quarter'
    distance_m      double precision,
    project         text,    -- permit object_name / domrf comm_name
    status          text,    -- permit_type / domrf site_status
    obj_class       text,
    permits_total       integer,
    domrf_objects_total integer,
    domrf_active_objects integer,
    in_both_sources bool
)
LANGUAGE sql
STABLE
AS $$
    WITH
    quarter AS (
        -- кадастровый квартал из номера участка: '66:41:0313010:2' → '66:41:0313010'
        SELECT substring(p_cad_num from '^(\d{2}:\d{2}:\d{6,7})') AS q
    ),
    parcel AS (
        SELECT geom, ST_Centroid(geom) AS centroid
        FROM cad_parcels_geom
        WHERE cad_num = p_cad_num AND geom IS NOT NULL
        LIMIT 1
    ),
    -- ── Ступень 1: точный кадастр (permits.cadastral_number = cad_num) ──────────────
    -- Две ветки, объединённые UNION (каждая sargable / узкая):
    --   1a. РОВНОЕ равенство = cad_num → Bitmap Index Scan по idx_ekburg_permits_cadastral
    --       (0.09мс; покрывает single-value строки = 98.6% permits);
    --   1b. multi-value строки РНС (несколько ЗУ через запятую/«;» — prod: 57 строк):
    --       cad_num как подстрока на границе. Regex применяется ТОЛЬКО к узкому
    --       подмножеству (LIKE '%,%' OR '%;%') — не Seq Scan по всем 4059.
    exact_match AS (
        SELECT
            fn_normalize_inn(e.developer_inn) AS inn,
            'permits'::text  AS source,
            'exact_cadastral'::text AS match_method,
            0.0::double precision AS distance_m,
            e.object_name AS project,
            e.permit_type AS status,
            e.object_type AS obj_class
        FROM ekburg_construction_permits e
        WHERE e.cadastral_number = p_cad_num          -- 1a: btree equality
        UNION
        SELECT
            fn_normalize_inn(e.developer_inn) AS inn,
            'permits'::text, 'exact_cadastral'::text, 0.0::double precision,
            e.object_name, e.permit_type, e.object_type
        FROM ekburg_construction_permits e
        WHERE (e.cadastral_number LIKE '%,%' OR e.cadastral_number LIKE '%;%')  -- 1b: multi-value subset
          AND e.cadastral_number ~ ('(^|[,; ])' || replace(p_cad_num, '.', '\.') || '($|[,; ])')
    ),
    -- ── Ступень 2: spatial 150 м (permit.geom рядом с geom участка) ─────────────────
    -- GIST-friendly: ST_DWithin(geom, geom, deg) даёт index-scan по
    -- gist_ekburg_permits_geom, затем geography-refine на точные метры. 0.0025° ⊃ 150м.
    spatial_match AS (
        SELECT
            fn_normalize_inn(e.developer_inn) AS inn,
            'permits'::text AS source,
            'spatial_150m'::text AS match_method,
            ST_Distance(e.geom::geography, pa.centroid::geography) AS distance_m,
            e.object_name AS project,
            e.permit_type AS status,
            e.object_type AS obj_class
        FROM ekburg_construction_permits e
        CROSS JOIN parcel pa
        WHERE e.geom IS NOT NULL
          AND ST_DWithin(e.geom, pa.geom, p_radius_m / 111000.0)
          AND ST_DWithin(e.geom::geography, pa.centroid::geography, p_radius_m)
    ),
    -- ── Ступень 3a: кадастровый квартал — permits ───────────────────────────────────
    -- LIKE 'QUARTER:%' — sargable через text_pattern_ops индекс (idx_ekburg_permits_
    -- cad_cpat): все ЗУ квартала имеют вид 'QUARTER:NNN'. → Bitmap Index Scan (0.74мс),
    -- вместо Seq Scan от regex '^' (6мс). НЕ range ['X:','X;') — он collation-unsafe на
    -- en_US.utf8 (вернул 0 строк в dry-run); LIKE-prefix на text_pattern_ops надёжен.
    -- '%' в q.q невозможен (квартал — '\d{2}:\d{2}:\d{6,7}'), escaping не нужен.
    quarter_permits AS (
        SELECT
            fn_normalize_inn(e.developer_inn) AS inn,
            'permits'::text AS source,
            'quarter'::text AS match_method,
            NULL::double precision AS distance_m,
            e.object_name AS project,
            e.permit_type AS status,
            e.object_type AS obj_class
        FROM ekburg_construction_permits e, quarter q
        WHERE q.q IS NOT NULL
          AND e.cadastral_number LIKE q.q || ':%'
    ),
    -- ── Ступень 3b: кадастровый квартал — domrf (latest snapshot per obj_id) ─────────
    quarter_domrf AS (
        SELECT DISTINCT ON (o.obj_id)
            fn_normalize_inn(o.dev_inn) AS inn,
            'domrf'::text AS source,
            'quarter'::text AS match_method,
            NULL::double precision AS distance_m,
            o.comm_name AS project,
            o.site_status AS status,
            o.obj_class
        FROM domrf_kn_objects o, quarter q
        WHERE q.q IS NOT NULL
          AND btrim(o.cad_quarter) = q.q
        ORDER BY o.obj_id, o.snapshot_date DESC
    ),
    -- Объединяем кандидатов всех ступеней. Если сработала ступень exact — caller
    -- получит её первой (distance_m=0); spatial/quarter дополняют контекст соседей.
    all_candidates AS (
        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM exact_match
        UNION ALL
        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM spatial_match
        UNION ALL
        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM quarter_permits
        UNION ALL
        SELECT inn, source, match_method, distance_m, project, status, obj_class FROM quarter_domrf
    )
    SELECT
        c.inn AS developer_inn,
        reg.developer_name,
        c.source,
        c.match_method,
        c.distance_m,
        c.project,
        c.status,
        c.obj_class,
        reg.permits_total,
        reg.domrf_objects_total,
        reg.domrf_active_objects,
        reg.in_both_sources
    FROM all_candidates c
    LEFT JOIN developer_registry reg ON reg.developer_inn = c.inn
    WHERE c.inn IS NOT NULL
    -- детерминированный порядок: лучший метод первым, ближе — раньше, затем ИНН
    ORDER BY
        CASE c.match_method
            WHEN 'exact_cadastral' THEN 1
            WHEN 'spatial_150m'    THEN 2
            WHEN 'quarter'         THEN 3
            ELSE 4
        END,
        c.distance_m ASC NULLS LAST,
        c.inn;
$$;

COMMENT ON FUNCTION fn_developer_for_parcel(text, double precision) IS
    '#1088: резолвер участок→застройщик. 3 ступени (exact_cadastral > spatial_150m > '
    'quarter), геометрия/квартал берутся из cad_parcels_geom по cad_num. Возвращает '
    'SETOF кандидатов с match_method/distance_m + track-record из developer_registry. '
    'Caller берёт топ-1 (ORDER BY method, distance) или ранжирует. p_radius_m=150 по '
    'умолчанию. GIST-friendly spatial (ST_DWithin geom degrees + geography refine).';

COMMIT;