"""ЭТАП 4 B2C launch — right-to-erasure mechanism (part C).

Covers app/services/data_erasure.py:
  - at least one identifier required (ValueError, no db.execute at all)
  - username (B2B pilot): estimates + their leads + web_support_threads deleted
  - estimate_ids only (anonymous, has the link/PDF): estimates + linked leads deleted,
    web_support/tg_support untouched (no username = nothing to key them by)
  - phone only: only leads deleted (no estimate/support action)
  - tg_chat_id only: only tg_support deleted (anonymous Telegram-support path)
  - ORDER: leads are captured/deleted BEFORE estimates (estimate_id FK is
    ON DELETE SET NULL -- deleting estimates first would orphan the join)
  - commits once at the end
"""

from __future__ import annotations

import os
import re
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock
from uuid import uuid4

import pytest

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

from app.services import data_erasure


class _Result:
    def __init__(self, rowcount: int = 0, scalar_ids: list[Any] | None = None) -> None:
        self.rowcount = rowcount
        self._scalar_ids = scalar_ids or []

    def scalars(self) -> SimpleNamespace:
        return SimpleNamespace(all=lambda: self._scalar_ids)


def _sql_of(call: Any) -> str:
    stmt = call.args[0]
    return str(getattr(stmt, "text", stmt))


def test_requires_at_least_one_identifier() -> None:
    db = MagicMock()
    with pytest.raises(ValueError, match="at least one identifier"):
        data_erasure.erase_person_data(db)
    assert not db.execute.called
    assert not db.commit.called


def test_erase_by_username_deletes_estimates_leads_and_web_support() -> None:
    db = MagicMock()
    owned_id = uuid4()
    db.execute.side_effect = [
        _Result(scalar_ids=[owned_id]),  # SELECT id FROM trade_in_estimates WHERE created_by
        _Result(rowcount=2),  # DELETE FROM trade_in_leads
        _Result(rowcount=1),  # DELETE FROM trade_in_estimates
        _Result(rowcount=3),  # DELETE FROM web_support_threads
    ]

    out = data_erasure.erase_person_data(db, username="kopylov")

    assert out == {
        "trade_in_estimates_deleted": 1,
        "trade_in_leads_deleted": 2,
        "web_support_deleted": 3,
        "tg_support_deleted": 0,
    }
    assert db.commit.called

    calls = db.execute.call_args_list
    assert "SELECT id FROM trade_in_estimates" in _sql_of(calls[0])
    assert "created_by" in _sql_of(calls[0])
    assert "DELETE FROM trade_in_leads" in _sql_of(calls[1])
    assert "DELETE FROM trade_in_estimates" in _sql_of(calls[2])
    assert "DELETE FROM web_support_threads" in _sql_of(calls[3])
    # estimate_ids captured from the SELECT reach the estimates DELETE.
    estimates_delete_params = calls[2].args[1]
    assert str(owned_id) in estimates_delete_params["ids"]


def test_leads_deleted_before_estimates_order() -> None:
    """FK trade_in_leads.estimate_id is ON DELETE SET NULL -- capturing/deleting
    leads must happen BEFORE the estimates DELETE, else the join key is gone."""
    db = MagicMock()
    owned_id = uuid4()
    db.execute.side_effect = [
        _Result(scalar_ids=[owned_id]),
        _Result(rowcount=0),
        _Result(rowcount=1),
        _Result(rowcount=0),
    ]
    data_erasure.erase_person_data(db, username="kopylov")
    calls = db.execute.call_args_list
    leads_idx = next(i for i, c in enumerate(calls) if "DELETE FROM trade_in_leads" in _sql_of(c))
    estimates_idx = next(
        i for i, c in enumerate(calls) if "DELETE FROM trade_in_estimates" in _sql_of(c)
    )
    assert leads_idx < estimates_idx


def test_erase_by_estimate_ids_only_no_web_or_tg_support_touched() -> None:
    db = MagicMock()
    eid = uuid4()
    db.execute.side_effect = [
        _Result(rowcount=1),  # DELETE FROM trade_in_leads (matches estimate_id)
        _Result(rowcount=1),  # DELETE FROM trade_in_estimates
    ]

    out = data_erasure.erase_person_data(db, estimate_ids=[eid])

    assert out == {
        "trade_in_estimates_deleted": 1,
        "trade_in_leads_deleted": 1,
        "web_support_deleted": 0,
        "tg_support_deleted": 0,
    }
    assert db.execute.call_count == 2  # no username -> no SELECT, no web_support DELETE


