"""Online switching: a large source's table is created with its sync
triggers in the request, its records copied in batches afterwards (each its
own short transaction) while writes keep flowing in through the triggers,
then reconciled, constrained and only then read. A build that can't finish
is undone."""

import uuid

import pytest
from sqlalchemy import text

from app.core import jobs
from tests.conftest import AsyncSessionLocal
from app.crud import entity_storage, field_indexes, record_sql
from app.models.master import Master

FIELDS = [{"name": "name", "type": "text", "required": True}, {"name": "amount", "type": "numeric"}]


@pytest.fixture
def small_online(monkeypatch):
    """A handful of records counts as large; batches of two."""
    monkeypatch.setattr(entity_storage, "ONLINE_THRESHOLD", 5)
    monkeypatch.setattr(entity_storage, "BATCH_SIZE", 2)


@pytest.fixture
def manual_build(monkeypatch):
    """The test drives the build's phases itself instead of the queued job."""

    async def no_job(db, tenant_id, source_id):
        return None

    monkeypatch.setattr(entity_storage, "_schedule", no_job)


async def _datasource(client, headers, n):
    m = (await client.post("/masters", headers=headers, json={"name": "Big", "field_schema": FIELDS})).json()
    ids = []
    for i in range(n):
        r = await client.post(f"/masters/{m['id']}/records", headers=headers, json={"field_values": {"name": f"R{i}", "amount": i}})
        ids.append(r.json()["id"])
    await field_indexes.wait_for_pending()
    return m, ids


async def _typed_rows(table: str) -> dict[str, dict]:
    async with AsyncSessionLocal() as db:
        rows = await db.execute(text(f'SELECT to_jsonb(t) FROM entity_data."{table.split(".", 1)[1]}" t'))
        return {r[0]["id"]: r[0] for r in rows}


async def test_built_in_the_background(client, auth_headers, small_online):
    m, ids = await _datasource(client, auth_headers, 9)
    resp = await client.post(f"/masters/{m['id']}/storage", headers=auth_headers, json={"tier": "table"})
    assert resp.status_code == 200, resp.text
    started = resp.json()
    assert started["tier"] == "building" and started["progress"]["total"] == 9
    assert (await client.get(f"/masters/{m['id']}", headers=auth_headers)).json()["storage_tier"] == "building"

    await entity_storage.wait_for_builds()
    info = (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()
    assert info["tier"] == "table" and info["row_count"] == 9 and info["progress"] is None
    assert info["history"][0]["summary"].startswith("Table entity_data.big__") and "9 records copied" in info["history"][0]["summary"]
    # Constraints came at the end: a required value is enforced now.
    assert any(c["field"] == "name" and c["required"] for c in info["columns"])


async def test_writes_during_the_copy_are_kept_and_reconciled(client, auth_headers, small_online, manual_build):
    m, ids = await _datasource(client, auth_headers, 7)
    started = (await client.post(f"/masters/{m['id']}/storage", headers=auth_headers, json={"tier": "table"})).json()
    table = started["table"]

    tenant_id, master_id = uuid.UUID(m["tenant_id"]), uuid.UUID(m["id"])
    # Nothing reads a table that's still being built.
    async with AsyncSessionLocal() as db:
        assert (await record_sql.load_source(db, tenant_id, "master", master_id, typed=True)).table is None

    # Writes keep flowing — straight into the table through the triggers.
    new = (await client.post(f"/masters/{m['id']}/records", headers=auth_headers, json={"field_values": {"name": "New", "amount": 100}})).json()
    await client.patch(f"/masters/{m['id']}/records/{ids[0]}", headers=auth_headers, json={"field_values": {"name": "R0 edited", "amount": 50}})
    assert set(await _typed_rows(table)) == {new["id"], ids[0]}

    async with AsyncSessionLocal() as db:
        source = entity_storage.master_source(await db.get(Master, master_id))
        row = await entity_storage.entity_table(db, tenant_id, master_id)
        specs = entity_storage._specs_from_registry(row)

    # The first batch is copied; then a record is deleted — and another one
    # vanishes past its trigger (a stray row a reconciliation must remove).
    last = None
    async with AsyncSessionLocal() as db:
        async with db.begin():
            copied, last = await entity_storage._copy_batch(db, tenant_id, source, row.table_name, specs, last)
    assert copied == 2
    assert (await client.delete(f"/masters/{m['id']}/records/{ids[3]}", headers=auth_headers)).status_code == 204
    async with AsyncSessionLocal() as db:
        async with db.begin():
            stray = (await db.execute(text("SELECT gen_random_uuid()"))).scalar()
            await db.execute(text(f"INSERT INTO {entity_storage.qualified(row.table_name)} (id, tenant_id, created_at, updated_at, name) VALUES (:id, :t, now(), now(), 'ghost')"), {"id": stray, "t": tenant_id})
    while True:
        async with AsyncSessionLocal() as db:
            async with db.begin():
                copied, last = await entity_storage._copy_batch(db, tenant_id, source, row.table_name, specs, last)
        if copied < entity_storage.BATCH_SIZE:
            break
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await entity_storage._finish_build(db, tenant_id, source)

    rows = await _typed_rows(table)
    live = set(ids) - {ids[3]} | {new["id"]}
    assert set(rows) == live  # the deleted one and the stray are gone
    assert rows[ids[0]]["name"] == "R0 edited" and rows[ids[0]]["amount"] == 50  # the copy didn't overwrite a newer write
    info = (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()
    assert info["tier"] == "table"
    async with AsyncSessionLocal() as db:
        assert (await record_sql.load_source(db, tenant_id, "master", source.id, typed=True)).table is not None


async def test_a_build_that_cant_finish_is_undone(client, auth_headers, small_online, manual_build):
    m, ids = await _datasource(client, auth_headers, 6)
    started = (await client.post(f"/masters/{m['id']}/storage", headers=auth_headers, json={"tier": "table"})).json()
    # A field change waits for the build.
    resp = await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": [*FIELDS, {"name": "extra", "type": "text"}]})
    assert resp.status_code == 409 and "still being built" in resp.json()["detail"]["message"]

    # Meanwhile a record without its required name arrives past the app's own checks —
    # the table can't be made to require names, so the build is undone.
    async with AsyncSessionLocal() as db:
        async with db.begin():
            await db.execute(
                text("INSERT INTO master_records (tenant_id, master_id, field_values) SELECT tenant_id, id, '{\"amount\": 1}'::jsonb FROM masters WHERE id = :m"),
                {"m": m["id"]},
            )
            tenant_id = (await db.execute(text("SELECT tenant_id FROM masters WHERE id = :m"), {"m": m["id"]})).scalar()
    with pytest.raises(jobs.Permanent):  # a data problem: undone at once, not retried
        await entity_storage._continue_build(tenant_id, uuid.UUID(m["id"]))

    info = (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()
    assert info["tier"] == "flexible"
    assert info["history"][0]["action"] == "demote" and "failed and was undone" in info["history"][0]["summary"]
    async with AsyncSessionLocal() as db:
        assert (await db.execute(text("SELECT to_regclass(:t)"), {"t": started["table"]})).scalar() is None
        assert (await db.execute(text("SELECT count(*) FROM pg_trigger WHERE tgname LIKE :p"), {"p": f"ff_ent_{m['id'].replace('-', '')}%"})).scalar() == 0