"""Повторный скрейп неизменившегося объявления не переписывает строку (#2992).

Апсерт listings безусловно присваивал 41 колонку при каждом повторном скрейпе —
заново тостил description и плодил TOAST-чанки. Прод 2026-08-20: n_tup_upd
20.87 млн на 107 тыс. строк (198 апдейтов на строку за 91 день), HOT 0.44 %,
TOAST 15 ГБ при ~230 МБ полезного, WAL 7 ГБ/сутки.

Гейт в `ON CONFLICT DO UPDATE … WHERE`: апдейт идёт, если итоговое
(post-COALESCE) значение хоть одной контентной колонки отличается ИЛИ
last_seen_at ещё не сегодняшний по МСК. Второе условие обязательно: метка
живости должна сдвигаться хотя бы раз в сутки — на неё завязаны деактиватор,
снапшоты, эстиматор (#2206), монитор свежести.

Проверяется на ЖИВОМ Postgres через реальный save_listings, в транзакции с
откатом. Счётчик апдейтов — pg_stat_xact_user_tables (текущая транзакция),
то есть ровно «повторный скрейп не увеличивает n_tup_upd» из acceptance.
Без БД — skip (как соседние live-тесты; в CI Trade-In есть postgres-сервис).
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import uuid
from typing import Any
from unittest.mock import MagicMock

import pytest
from scraper_kit.base import ScrapedLot, save_listings
from sqlalchemy import text


def _live_session() -> Any | None:
    try:
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        conn.execute(text("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


def _matcher() -> MagicMock:
    m = MagicMock()
    m.match_or_create_house.return_value = (None, 0.0, "no_address")
    m.upsert_listing_source.return_value = None
    return m


def _lot(
    src_id: str,
    price: int = 5_000_000,
    description: str | None = "тестовое описание",
    address: str | None = None,
) -> ScrapedLot:
    return ScrapedLot(
        source="cian",
        source_url=f"https://ekb.cian.ru/sale/flat/t2992-{src_id}/",
        source_id=f"t2992-{src_id}",
        price_rub=price,
        description=description,
        address=address,
    )


def _row(db: Any, src_id: str) -> Any:
    """Строка + её физическая версия (ctid). save_listings КОММИТИТ, поэтому
    транзакционные счётчики pg_stat_xact_* тут бесполезны (сбрасываются). UPDATE
    всегда создаёт новую версию строки → ctid меняется; пропущенный апдейт → ctid
    тот же. Это и есть прямой замер «переписана ли строка»."""
    return db.execute(
        text(
            "SELECT id, ctid::text AS ctid, price_rub, city, description, last_seen_at "
            "FROM listings WHERE source = 'cian' AND source_id = :sid"
        ),
        {"sid": f"t2992-{src_id}"},
    ).fetchone()


def _cleanup(db: Any) -> None:
    """save_listings коммитит — rollback не спасает. Убираем свои строки явно."""
    try:
        db.rollback()
        db.execute(
            text(
                "DELETE FROM listings_snapshots WHERE listing_id IN "
                "(SELECT id FROM listings WHERE source='cian' AND source_id LIKE 't2992-%')"
            )
        )
        db.execute(text("DELETE FROM listing_sources WHERE ext_id LIKE 't2992-%'"))
        db.execute(text("DELETE FROM listings WHERE source='cian' AND source_id LIKE 't2992-%'"))
        db.commit()
    finally:
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_unchanged_rescrape_same_day_does_not_update_the_row() -> None:
    """Головной: второй апсерт неизменного лота в тот же МСК-день — n_tup_upd не растёт.

    На origin/main (без гейта) второй save_listings переписывает строку — счётчик +1.
    """
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66)
        first = _row(db, sid)
        assert first is not None, "первая вставка не прошла"
        ins, upd = save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66)
        second = _row(db, sid)
        assert second.ctid == first.ctid, (
            f"повторный скрейп неизменного лота переписал строку: ctid {first.ctid}→{second.ctid}"
        )
        assert (ins, upd) == (0, 0), f"счётчики: inserted={ins} updated={upd}, ждали 0/0"
        assert second.last_seen_at == first.last_seen_at, "last_seen_at сдвинулся без нужды"
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_changed_price_still_updates() -> None:
    """Контроль: изменение контента проходит — гейт не глушит настоящие изменения."""
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid, price=5_000_000)], matcher=_matcher(), region_code=66)
        first = _row(db, sid)
        save_listings(db, [_lot(sid, price=5_100_000)], matcher=_matcher(), region_code=66)
        second = _row(db, sid)
        assert second.ctid != first.ctid, "изменение цены не записалось (строка не переписана)"
        assert second.price_rub == 5_100_000
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_coalesce_backfill_still_updates() -> None:
    """Контроль: COALESCE-дозаполнение (город NULL → значение) проходит.

    Именно за это забраковали гейт по card_hash: он бы блокировал дозаполнение.
    Сравнение идёт по ИТОГОВЫМ post-COALESCE значениям — если дописал, строка
    отличается.
    """
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        # city приходит ПАРАМЕТРОМ save_listings (развёртка города), не полем лота.
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66, city=None)
        first = _row(db, sid)
        assert first.city is None
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66, city="Екатеринбург")
        second = _row(db, sid)
        assert second.ctid != first.ctid, "дозаполнение города заблокировано гейтом"
        assert second.city == "Екатеринбург"
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_next_day_rescrape_updates_even_if_unchanged() -> None:
    """Живость: если last_seen_at вчерашний — апдейт идёт и без изменений контента.

    Иначе деактиватор/снапшоты/эстиматор потеряли бы живой инвентарь. Симулируем
    «вчера», отодвинув last_seen_at у только что вставленной строки.
    """
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66)
        db.execute(
            text(
                "UPDATE listings SET last_seen_at = last_seen_at - interval '1 day' "
                "WHERE source='cian' AND source_id = :sid"
            ),
            {"sid": f"t2992-{sid}"},
        )
        db.commit()
        first = _row(db, sid)
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66)
        second = _row(db, sid)
        assert second.ctid != first.ctid, "суточное ре-подтверждение живости заблокировано"
        assert second.last_seen_at > first.last_seen_at, "last_seen_at не сдвинулся"
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_skipped_row_still_yields_listing_id_for_downstream() -> None:
    """При пропуске апдейта listing_id всё равно известен — downstream (снапшот,
    матчинг, listing_sources) идёт; иначе гейт молча отключил бы сопоставление."""
    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    m = _matcher()
    lot = _lot(sid, address="Екатеринбург, ул. Тестовая, 1")  # матчинг идёт только с адресом/гео
    try:
        save_listings(db, [lot], matcher=m, region_code=66)
        calls_before = m.match_or_create_house.call_count
        assert calls_before == 1, (
            "контроль сконструирован неверно: матчинг не зовётся и в первый раз"
        )
        save_listings(db, [lot], matcher=m, region_code=66)
        assert m.match_or_create_house.call_count == calls_before + 1, (
            "при пропущенном апдейте матчинг не вызван — listing_id потерян"
        )
    finally:
        _cleanup(db)


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_listing_sources_unchanged_rescrape_same_day_does_not_update() -> None:
    """Симметричный гейт на listing_sources — второй по нагрузке таблице (10.27 млн
    апдейтов, HOT 0.10 %). Повторный upsert_listing_source того же лота в тот же
    МСК-день не переписывает строку; изменение цены — переписывает."""
    from app.services.matching.listings import upsert_listing_source

    db = _live_session()
    sid = uuid.uuid4().hex[:8]
    try:
        save_listings(db, [_lot(sid)], matcher=_matcher(), region_code=66)
        lid = int(_row(db, sid).id)

        def ls_row() -> Any:
            return db.execute(
                text(
                    "SELECT ctid::text AS ctid, price_rub, last_seen_at FROM listing_sources "
                    "WHERE ext_source='cian' AND ext_id = :e"
                ),
                {"e": f"t2992-{sid}"},
            ).fetchone()

        kw = dict(ext_source="cian", ext_id=f"t2992-{sid}", method="test", confidence=0.9)
        upsert_listing_source(db, listing_id=lid, price_rub=5_000_000, **kw)
        db.commit()
        first = ls_row()
        assert first is not None
        upsert_listing_source(db, listing_id=lid, price_rub=5_000_000, **kw)
        db.commit()
        second = ls_row()
        assert second.ctid == first.ctid, (
            f"listing_sources переписана без изменений: ctid {first.ctid}→{second.ctid}"
        )
        upsert_listing_source(db, listing_id=lid, price_rub=5_100_000, **kw)
        db.commit()
        third = ls_row()
        assert third.ctid != second.ctid and third.price_rub == 5_100_000, (
            "изменение цены не записалось"
        )
    finally:
        _cleanup(db)