"""Datasource storage tiers — a datasource stored as a real table: typed
columns kept in step with every write, constraints the database enforces,
migrations when its fields change, and a refusal (listing the records)
whenever existing data wouldn't fit."""

import uuid

import pytest
from sqlalchemy import text

from tests.conftest import AsyncSessionLocal
from app.crud import field_indexes

CUSTOMER_FIELDS = [
    {"name": "name", "type": "text", "required": True},
    {"name": "credit", "type": "numeric"},
    {"name": "tier", "type": "single_select", "options": ["Gold", "Silver"]},
    {"name": "since", "type": "date"},
    {"name": "active", "type": "boolean"},
]


async def _master(client, headers, name, fields):
    resp = await client.post("/masters", headers=headers, json={"name": name, "field_schema": fields})
    assert resp.status_code == 201, resp.text
    return resp.json()


async def _record(client, headers, master_id, values):
    resp = await client.post(f"/masters/{master_id}/records", headers=headers, json={"field_values": values})
    assert resp.status_code == 201, resp.text
    return resp.json()


async def _rows(table: str, order: str = "created_at") -> list[dict]:
    async with AsyncSessionLocal() as db:
        result = await db.execute(text(f'SELECT to_jsonb(t) FROM entity_data."{table.split(".", 1)[1]}" t ORDER BY {order}'))
        return [r[0] for r in result]


async def _promote(client, headers, master_id):
    # An "Indexed" field's index builds in the background; let it finish
    # first, as a person clicking through would.
    await field_indexes.wait_for_pending()
    return await client.post(f"/masters/{master_id}/storage", headers=headers, json={"tier": "table"})


async def test_promote_types_the_data_and_keeps_it_in_step(client, auth_headers):
    m = await _master(client, auth_headers, "Customers", CUSTOMER_FIELDS)
    a = await _record(client, auth_headers, m["id"], {"name": "Asha", "credit": 1500.5, "tier": "Gold", "since": "2024-02-01", "active": True})

    preview = (await client.get(f"/masters/{m['id']}/storage/preview", headers=auth_headers)).json()
    assert preview["issues"] == []
    assert {c["field"]: c["sql_type"] for c in preview["columns"]} == {"name": "text", "credit": "numeric", "tier": "text", "since": "date", "active": "boolean"}

    resp = await _promote(client, auth_headers, m["id"])
    assert resp.status_code == 200, resp.text
    info = resp.json()
    assert info["tier"] == "table" and info["row_count"] == 1
    assert info["history"][0]["action"] == "promote"
    assert (await client.get(f"/masters/{m['id']}", headers=auth_headers)).json()["storage_tier"] == "table"

    [row] = await _rows(info["table"])
    assert row["id"] == a["id"]
    assert row["credit"] == 1500.5 and row["since"] == "2024-02-01" and row["active"] is True

    # Every later write lands in the table too.
    b = await _record(client, auth_headers, m["id"], {"name": "Ravi", "credit": 20})
    await client.patch(f"/masters/{m['id']}/records/{a['id']}", headers=auth_headers, json={"field_values": {"name": "Asha K", "credit": 99, "tier": "Silver"}})
    assert (await client.delete(f"/masters/{m['id']}/records/{b['id']}", headers=auth_headers)).status_code == 204
    [row] = await _rows(info["table"])
    assert row["name"] == "Asha K" and row["credit"] == 99 and row["tier"] == "Silver" and row["since"] is None


async def test_the_table_refuses_what_it_cant_hold(client, auth_headers):
    m = await _master(client, auth_headers, "Typed", CUSTOMER_FIELDS)
    await _promote(client, auth_headers, m["id"])
    # Straight into master_records, past the API's own validation: the
    # database still refuses it, naming the field.
    async with AsyncSessionLocal() as db:
        with pytest.raises(Exception, match="FF:type:credit:number"):
            async with db.begin():
                await db.execute(
                    text("INSERT INTO master_records (tenant_id, master_id, field_values) SELECT tenant_id, id, :v FROM masters WHERE id = :m"),
                    {"v": '{"name": "X", "credit": "lots"}', "m": m["id"]},
                )
        with pytest.raises(Exception, match="tier__options"):
            async with db.begin():
                await db.execute(
                    text("INSERT INTO master_records (tenant_id, master_id, field_values) SELECT tenant_id, id, :v FROM masters WHERE id = :m"),
                    {"v": '{"name": "X", "tier": "Platinum"}', "m": m["id"]},
                )


