-- =============================================================================
-- 92_cad_bulk_layers.sql
-- Bulk Cadastre Ingest — schema migration for GH issue #168
-- =============================================================================
--
-- Context:
--   Creates 7 new cadastre layer tables + 1 saga-state jobs table for
--   bulk NSPD ingest of all 2408 EKB (66:41:*) quarters.
--   See: research/Bulk_Cadastre_Ingest_Plan_May15.md in Obsidian vault.
--
-- Dependencies:
--   - public.cad_quarters_geom      (FK target for cad_quarter_stats)
--   - public.complexes               (FK kept in new cad_buildings.complex_id)
--   - v_complex_full, v_complex_buildings  (views that JOIN cad_buildings — must survive)
--   - PostGIS extension (geometry types)
--
-- Deploy order:
--   1. This file (auto-applied by deploy.yml after merge PR #168 part 1)
--   2. Backend PR #169 (NSPDClient + bulk_harvest_quarter task)
--   3. Frontend PR #170 (admin scrape/cadastre UI)
--
-- Migration steps performed here:
--   A. CREATE TABLE cad_parcels          (new, replaces cad_parcels_geom)
--   B. CREATE TABLE cad_buildings_new    (interim name — wider schema)
--   C. CREATE TABLE cad_constructions
--   D. CREATE TABLE cad_oncs
--   E. CREATE TABLE cad_enk
--   F. CREATE TABLE cad_zouit
--   G. CREATE TABLE cad_quarter_stats
--   H. CREATE TABLE cadastre_jobs        (saga state, mirrors nspd_geo_jobs pattern)
--   I. All indexes (GIST + BTREE)
--   J. Migrate data: cad_parcels_geom -> cad_parcels
--   K. Backup + replace: cad_buildings -> cad_buildings_old_apr26 -> DROP + rename
--   L. Restore FK + views
--   M. Replace cad_parcels_geom TABLE with back-compat VIEW over cad_parcels
--   N. COMMENT ON TABLE for all 8 tables
--   O. set_updated_at triggers for all tables with updated_at column
--
-- Idempotency: safe to re-run — all DDL uses IF NOT EXISTS / IF EXISTS guards.
--   Data migrations use INSERT ... ON CONFLICT DO NOTHING.
--   If cad_buildings_old_apr26 already exists, backup step is skipped.
--
-- Rollback: restore from cad_buildings_old_apr26 manually if needed.
-- =============================================================================

BEGIN;

-- ---------------------------------------------------------------------------
-- A. cad_parcels  (replaces cad_parcels_geom — adds full NSPD props)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_parcels (
    cad_num                             TEXT        NOT NULL,
    quarter_cad_number                  TEXT        NOT NULL,
    category_id                         INT         NOT NULL DEFAULT 36368,
    land_record_area                    NUMERIC,
    specified_area                      NUMERIC,
    declared_area                       NUMERIC,
    land_record_category_type           TEXT,
    land_record_subtype                 TEXT,
    land_record_type                    TEXT,
    permitted_use_established_by_document TEXT,
    cost_value                          NUMERIC,
    cost_index                          NUMERIC,
    cost_application_date               DATE,
    cost_registration_date              DATE,
    cost_determination_date             DATE,
    determination_couse                 TEXT,
    ownership_type                      TEXT,
    right_type                          TEXT,
    readable_address                    TEXT,
    status                              TEXT,
    previously_posted                   TEXT,
    registration_date                   DATE,
    cadastral_districts_code            INT,
    subcategory                         INT,
    geom                                GEOMETRY(Polygon, 4326),
    raw_props                           JSONB,
    source                              TEXT        NOT NULL,
    fetched_at                          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at                          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_parcels_pkey PRIMARY KEY (cad_num)
);

-- ---------------------------------------------------------------------------
-- B. cad_buildings  (wider schema — replaces existing table after data backup)
--    We create under a temporary name first, then swap after backup.
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_buildings_new (
    cad_num                         TEXT        NOT NULL,
    quarter_cad_number              TEXT        NOT NULL,
    objdoc_id                       BIGINT,
    registers_id                    INT         DEFAULT 36441,
    purpose                         TEXT,
    building_name                   TEXT,
    floors                          INT,
    underground_floors              INT,
    area                            NUMERIC,
    build_record_area               NUMERIC,
    cost_value                      NUMERIC,
    cost_index                      NUMERIC,
    year_built                      INT,
    year_commisioning               INT,
    build_record_registration_date  DATE,
    registration_date               DATE,
    ownership_type                  TEXT,
    cultural_heritage_object        BOOLEAN,
    cultural_heritage_val           TEXT,
    readable_address                TEXT,
    status                          TEXT,
    common_data_status              TEXT,
    type                            TEXT,
    build_record_type_value         TEXT,
    complex_id                      BIGINT,
    geom                            GEOMETRY(Geometry, 4326),
    raw_props                       JSONB,
    source                          TEXT        NOT NULL,
    fetched_at                      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at                      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_buildings_new_pkey PRIMARY KEY (cad_num)
);

-- ---------------------------------------------------------------------------
-- C. cad_constructions  (36383 — Сооружения + Сооружения ЕГРН)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_constructions (
    cad_num                     TEXT        NOT NULL,
    quarter_cad_number          TEXT        NOT NULL,
    category_name               TEXT        NOT NULL,
    is_egrn                     BOOLEAN     NOT NULL DEFAULT FALSE,
    params_purpose              TEXT,
    params_name                 TEXT,
    params_area                 NUMERIC,
    params_built_up_area        NUMERIC,
    params_year_built           INT,
    params_year_commisioning    INT,
    params_height               NUMERIC,
    params_volume               NUMERIC,
    params_extension            NUMERIC,
    params_depth                NUMERIC,
    params_occurence_depth      NUMERIC,
    params_floors               INT,
    params_underground_floors   INT,
    object_type_value           TEXT,
    facility_cad_number         TEXT,
    cost_value                  NUMERIC,
    cost_index                  NUMERIC,
    ownership_type              TEXT,
    right_type                  TEXT,
    readable_address            TEXT,
    address_readable_address    TEXT,
    status                      TEXT,
    common_data_status          TEXT,
    registration_date           DATE,
    registers_id                INT,
    united_cad_number           TEXT,
    geom                        GEOMETRY(Geometry, 4326),
    raw_props                   JSONB,
    source                      TEXT        NOT NULL,
    fetched_at                  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at                  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_constructions_pkey PRIMARY KEY (cad_num)
);

-- ---------------------------------------------------------------------------
-- D. cad_oncs  (36384 — Объекты незавершённого строительства)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_oncs (
    cad_num             TEXT        NOT NULL,
    quarter_cad_number  TEXT        NOT NULL,
    purpose             TEXT,
    area                NUMERIC,
    build_record_area   NUMERIC,
    cost_value          NUMERIC,
    status              TEXT,
    readable_address    TEXT,
    geom                GEOMETRY(Geometry, 4326),
    raw_props           JSONB,
    source              TEXT        NOT NULL,
    fetched_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_oncs_pkey PRIMARY KEY (cad_num)
);

-- ---------------------------------------------------------------------------
-- E. cad_enk  (39663 — Единые недвижимые комплексы)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_enk (
    cad_num             TEXT        NOT NULL,
    quarter_cad_number  TEXT        NOT NULL,
    name                TEXT,
    purpose             TEXT,
    type                TEXT,
    cost_value          NUMERIC,
    facility_cad_number TEXT,
    readable_address    TEXT,
    registration_date   DATE,
    registers_id        INT,
    geom                GEOMETRY(Geometry, 4326),
    raw_props           JSONB,
    fetched_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_enk_pkey PRIMARY KEY (cad_num)
);

-- ---------------------------------------------------------------------------
-- F. cad_zouit  (4 categoryIds: 36940 / 469039 / 469040 / 469042)
--    Natural key: (reg_numb_border, category_id) — one border can appear in
--    multiple category subtypes.
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_zouit (
    id                              BIGSERIAL,
    reg_numb_border                 TEXT        NOT NULL,
    category_id                     INT         NOT NULL,
    category_name                   TEXT,
    type_zone                       TEXT        NOT NULL,
    type_boundary_value             TEXT,
    subcategory                     INT,
    cadastral_district              TEXT,
    content_restrict_encumbrances   TEXT,
    name_by_doc                     TEXT,
    legal_act_name                  TEXT,
    legal_act_date                  DATE,
    legal_act_number                TEXT,
    legal_act_issuer                TEXT,
    registration_date               DATE,
    old_account_number              TEXT,
    geom                            GEOMETRY(MultiPolygon, 4326),
    raw_props                       JSONB,
    source                          TEXT        NOT NULL,
    fetched_at                      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at                      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_zouit_pkey PRIMARY KEY (id),
    CONSTRAINT cad_zouit_reg_category_uniq UNIQUE (reg_numb_border, category_id)
);

-- ---------------------------------------------------------------------------
-- G. cad_quarter_stats  (enrichment 1-to-1 with cad_quarters_geom)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cad_quarter_stats (
    cad_number              TEXT        NOT NULL,
    cnt_land                INT,
    cnt_oks                 INT,
    cnt_oks_area            INT,
    cnt_oks_geom            INT,
    cnt_oks_geom_area       INT,
    cnt_oks_geom_linear     INT,
    cnt_oks_linear          INT,
    cnt_oks_not_geom        INT,
    cnt_land_geom           INT,
    cnt_land_not_geom       INT,
    cnt_land_eou            INT,
    cnt_land_geom_eou       INT,
    cnt_enk                 INT,
    cnt_pik                 INT,
    cost_value_total        NUMERIC,
    cost_value_total_eou    NUMERIC,
    cost_value_total_geom   NUMERIC,
    sum_land_area           NUMERIC,
    sum_land_area_eou       NUMERIC,
    sum_land_geom_area      NUMERIC,
    sum_land_geom_area_eou  NUMERIC,
    date_cr                 DATE,
    real_srid               INT,
    raw_props               JSONB,
    fetched_at              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at              TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT cad_quarter_stats_pkey PRIMARY KEY (cad_number),
    CONSTRAINT cad_quarter_stats_quarter_fkey
        FOREIGN KEY (cad_number)
        REFERENCES cad_quarters_geom(cad_number)
        ON DELETE CASCADE
);

-- ---------------------------------------------------------------------------
-- H. cadastre_jobs  (saga state — mirrors nspd_geo_jobs pattern)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS cadastre_jobs (
    job_id          BIGSERIAL   NOT NULL,
    name            TEXT,
    job_kind        TEXT        NOT NULL,
    scope           TEXT        NOT NULL,
    status          TEXT        NOT NULL DEFAULT 'queued',
    triggered_by    TEXT,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    started_at      TIMESTAMPTZ,
    finished_at     TIMESTAMPTZ,
    heartbeat_at    TIMESTAMPTZ,
    targets_total   INT         NOT NULL DEFAULT 0,
    targets_done    INT         NOT NULL DEFAULT 0,
    targets_failed  INT         NOT NULL DEFAULT 0,
    targets_skipped INT         NOT NULL DEFAULT 0,
    requests_count  INT         NOT NULL DEFAULT 0,
    waf_blocked_count INT       NOT NULL DEFAULT 0,
    rate_ms         INT         NOT NULL DEFAULT 333,
    error           TEXT,
    phase_state     JSONB,
    raw_targets     JSONB,
    CONSTRAINT cadastre_jobs_pkey PRIMARY KEY (job_id),
    CONSTRAINT cadastre_jobs_status_chk
        CHECK (status IN (
            'queued', 'running', 'done', 'failed', 'zombie', 'cancelled', 'paused'
        ))
);

-- ===========================================================================
-- I. INDEXES
-- ===========================================================================

-- cad_parcels
CREATE INDEX IF NOT EXISTS cad_parcels_geom_gist
    ON cad_parcels USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_parcels_quarter_idx
    ON cad_parcels USING BTREE (quarter_cad_number);

-- cad_buildings_new
CREATE INDEX IF NOT EXISTS cad_buildings_new_geom_gist
    ON cad_buildings_new USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_buildings_new_quarter_idx
    ON cad_buildings_new USING BTREE (quarter_cad_number);
CREATE INDEX IF NOT EXISTS cad_buildings_new_objdoc_idx
    ON cad_buildings_new USING BTREE (objdoc_id)
    WHERE objdoc_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS cad_buildings_new_complex_idx
    ON cad_buildings_new USING BTREE (complex_id)
    WHERE complex_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS cad_buildings_new_purpose_idx
    ON cad_buildings_new USING BTREE (purpose);

-- cad_constructions
CREATE INDEX IF NOT EXISTS cad_constructions_geom_gist
    ON cad_constructions USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_constructions_quarter_idx
    ON cad_constructions USING BTREE (quarter_cad_number);
CREATE INDEX IF NOT EXISTS cad_constructions_purpose_idx
    ON cad_constructions USING BTREE (params_purpose)
    WHERE params_purpose IS NOT NULL;

-- cad_oncs
CREATE INDEX IF NOT EXISTS cad_oncs_geom_gist
    ON cad_oncs USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_oncs_quarter_idx
    ON cad_oncs USING BTREE (quarter_cad_number);

-- cad_enk
CREATE INDEX IF NOT EXISTS cad_enk_geom_gist
    ON cad_enk USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_enk_quarter_idx
    ON cad_enk USING BTREE (quarter_cad_number);

-- cad_zouit
CREATE INDEX IF NOT EXISTS cad_zouit_geom_gist
    ON cad_zouit USING GIST (geom);
CREATE INDEX IF NOT EXISTS cad_zouit_reg_numb_idx
    ON cad_zouit USING BTREE (reg_numb_border);
CREATE INDEX IF NOT EXISTS cad_zouit_category_idx
    ON cad_zouit USING BTREE (category_id);
CREATE INDEX IF NOT EXISTS cad_zouit_subcategory_idx
    ON cad_zouit USING BTREE (subcategory)
    WHERE subcategory IS NOT NULL;

-- cadastre_jobs
CREATE INDEX IF NOT EXISTS cadastre_jobs_status_idx
    ON cadastre_jobs USING BTREE (status, created_at DESC);
CREATE INDEX IF NOT EXISTS cadastre_jobs_running_idx
    ON cadastre_jobs USING BTREE (heartbeat_at DESC)
    WHERE status = 'running';

-- ===========================================================================
-- J. Migrate existing rows: cad_parcels_geom -> cad_parcels
--    Source has: cad_num, geom, raw_props, fetched_at
--    We set quarter_cad_number from cad_num (first 3 parts: XX:YY:ZZZZZZ)
--    source = 'nspd' (original scraper)
-- ===========================================================================
INSERT INTO cad_parcels (
    cad_num,
    quarter_cad_number,
    geom,
    raw_props,
    source,
    fetched_at,
    updated_at
)
SELECT
    p.cad_num,
    -- Derive quarter from cad_num "66:41:0101001:123" -> "66:41:0101001"
    regexp_replace(p.cad_num, '^(\d+:\d+:\d+):.*$', '\1'),
    p.geom,
    p.raw_props,
    'nspd',
    p.fetched_at,
    p.fetched_at
FROM cad_parcels_geom p
ON CONFLICT (cad_num) DO NOTHING;

-- ===========================================================================
-- K. Backup existing cad_buildings + migrate data into cad_buildings_new
--    then drop old and rename new.
--
--    Old schema columns we map to new:
--      quarter_cad_num     -> quarter_cad_number
--      floors TEXT         -> floors INT (NULLIF on non-numeric)
--      underground_floors TEXT -> underground_floors INT
--      cultural_heritage TEXT  -> cultural_heritage_object BOOLEAN (NOT NULL = FALSE → NULL),
--                                  cultural_heritage_val TEXT
--      build_record_type   -> build_record_type_value
--      obj_type            -> type
--      (no source in old table, default 'nspd')
-- ===========================================================================

-- K.1  Backup: rename old table (only if backup doesn't already exist)
DO $$
BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM information_schema.tables
        WHERE table_schema = 'public'
          AND table_name = 'cad_buildings_old_apr26'
    ) THEN
        ALTER TABLE cad_buildings RENAME TO cad_buildings_old_apr26;
    END IF;
END $$;

-- K.2  Migrate data into cad_buildings_new
--      Old cad_buildings (now _old_apr26) -> new wider schema.
--      complex_id FK preserved as-is.
INSERT INTO cad_buildings_new (
    cad_num,
    quarter_cad_number,
    purpose,
    building_name,
    floors,
    underground_floors,
    area,
    build_record_area,
    cost_value,
    year_built,
    year_commisioning,
    registration_date,
    ownership_type,
    cultural_heritage_object,
    cultural_heritage_val,
    readable_address,
    status,
    common_data_status,
    type,
    build_record_type_value,
    complex_id,
    geom,
    raw_props,
    source,
    fetched_at,
    updated_at
)
SELECT
    o.cad_num,
    COALESCE(o.quarter_cad_num,
        regexp_replace(o.cad_num, '^(\d+:\d+:\d+):.*$', '\1')),
    o.purpose,
    o.building_name,
    -- floors was TEXT in old schema; cast safely
    CASE
        WHEN o.floors ~ '^\d+$' THEN o.floors::INT
        ELSE NULL
    END,
    -- underground_floors was TEXT
    CASE
        WHEN o.underground_floors ~ '^\d+$' THEN o.underground_floors::INT
        ELSE NULL
    END,
    o.area,
    o.build_record_area,
    o.cost_value,
    o.year_built,
    o.year_commisioning,
    o.registration_date,
    o.ownership_type,
    -- cultural_heritage TEXT: treat non-NULL / non-empty string as TRUE
    CASE
        WHEN o.cultural_heritage IS NOT NULL
             AND o.cultural_heritage <> '' THEN TRUE
        ELSE NULL
    END,
    o.cultural_heritage,
    o.readable_address,
    o.status,
    o.common_data_status,
    o.obj_type,
    o.build_record_type,
    o.complex_id,
    o.geom,
    o.raw_props,
    'nspd',
    o.fetched_at,
    o.fetched_at
FROM cad_buildings_old_apr26 o
ON CONFLICT (cad_num) DO NOTHING;

-- K.3  Rename cad_buildings_new -> cad_buildings
--      (only if the rename hasn't happened already)
DO $$
BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM information_schema.tables
        WHERE table_schema = 'public' AND table_name = 'cad_buildings'
    ) THEN
        ALTER TABLE cad_buildings_new RENAME TO cad_buildings;
        -- First, retire the legacy PK index name that's still attached to
        -- cad_buildings_old_apr26 (PG does NOT auto-rename PK indexes on
        -- table rename — pre-K.1 backup left cad_buildings_pkey behind).
        -- Without this rename, the next line collides with the existing
        -- index name → 'relation already exists' → tx rollback.
        ALTER INDEX IF EXISTS cad_buildings_pkey              RENAME TO cad_buildings_old_apr26_pkey;
        -- Rename indexes to drop the _new suffix
        ALTER INDEX IF EXISTS cad_buildings_new_pkey          RENAME TO cad_buildings_pkey;
        -- Note: ALTER INDEX RENAME on a PK index also renames the constraint
        --       automatically in PG 16. No separate RENAME CONSTRAINT needed.
        ALTER INDEX IF EXISTS cad_buildings_new_geom_gist     RENAME TO cad_buildings_geom_gist;
        ALTER INDEX IF EXISTS cad_buildings_new_quarter_idx   RENAME TO cad_buildings_quarter_idx;
        ALTER INDEX IF EXISTS cad_buildings_new_objdoc_idx    RENAME TO cad_buildings_objdoc_idx;
        ALTER INDEX IF EXISTS cad_buildings_new_complex_idx   RENAME TO cad_buildings_complex_idx;
        ALTER INDEX IF EXISTS cad_buildings_new_purpose_idx   RENAME TO cad_buildings_purpose_idx;
    END IF;
END $$;

-- ---------------------------------------------------------------------------
-- L.1  Re-add FK: cad_buildings.complex_id -> complexes.id
--      (was dropped implicitly when old table was renamed)
--      Guard: only add if FK doesn't already exist.
-- ---------------------------------------------------------------------------
DO $$
BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM information_schema.table_constraints
        WHERE table_schema = 'public'
          AND table_name = 'cad_buildings'
          AND constraint_name = 'cad_buildings_complex_id_fkey'
          AND constraint_type = 'FOREIGN KEY'
    ) THEN
        ALTER TABLE cad_buildings
            ADD CONSTRAINT cad_buildings_complex_id_fkey
            FOREIGN KEY (complex_id)
            REFERENCES complexes(id)
            ON DELETE SET NULL;
    END IF;
