-- 094_cadastral_unify.sql
-- Purpose: Унификация имени кадастрового столбца kadastr_num → cadastral_number
--          во всех таблицах tradein-БД.
--
-- Контекст (Issue #732):
--   deals.kadastr_num          → RENAME TO cadastral_number
--   house_metadata.kadastr_num → RENAME TO cadastral_number
--   listings.kadastr_num       → DROP (данные 0%; listings.cadastral_number уже существует
--                                       — добавлен в 019_listings_alter_cian.sql)
--   listings_search_mv         → DROP + RECREATE (ссылался на l.kadastr_num → l.cadastral_number)
--   v_data_quality             → DROP + RECREATE (ссылался на kadastr_num в active_listings CTE)
--
-- Idempotency:
--   * RENAME: обёрнуты в DO $$ ... $$ блок с проверкой information_schema.columns.
--     Повторный запуск: столбец уже переименован → IF NOT EXISTS → skip.
--   * DROP/DROP IF EXISTS: безопасны при повторном запуске.
--   * RECREATE: CREATE OR REPLACE VIEW / CREATE MATERIALIZED VIEW после DROP IF EXISTS.
--
-- ВНИМАНИЕ: DESTRUCTIVE DDL. Применять только после проверки данных на проде.
--   AUTO-APPLIES ON DEPLOY — merge-gate требуется (post-demo).
--
-- Dependencies:
--   002_core_tables.sql (deals, listings, house_metadata)
--   019_listings_alter_cian.sql (listings.cadastral_number already exists)
--   046_views.sql (v_data_quality, v_price_divergence)
--   050_search_optimization.sql (listings_search_mv)

BEGIN;

-- =============================================================================
-- 1. Drop dependent objects (view + materialized view)
--    Нужно до RENAME/DROP колонок, иначе PG откажет (зависимые объекты).
-- =============================================================================

DROP MATERIALIZED VIEW IF EXISTS listings_search_mv;
DROP VIEW IF EXISTS v_data_quality;

-- =============================================================================
-- 2. Rename deals.kadastr_num → cadastral_number (idempotent)
-- =============================================================================

DO $$
BEGIN
    IF EXISTS (
        SELECT 1 FROM information_schema.columns
         WHERE table_schema = 'public'
           AND table_name   = 'deals'
           AND column_name  = 'kadastr_num'
    ) THEN
        ALTER TABLE deals RENAME COLUMN kadastr_num TO cadastral_number;
    END IF;
END
$$;

-- =============================================================================
-- 3. Rename house_metadata.kadastr_num → cadastral_number (idempotent)
--    house_metadata.kadastr_num имеет UNIQUE constraint — переименование сохраняет его.
-- =============================================================================

DO $$
BEGIN
    IF EXISTS (
        SELECT 1 FROM information_schema.columns
         WHERE table_schema = 'public'
           AND table_name   = 'house_metadata'
           AND column_name  = 'kadastr_num'
    ) THEN
        ALTER TABLE house_metadata RENAME COLUMN kadastr_num TO cadastral_number;
    END IF;
END
$$;

-- =============================================================================
-- 4. Удаляем listings.kadastr_num + его индекс
--    listings.cadastral_number (кадастровый номер квартиры, из Cian) уже существует
--    с 019_listings_alter_cian.sql — данные в kadastr_num 0%, копировать не нужно.
-- =============================================================================

DROP INDEX IF EXISTS listings_kadastr_idx;

DO $$
BEGIN
    IF EXISTS (
        SELECT 1 FROM information_schema.columns
         WHERE table_schema = 'public'
           AND table_name   = 'listings'
           AND column_name  = 'kadastr_num'
    ) THEN
        ALTER TABLE listings DROP COLUMN kadastr_num;
    END IF;
END
$$;

-- =============================================================================
-- 5. Recreate v_data_quality
--    Изменение: kadastr_num → cadastral_number (listings.cadastral_number из 019).
--    CREATE OR REPLACE VIEW — идемпотентно.
-- =============================================================================

CREATE OR REPLACE VIEW v_data_quality AS
WITH active_listings AS (
    SELECT * FROM listings WHERE is_active = true
)
SELECT
    (SELECT count(*) FROM houses)                                                  AS houses_total,
    (SELECT count(*) FROM houses h
       WHERE EXISTS (SELECT 1 FROM house_sources hs WHERE hs.house_id = h.id))      AS houses_with_source,
    (SELECT count(*) FROM houses h
       WHERE EXISTS (SELECT 1 FROM house_sources hs
                      WHERE hs.house_id = h.id AND hs.ext_source = 'avito'))        AS houses_with_avito,
    (SELECT count(*) FROM houses h
       WHERE EXISTS (SELECT 1 FROM house_sources hs
                      WHERE hs.house_id = h.id AND hs.ext_source LIKE 'cian%'))     AS houses_with_cian,
    (SELECT count(*) FROM houses h
       WHERE EXISTS (SELECT 1 FROM house_sources hs
                      WHERE hs.house_id = h.id AND hs.ext_source = 'yandex'))       AS houses_with_yandex,
    (SELECT count(*) FROM (
        SELECT house_id FROM house_sources GROUP BY house_id HAVING count(*) >= 2
    ) sub)                                                                          AS houses_2plus_sources,
    (SELECT count(*) FROM (
        SELECT house_id FROM house_sources GROUP BY house_id HAVING count(*) >= 3
    ) sub)                                                                          AS houses_3plus_sources,
    (SELECT count(*) FROM active_listings)                                         AS listings_active,
    (SELECT count(*) FROM (
        SELECT listing_id FROM listing_sources
         WHERE listing_id IN (SELECT id FROM active_listings)
         GROUP BY listing_id HAVING count(*) >= 2
    ) sub)                                                                          AS listings_dedup_2sources,
    (SELECT count(*) FROM active_listings WHERE lat IS NOT NULL) * 100.0
        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_geocoded,
    (SELECT count(*) FROM active_listings WHERE cadastral_number IS NOT NULL) * 100.0
        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_cadastr,
    (SELECT count(*) FROM active_listings WHERE description IS NOT NULL) * 100.0
        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_description,
    (SELECT count(*) FROM active_listings l
       JOIN houses h ON h.id = l.house_id_fk
       WHERE h.year_built IS NOT NULL) * 100.0
        / NULLIF((SELECT count(*) FROM active_listings), 0)                         AS pct_year_built,
    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'avito')           AS avito_last_scrape_ago,
    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'cian')            AS cian_last_scrape_ago,
    NOW() - (SELECT max(scraped_at) FROM listings WHERE source = 'yandex')          AS yandex_last_scrape_ago,
    (SELECT count(*) FROM v_price_divergence)                                       AS price_disagreements_count,
    (SELECT count(*) FROM listings WHERE is_outlier = true)                         AS outliers_flagged;