async def test_records_that_dont_fit_block_the_change_and_are_listed(client, auth_headers):
    m = await _master(client, auth_headers, "Contacts", [{"name": "name", "type": "text"}, {"name": "email", "type": "text"}])
    await _record(client, auth_headers, m["id"], {"name": "Has email", "email": "a@b.c"})
    missing = await _record(client, auth_headers, m["id"], {"name": "No email"})
    # Flexible storage lets a field become required over existing records...
    fields = [{"name": "name", "type": "text"}, {"name": "email", "type": "text", "required": True}]
    assert (await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": fields})).status_code == 200

    # ...a table doesn't: the record in the way is named.
    issues = (await client.get(f"/masters/{m['id']}/storage/preview", headers=auth_headers)).json()["issues"]
    assert [(i["field"], i["problem"], i["count"], i["record_ids"]) for i in issues] == [("email", "required", 1, [missing["id"]])]
    resp = await _promote(client, auth_headers, m["id"])
    assert resp.status_code == 422
    assert resp.json()["detail"]["issues"][0]["problem"] == "required"
    assert (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()["tier"] == "flexible"


async def test_field_changes_migrate_the_table_or_are_refused(client, auth_headers):
    m = await _master(client, auth_headers, "Products", [{"name": "sku", "type": "text", "required": True}, {"name": "price", "type": "numeric"}])
    await _record(client, auth_headers, m["id"], {"sku": "A1", "price": 12.5})
    info = (await _promote(client, auth_headers, m["id"])).json()

    # Added and removed fields become added and dropped columns.
    fields = [{"name": "sku", "type": "text", "required": True}, {"name": "price", "type": "numeric"}, {"name": "stock", "type": "integer", "indexed": True}]
    assert (await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": fields})).status_code == 200
    [row] = await _rows(info["table"])
    assert "stock" in row and row["stock"] is None

    # A type the records don't fit (12.5 isn't a whole number) is refused,
    # and the whole edit with it.
    await field_indexes.wait_for_pending()
    bad = [{"name": "sku", "type": "text", "required": True}, {"name": "price", "type": "integer"}, {"name": "stock", "type": "integer"}]
    resp = await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": bad})
    assert resp.status_code == 422
    assert resp.json()["detail"]["issues"][0]["samples"] == ["12.5"]
    master = (await client.get(f"/masters/{m['id']}", headers=auth_headers)).json()
    assert [f["type"] for f in master["field_schema"]] == ["text", "numeric", "integer"]

    # A change the records do fit goes through: whole-number stock becomes a decimal.
    fine = [{"name": "sku", "type": "text", "required": True}, {"name": "price", "type": "numeric"}, {"name": "stock", "type": "numeric", "indexed": True}]
    assert (await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": fine})).status_code == 200
    await _record(client, auth_headers, m["id"], {"sku": "B2", "price": 3, "stock": 2.5})
    stock_column = [c for c in (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()["columns"] if c["field"] == "stock"][0]
    assert stock_column["sql_type"] == "numeric" and stock_column["column"] != "stock"  # a type change comes back under a new name
    assert [r[stock_column["column"]] for r in await _rows(info["table"])] == [None, 2.5]

    storage = (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()
    assert [h["action"] for h in storage["history"]] == ["migrate", "migrate", "promote"]
    assert any(c["field"] == "stock" and c["indexed"] for c in storage["columns"])


async def test_links_between_tables_are_foreign_keys(client, auth_headers):
    customers = await _master(client, auth_headers, "Clients", [{"name": "name", "type": "text"}])
    orders = await _master(
        client, auth_headers, "Orders",
        [{"name": "ref", "type": "text"}, {"name": "client", "type": "reference", "reference_master_id": customers["id"], "reference_display_field": "name"}],
    )
    asha = await _record(client, auth_headers, customers["id"], {"name": "Asha"})
    await _record(client, auth_headers, orders["id"], {"ref": "O-1", "client": asha["id"]})

    # Orders first: its link isn't enforced until Clients is a table too...
    info = (await _promote(client, auth_headers, orders["id"])).json()
    assert [c["link_enforced"] for c in info["columns"] if c["field"] == "client"] == [False]
    await _promote(client, auth_headers, customers["id"])
    # ...then it becomes a real foreign key.
    info = (await client.get(f"/masters/{orders['id']}/storage", headers=auth_headers)).json()
    assert [c["link_enforced"] for c in info["columns"] if c["field"] == "client"] == [True]

    # A linked-to record can't be deleted out from under its links.
    resp = await client.delete(f"/masters/{customers['id']}/records/{asha['id']}", headers=auth_headers)
    assert resp.status_code == 409, resp.text
    assert "still linked from Orders" in resp.json()["detail"]["message"]

    # Nor can its table go while the link depends on it.
    resp = await client.post(f"/masters/{customers['id']}/storage", headers=auth_headers, json={"tier": "flexible"})
    assert resp.status_code == 409 and "Orders" in resp.json()["detail"]["message"]
    assert (await client.post(f"/masters/{orders['id']}/storage", headers=auth_headers, json={"tier": "flexible"})).status_code == 200
    assert (await client.post(f"/masters/{customers['id']}/storage", headers=auth_headers, json={"tier": "flexible"})).status_code == 200
    async with AsyncSessionLocal() as db:
        left = (await db.execute(text("SELECT count(*) FROM pg_tables WHERE schemaname = 'entity_data' AND (tablename LIKE 'clients__%' OR tablename LIKE 'orders__%')"))).scalar()
        assert left == 0
    # Back to flexible, records are untouched.
    assert (await client.delete(f"/masters/{customers['id']}/records/{asha['id']}", headers=auth_headers)).status_code == 204


async def test_deleting_a_table_datasource_drops_its_table(client, auth_headers):
    m = await _master(client, auth_headers, "Scratch", [{"name": "x", "type": "text"}])
    table = (await _promote(client, auth_headers, m["id"])).json()["table"]
    assert (await client.delete(f"/masters/{m['id']}", headers=auth_headers)).status_code == 204
    async with AsyncSessionLocal() as db:
        exists = (await db.execute(text("SELECT to_regclass(:t)"), {"t": table})).scalar()
        triggers = (await db.execute(text("SELECT count(*) FROM pg_trigger WHERE tgname LIKE :p"), {"p": f"ff_ent_{uuid.UUID(m['id']).hex}%"})).scalar()
    assert exists is None and triggers == 0


async def test_a_busy_records_table_is_a_clear_try_again(client, auth_headers):
    m = await _master(client, auth_headers, "Busy", [{"name": "x", "type": "text"}])
    await field_indexes.wait_for_pending()
    # Another change to master_records under way (an index build takes this lock).
    async with AsyncSessionLocal() as other:
        async with other.begin():
            await other.execute(text("LOCK TABLE master_records IN SHARE UPDATE EXCLUSIVE MODE"))
            resp = await client.post(f"/masters/{m['id']}/storage", headers=auth_headers, json={"tier": "table"})
    assert resp.status_code == 409
    assert "try again" in resp.json()["detail"]["message"]
    assert (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()["tier"] == "flexible"
    # Once it's done, the same request goes through.
    assert (await _promote(client, auth_headers, m["id"])).status_code == 200


async def test_a_datasource_can_be_created_as_a_table(client, auth_headers):
    resp = await client.post("/masters", headers=auth_headers, json={"name": "Born Typed", "field_schema": CUSTOMER_FIELDS, "storage_tier": "table"})
    assert resp.status_code == 201, resp.text
    m = resp.json()
    assert m["storage_tier"] == "table"
    info = (await client.get(f"/masters/{m['id']}/storage", headers=auth_headers)).json()
    assert info["tier"] == "table" and info["row_count"] == 0
    # The table is there straight away, with its typed columns…
    async with AsyncSessionLocal() as db:
        columns = dict((await db.execute(text(
            "SELECT column_name, data_type FROM information_schema.columns WHERE table_schema = 'entity_data' AND table_name = :t"
        ), {"t": info["table"].split(".", 1)[1]})).all())
    assert columns["credit"] == "numeric" and columns["since"] == "date" and columns["active"] == "boolean"
    # …takes every record, and follows later field changes.
    await _record(client, auth_headers, m["id"], {"name": "Asha", "credit": 10})
    resp = await client.patch(f"/masters/{m['id']}", headers=auth_headers, json={"field_schema": [*CUSTOMER_FIELDS, {"name": "city", "type": "text"}]})
    assert resp.status_code == 200, resp.text
    [row] = await _rows(info["table"])
    assert row["name"] == "Asha" and row["credit"] == 10 and "city" in row

    # Flexible stays the default.
    plain = await _master(client, auth_headers, "Born Flexible", CUSTOMER_FIELDS)
    assert plain["storage_tier"] == "flexible"
    assert (await client.post("/masters", headers=auth_headers, json={"name": "Bad", "storage_tier": "sheet"})).status_code == 422