END $$;

-- ---------------------------------------------------------------------------
-- L.2  Restore views that JOIN cad_buildings.
--      Both views only reference: cad_num, complex_id, canonical_name.
--      These columns exist in new cad_buildings — views will still be valid
--      after the rename. However, we drop + recreate explicitly to ensure
--      pg_depend is attached to the renamed table (not the old OID).
-- ---------------------------------------------------------------------------
DROP VIEW IF EXISTS v_complex_buildings;
CREATE OR REPLACE VIEW v_complex_buildings AS
SELECT
    c.id                                                    AS complex_id,
    c.canonical_name,
    array_agg(DISTINCT b.cad_num ORDER BY b.cad_num)       AS cad_building_nums,
    COUNT(b.cad_num)                                        AS buildings_count
FROM complexes c
LEFT JOIN cad_buildings b ON b.complex_id = c.id
GROUP BY c.id, c.canonical_name;

DROP VIEW IF EXISTS v_complex_full;
CREATE OR REPLACE VIEW v_complex_full AS
SELECT
    c.id                        AS complex_id,
    c.canonical_name,
    c.developer_name,
    c.region_id,
    c.district_name,
    c.obj_class,
    c.address,
    c.latitude,
    c.longitude,
    (c.geom IS NOT NULL)        AS has_geom,
    c.flat_count,
    c.square_living,
    c.site_status,
    c.cad_quarter,
    (
        SELECT array_agg(b.cad_num ORDER BY b.cad_num)
        FROM cad_buildings b
        WHERE b.complex_id = c.id
    )                           AS cad_building_nums,
    (
        SELECT COUNT(*)
        FROM cad_buildings b
        WHERE b.complex_id = c.id
    )                           AS cad_buildings_n,
    (
        SELECT array_agg(DISTINCT cs.source ORDER BY cs.source)
        FROM complex_sources cs
        WHERE cs.complex_id = c.id
    )                           AS sources,
    (
        SELECT COUNT(*)
        FROM complex_sources cs
        WHERE cs.complex_id = c.id
    )                           AS source_links_n,
    (
        SELECT COUNT(DISTINCT cs.source)
        FROM complex_sources cs
        WHERE cs.complex_id = c.id
    )                           AS source_kinds_n,
    (
        SELECT array_agg(cs.source_external_id ORDER BY cs.source_external_id)
        FROM complex_sources cs
        WHERE cs.complex_id = c.id
          AND cs.source = 'domrf_kn'
    )                           AS domrf_obj_ids,
    (
        SELECT cs.source_id
        FROM complex_sources cs
        WHERE cs.complex_id = c.id
          AND cs.source = 'objective'
        LIMIT 1
    )                           AS objective_project_name,
    (
        SELECT COUNT(*)
        FROM domrf_kn_flats f
        WHERE f.obj_id IN (
            SELECT cs.source_external_id
            FROM complex_sources cs
            WHERE cs.complex_id = c.id
              AND cs.source = 'domrf_kn'
              AND cs.source_external_id IS NOT NULL
        )
    )                           AS kn_flats_n,
    (
        SELECT COUNT(*)
        FROM objective_lots ol
        WHERE ol.complex_id = c.id
    )                           AS objective_lots_n,
    c.primary_source,
    c.created_at,
    c.updated_at