COMMENT ON VIEW v_data_quality IS
  'KPI snapshot. Refreshed on-demand by /api/v1/admin/data-quality endpoint (Master Plan sec 8.1).';

-- =============================================================================
-- 6. Recreate listings_search_mv
--    Изменение: l.kadastr_num → l.cadastral_number (apartment-level из 019_listings_alter_cian).
--    Все 6 индексов воссоздаются (см. 050_search_optimization.sql).
-- =============================================================================

CREATE MATERIALIZED VIEW listings_search_mv AS
SELECT
    l.id                AS listing_id,
    l.source,
    l.source_url,
    l.address,
    l.geom,
    l.lat,
    l.lon              AS lng,
    l.rooms,
    l.area_m2          AS total_area,
    l.floor,
    l.total_floors,
    l.price_rub,
    l.price_per_m2,
    l.cadastral_number,
    l.is_active,
    l.scraped_at,
    -- House denorm
    h.id               AS house_id,
    h.year_built,
    h.house_class,
    h.developer_name,
    h.rating           AS house_rating,
    h.reviews_count    AS house_ratings_count,
    -- Cross-source aggregates
    (SELECT count(*)               FROM listing_sources ls WHERE ls.listing_id = l.id) AS source_count,
    (SELECT array_agg(DISTINCT ext_source) FROM listing_sources ls WHERE ls.listing_id = l.id) AS sources,
    (SELECT bool_or(ext_source = 'avito')         FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_avito,
    (SELECT bool_or(ext_source = 'cian')          FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_cian,
    (SELECT bool_or(ext_source = 'yandex_realty') FROM listing_sources ls WHERE ls.listing_id = l.id) AS has_yandex,
    -- Price percentile within house
    (SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY ll.price_per_m2)
       FROM listings ll
      WHERE ll.house_id_fk = l.house_id_fk AND ll.is_active = true) AS house_median_ppm2,
    -- DEFAULT: district / distance_to_metro_m / last_price_change / photos_count
    -- not present in current schema — placeholders.
    NULL::text                          AS district,
    NULL::int                           AS distance_to_metro_m,
    NULL::timestamptz                   AS last_price_change,
    NULL::int                           AS photos_count,
    -- Trigram-ready columns
    l.address                           AS address_trgm,
    -- Aggregated tsv (description + address + developer_name)
    to_tsvector('russian',
        coalesce(l.description, '') || ' ' ||
        coalesce(l.address, '')     || ' ' ||
        coalesce(h.developer_name, '')
    ) AS tsv
FROM listings l
LEFT JOIN houses h ON h.id = l.house_id_fk
WHERE l.is_active = true
  AND COALESCE(l.canonical, true) = true;

-- 6 matview indexes (воссоздаются без IF NOT EXISTS — MV была только что создана)
CREATE UNIQUE INDEX listings_search_mv_id_idx
  ON listings_search_mv (listing_id);

CREATE INDEX listings_search_mv_geom_idx
  ON listings_search_mv USING GIST (geom);

CREATE INDEX listings_search_mv_filters_idx
  ON listings_search_mv (rooms, price_rub, total_area, scraped_at DESC);

CREATE INDEX listings_search_mv_address_trgm_idx
  ON listings_search_mv USING GIN (address_trgm gin_trgm_ops);

CREATE INDEX listings_search_mv_tsv_idx
  ON listings_search_mv USING GIN (tsv);

CREATE INDEX listings_search_mv_sources_idx
  ON listings_search_mv (has_avito, has_cian, has_yandex);

-- Populate materialized view (первый запуск — CONCURRENTLY недоступен на новой MV).
REFRESH MATERIALIZED VIEW listings_search_mv;

COMMIT;