def test_erase_by_phone_only_touches_only_leads() -> None:
    db = MagicMock()
    db.execute.side_effect = [_Result(rowcount=1)]  # DELETE FROM trade_in_leads WHERE phone=...

    out = data_erasure.erase_person_data(db, phone="+79123456789")

    assert out == {
        "trade_in_estimates_deleted": 0,
        "trade_in_leads_deleted": 1,
        "web_support_deleted": 0,
        "tg_support_deleted": 0,
    }
    assert db.execute.call_count == 1
    params = db.execute.call_args_list[0].args[1]
    assert params["phone"] == "+79123456789"
    assert params["ids"] == []


def test_phone_delete_normalizes_digits_on_both_sides() -> None:
    """Regression guard for the deep-review MEDIUM finding (2026-08-06) +
    follow-up (RU 8-vs-7 trunk prefix): lead.py stores phone exactly as typed
    (no E.164 normalization, by design), so a differently-formatted-but-
    same-number erasure request must still match, AND the RU '8...' vs
    '+7...' trunk-prefix pair must collapse to the same canonical value. The
    old exact `phone = :phone` comparison silently deleted 0 rows and still
    returned HTTP 200 'erased' -- worse than an honest error under 152-ФЗ.
    Both sides must go through the SAME normalization (_PHONE_COLUMN_NORM_SQL
    / _PHONE_PARAM_NORM_SQL, see _ru_phone_norm_sql), and the literal-equality
    path must be gone."""
    db = MagicMock()
    db.execute.side_effect = [_Result(rowcount=1)]

    data_erasure.erase_person_data(db, phone="+7 999 123-45-67")

    sql = _sql_of(db.execute.call_args_list[0])
    # The comparison uses EXACTLY the two module-level normalized fragments
    # (not a hand-rolled inline duplicate) -- pins that both sides go through
    # the SAME normalization function, not two independently-drifting copies.
    col_norm = data_erasure._PHONE_COLUMN_NORM_SQL
    param_norm = data_erasure._PHONE_PARAM_NORM_SQL
    assert f"{col_norm} = {param_norm}" in sql
    # RU trunk-prefix collapse present on BOTH sides (11 digits, leading '8' -> '7').
    assert sql.count("length(regexp_replace") == 2
    assert sql.count("= '8'") == 2
    assert sql.count("'7' ||") == 2
    assert "phone = :phone" not in sql  # old literal-equality path must be GONE
    assert not re.search(r":\w+::", sql)  # psycopg v3 CAST trap


def test_ru_phone_norm_sql_only_ever_takes_static_expressions() -> None:
    """`_ru_phone_norm_sql` is a query-STRUCTURE builder, not a data path --
    the two module-level constants are the ONLY call sites, and both pass a
    column name / CAST(:bind AS type), never an actual phone value. This
    pins that contract so a future call site can't accidentally splice a
    real phone string into the SQL text."""
    assert data_erasure._PHONE_COLUMN_NORM_SQL == data_erasure._ru_phone_norm_sql("phone")
    assert data_erasure._PHONE_PARAM_NORM_SQL == data_erasure._ru_phone_norm_sql(
        "CAST(:phone AS text)"
    )
    # column side references the column, never the bind param; param side is the reverse.
    assert ":phone" not in data_erasure._PHONE_COLUMN_NORM_SQL
    assert "CAST(:phone AS text)" in data_erasure._PHONE_PARAM_NORM_SQL


def test_erase_by_tg_chat_id_only_touches_only_tg_support() -> None:
    """Anonymous person with NO username, NO estimate link, NO lead phone -- but
    they DID message @MERAsupport_bot -- can still be identified by their own
    Telegram chat_id (see module docstring: not spoofable by a third party)."""
    db = MagicMock()
    db.execute.side_effect = [
        _Result(rowcount=0),  # DELETE FROM trade_in_leads (no ids, no phone -> matches nothing)
        _Result(rowcount=5),  # DELETE FROM tg_support_users
    ]

    out = data_erasure.erase_person_data(db, tg_chat_id=123456789)

    assert out == {
        "trade_in_estimates_deleted": 0,
        "trade_in_leads_deleted": 0,
        "web_support_deleted": 0,
        "tg_support_deleted": 5,
    }
    calls = db.execute.call_args_list
    assert "DELETE FROM tg_support_users" in _sql_of(calls[-1])
    assert calls[-1].args[1]["chat_id"] == 123456789


# ── Optional real-Postgres behavioural test (self-skips without a DB) ──────────
# Same pattern as tests/test_house_dedup_merge.py::_live_session -- CI runs the
# mock-only lane (DATABASE_URL is a placeholder), so this self-skips there; it
# only executes with a real reachable Postgres (e.g. local dev DB).