FROM complexes c;

-- ===========================================================================
-- M. Replace cad_parcels_geom TABLE with a back-compat VIEW over cad_parcels.
--    Original table had exactly: cad_num, geom, raw_props, fetched_at.
--    The view preserves the same 4-column signature so that any legacy callers
--    (e.g. site-finder analyze fallback) continue to work without code changes.
--    Bulk PR2/5 will switch callers to cad_parcels directly, after which this
--    view can be dropped.
-- ===========================================================================
DO $$
BEGIN
    -- Drop the old table only if cad_parcels (migration target) already exists.
    -- Guard prevents accidental drop if J INSERT was skipped on re-run.
    IF EXISTS (
        SELECT 1 FROM information_schema.tables
        WHERE table_schema = 'public' AND table_name = 'cad_parcels'
    ) AND EXISTS (
        SELECT 1 FROM information_schema.tables
        WHERE table_schema = 'public'
          AND table_name = 'cad_parcels_geom'
          AND table_type = 'BASE TABLE'
    ) THEN
        DROP TABLE cad_parcels_geom;
    END IF;
END $$;

-- Create back-compat VIEW (idempotent via CREATE OR REPLACE).
-- Matches original column order and types exactly.
CREATE OR REPLACE VIEW cad_parcels_geom AS
    SELECT
        cad_num,
        geom,
        raw_props,
        fetched_at
    FROM cad_parcels;

COMMENT ON VIEW cad_parcels_geom IS
    'Back-compat shim: preserves original 4-column signature (cad_num, geom, '
    'raw_props, fetched_at) for legacy site-finder analyze fallback. '
    'Bulk PR2/5 will migrate callers to cad_parcels directly; drop this view then.';

-- ===========================================================================
-- N. COMMENT ON TABLE for all 8 tables
-- ===========================================================================
COMMENT ON TABLE cad_parcels IS
    'Земельные участки ЕГРН из NSPD (categoryId 36368, EKB кадастровый округ 66:41:*). '
    'Широкая схема с полными атрибутами NSPD + геометрия. Заменяет cad_parcels_geom.';

COMMENT ON TABLE cad_buildings IS
    'Здания ЕГРН из NSPD (categoryId 36369). Широкая схема: objdoc_id для '
    'tab-group-data lookup, complex_id FK на complexes, cultural_heritage_object BOOLEAN.';

COMMENT ON TABLE cad_constructions IS
    'Сооружения из NSPD (categoryId 36383). Объединяет Сооружения и Сооружения ЕГРН '
    '(is_egrn=TRUE). Полный набор технических параметров (высота, объём, протяжённость).';

COMMENT ON TABLE cad_oncs IS
    'Объекты незавершённого строительства из NSPD (categoryId 36384, EKB 66:41:*).';