def _live_session() -> Any | None:
    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else 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()
        from sqlalchemy import text as _t

        conn.execute(_t("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_erase_by_phone_finds_differently_formatted_number() -> None:
    """End-to-end on a real DB: a lead stored with phone EXACTLY as typed
    ('89991234567', no separators) must still be found and deleted when the
    erasure requester supplies the SAME digits in a DIFFERENT format
    ('8 (999) 123-45-67') -- proves the regexp_replace normalization fix
    actually matches, not just that the SQL text contains the function name."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    lead_id: Any = None
    try:
        row = db.execute(
            _t(
                "INSERT INTO trade_in_leads (phone, consent, expires_at) "
                "VALUES (:phone, TRUE, NOW() + interval '180 days') "
                "RETURNING id"
            ),
            {"phone": "89991234567"},
        ).fetchone()
        assert row is not None
        lead_id = row[0]
        db.commit()

        out = data_erasure.erase_person_data(db, phone="8 (999) 123-45-67")

        assert out["trade_in_leads_deleted"] == 1
        remaining = db.execute(
            _t("SELECT count(*) FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
            {"id": str(lead_id)},
        ).scalar()
        assert remaining == 0
    finally:
        if lead_id is not None:
            db.execute(
                _t("DELETE FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
                {"id": str(lead_id)},
            )
            db.commit()
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_erase_by_phone_finds_ru_trunk_prefix_variant() -> None:
    """End-to-end on a real DB: the coordinator's exact follow-up gap
    (2026-08-06) -- a lead stored as '89991234567' (leading '8') must be
    found and deleted when the erasure requester supplies '+7 999 123-45-67'
    (leading '+7'). Pure digit-stripping does NOT close this: stripped, the
    two are '89991234567' vs '79991234567' -- different at digit 1. Only the
    explicit 11-digit '8'->'7' collapse in _ru_phone_norm_sql makes them
    equal. Counterfactual proven manually against this same DB (raw SQL,
    see PR discussion): WITHOUT the collapse, `regexp_replace` alone finds 0
    rows for this exact pair."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    lead_id: Any = None
    try:
        row = db.execute(
            _t(
                "INSERT INTO trade_in_leads (phone, consent, expires_at) "
                "VALUES (:phone, TRUE, NOW() + interval '180 days') "
                "RETURNING id"
            ),
            {"phone": "89991234567"},
        ).fetchone()
        assert row is not None
        lead_id = row[0]
        db.commit()

        # Counterfactual: plain digit-stripping (the PRE-follow-up fix) does NOT
        # match this pair -- proves the 8-vs-7 gap was real, not a strawman.
        digits_only_match = db.execute(
            _t(
                "SELECT count(*) FROM trade_in_leads WHERE id = CAST(:id AS uuid) "
                "AND regexp_replace(phone, '\\D', '', 'g') "
                "= regexp_replace(CAST(:phone AS text), '\\D', '', 'g')"
            ),
            {"id": str(lead_id), "phone": "+7 999 123-45-67"},
        ).scalar()
        assert digits_only_match == 0, "digit-stripping alone must NOT match 8- vs 7-prefix"

        out = data_erasure.erase_person_data(db, phone="+7 999 123-45-67")

        assert out["trade_in_leads_deleted"] == 1
        remaining = db.execute(
            _t("SELECT count(*) FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
            {"id": str(lead_id)},
        ).scalar()
        assert remaining == 0
    finally:
        if lead_id is not None:
            db.execute(
                _t("DELETE FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
                {"id": str(lead_id)},
            )
            db.commit()
        db.close()


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_erase_by_phone_does_not_match_different_number() -> None:
    """Negative control: a number differing in even ONE significant digit
    must NOT be found -- proves the normalization is an exact-identity
    check, not a fuzzy/truncated match that could delete a STRANGER's data.
    Stored '89991234567' vs requested '+7 999 123-45-68' (last digit 7->8)
    -- same length, same RU-looking shape, one digit off -- zero rows."""
    from sqlalchemy import text as _t

    db = _live_session()
    assert db is not None
    lead_id: Any = None
    try:
        row = db.execute(
            _t(
                "INSERT INTO trade_in_leads (phone, consent, expires_at) "
                "VALUES (:phone, TRUE, NOW() + interval '180 days') "
                "RETURNING id"
            ),
            {"phone": "89991234567"},
        ).fetchone()
        assert row is not None
        lead_id = row[0]
        db.commit()

        out = data_erasure.erase_person_data(db, phone="+7 999 123-45-68")

        assert out["trade_in_leads_deleted"] == 0, "one differing digit must NOT match"
        remaining = db.execute(
            _t("SELECT count(*) FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
            {"id": str(lead_id)},
        ).scalar()
        assert remaining == 1, "row must survive an erasure request for a DIFFERENT number"
    finally:
        if lead_id is not None:
            db.execute(
                _t("DELETE FROM trade_in_leads WHERE id = CAST(:id AS uuid)"),
                {"id": str(lead_id)},
            )
            db.commit()
        db.close()