COMMENT ON TABLE cad_enk IS
    'Единые недвижимые комплексы из NSPD (categoryId 39663, EKB 66:41:*).';

COMMENT ON TABLE cad_zouit IS
    'Зоны с особыми условиями использования территории из NSPD '
    '(categoryIds 36940 / 469039 / 469040 / 469042, все в одной таблице). '
    'Натуральный ключ: (reg_numb_border, category_id) — один border может входить '
    'в несколько категорий.';

COMMENT ON TABLE cad_quarter_stats IS
    'Агрегированная статистика кадастрового квартала (1-to-1 с cad_quarters_geom). '
    'Заполняется bulk-harvest worker''ом: счётчики объектов, суммы площадей, кадастровая '
    'стоимость, real_srid ответа NSPD.';

COMMENT ON TABLE cadastre_jobs IS
    'Resumable saga state для bulk harvest runs (один job = один кадастровый квартал '
    'или batch). Зеркалит паттерн nspd_geo_jobs. heartbeat_at обновляется worker''ом '
    'каждые N секунд; zombie-детектор переводит зависшие jobs в статус zombie.';

-- ===========================================================================
-- O. set_updated_at triggers for all tables with updated_at column.
--    set_updated_at() function already exists in this DB (verified).
--    CREATE OR REPLACE TRIGGER is not available in PG 16 — use DO guard.
-- ===========================================================================

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_parcels_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_parcels_set_updated_at
            BEFORE UPDATE ON cad_parcels
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_buildings_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_buildings_set_updated_at
            BEFORE UPDATE ON cad_buildings
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_constructions_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_constructions_set_updated_at
            BEFORE UPDATE ON cad_constructions
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_oncs_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_oncs_set_updated_at
            BEFORE UPDATE ON cad_oncs
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_enk_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_enk_set_updated_at
            BEFORE UPDATE ON cad_enk
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_zouit_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_zouit_set_updated_at
            BEFORE UPDATE ON cad_zouit
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

DO $$ BEGIN
    IF NOT EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgname = 'cad_quarter_stats_set_updated_at'
    ) THEN
        CREATE TRIGGER cad_quarter_stats_set_updated_at
            BEFORE UPDATE ON cad_quarter_stats
            FOR EACH ROW EXECUTE FUNCTION set_updated_at();
    END IF;
END $$;

-- cadastre_jobs intentionally excluded — heartbeat_at is the liveness mechanism,
-- not updated_at (which the table does not have).

COMMIT;