"""The real-Postgres lane (family rule 6).

Everything else in this suite is deliberately DB-free, which means the schema the
migrations build has never been checked against the models that read it. That gap is
what this file closes, and it is not theoretical: M13 dropped three columns and
rebuilt a generated column, and until now `alembic upgrade head` ran for the first
time when the operator's container started.

Marked `integration` and excluded from the unit lane by `-m "not integration"`, so a
workstation without Postgres runs the rest of the suite unchanged.

The schema comes from real migrations, never `metadata.create_all` (rule 82) — the
point is to test what actually ships, and `create_all` would build a schema no
deployment has ever seen.
"""
from __future__ import annotations

import uuid

import pytest
import pytest_asyncio
from sqlalchemy import select, text

from thoughtsync import ratelimit
from thoughtsync.app import create_app
from thoughtsync.db import dispose_engine, session_scope
from thoughtsync.models.note import Note
from thoughtsync.models.note_item import NoteItem
from thoughtsync.models.user import User
from thoughtsync.settings import get_setting, live, refresh_live, reset_live, set_settings
from thoughtsync.notes.helpers import derive_display_title
from thoughtsync.models.note_link_preview import NoteLinkPreview
from thoughtsync.sync import _apply_note_items
from thoughtsync.unfurl_queue import _unfurl_new_urls, detect_urls

pytestmark = pytest.mark.integration

# Every table the tests touch, child-first so FKs never block the truncate.
# RESTART IDENTITY + CASCADE keeps this honest if a table gains children later.
_TABLES = "notes, note_items, note_revisions, note_labels, note_link_previews, labels, users"


@pytest_asyncio.fixture
async def db():
    """A session against the migrated database, wiped before each test.

    Wiped BEFORE rather than after so a failed test leaves its rows behind to look at.
    """
    async with session_scope() as session:
        await session.execute(text(f"TRUNCATE {_TABLES} RESTART IDENTITY CASCADE"))
        await session.commit()
        yield session
    await dispose_engine()


@pytest_asyncio.fixture
async def app_client(db):
    """A test client against the real app, over the migrated database.

    The credential throttle is process-global and its counters outlive a single
    test, so they are cleared here — otherwise a suite that registers a few times
    starts handing out 429s for reasons that have nothing to do with the test.
    """
    ratelimit.reset_all()
    yield create_app().test_client()
    ratelimit.reset_all()


@pytest_asyncio.fixture
async def owner(db):
    """A user to hang notes off — `notes.owner_id` is a real foreign key."""
    user = User(email=f"{uuid.uuid4().hex}@example.test", display_name="Integration")
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user


async def test_the_migrated_schema_matches_the_models(db, owner):
    """The check that has never run: insert through the ORM, read it back.

    A column the models expect and the migrations never created — or the reverse —
    fails right here, instead of when a container starts.
    """
    note = Note(owner_id=owner.id, body="a thought", display_title="a thought")
    db.add(note)
    await db.commit()
    await db.refresh(note)

    found = await db.scalar(select(Note).where(Note.id == note.id))
    assert found is not None
    assert found.body == "a thought"
    assert found.display_title == "a thought"


async def test_the_dropped_columns_are_actually_gone(db):
    """M13 dropped three. If a migration silently no-opped, this is where it shows."""
    cols = set(
        (
            await db.execute(
                text("SELECT column_name FROM information_schema.columns WHERE table_name = 'notes'")
            )
        )
        .scalars()
        .all()
    )
    assert "title" not in cols, "notes.title should have gone in 0026"
    assert "kind" not in cols, "notes.kind should have gone in 0025"
    assert "display_title" in cols and "body" in cols

    rev_cols = set(
        (
            await db.execute(
                text("SELECT column_name FROM information_schema.columns WHERE table_name = 'note_revisions'")
            )
        )
        .scalars()
        .all()
    )
    assert "title" not in rev_cols, "note_revisions.title should have gone in 0026"

    tables = set(
        (await db.execute(text("SELECT table_name FROM information_schema.tables WHERE table_schema = 'public'")))
        .scalars()
        .all()
    )
    assert "note_links" not in tables, "note_links should have gone in 0024"


async def test_the_search_vector_was_rebuilt_over_the_name(db, owner):
    """0026 had to drop and recreate a STORED GENERATED column.

    Postgres refuses to drop a column another generated column depends on, so getting
    this wrong doesn't produce a subtly wrong ranking — it produces a migration that
    won't run at all. Worth proving the replacement actually indexes something.
    """
    note = Note(owner_id=owner.id, body="ferry tickets\nbook before friday", display_title="ferry tickets")
    db.add(note)
    await db.commit()

    hit = await db.scalar(
        text(
            "SELECT count(*) FROM notes "
            "WHERE search_vector @@ websearch_to_tsquery('english', :q)"
        ).bindparams(q="ferry")
    )
    assert hit == 1

    # The NAME is weight A and the body weight B, which is what makes a name match
    # rank above a body-only one. Both must be in the vector at all.
    body_only = await db.scalar(
        text(
            "SELECT count(*) FROM notes "
            "WHERE search_vector @@ websearch_to_tsquery('english', :q)"
        ).bindparams(q="friday")
    )
    assert body_only == 1


async def test_a_note_keeps_both_its_body_and_its_items(db, owner):
    """The shape M13 step 2 made normal: a note HAS a checklist, it isn't one."""
    note = Note(owner_id=owner.id, body="weekend shop", display_title="weekend shop")
    db.add(note)
    await db.flush()
    db.add_all(
        [
            NoteItem(note_id=note.id, text="milk", position=0),
            NoteItem(note_id=note.id, text="eggs", position=1),
        ]
    )
    await db.commit()

    items = (
        await db.scalars(select(NoteItem).where(NoteItem.note_id == note.id).order_by(NoteItem.position))
    ).all()
    assert [i.text for i in items] == ["milk", "eggs"]
    assert (await db.scalar(select(Note.body).where(Note.id == note.id))) == "weekend shop"


async def test_sync_no_longer_deletes_items_from_a_note_with_a_body(db, owner):
    """The data-loss path step 2 removed, pinned against a real database.

    `_apply_note_items` used to delete every item when the note wasn't `kind = "list"`.
    Nothing can produce that state any more, but this is the regression that would
    have silently eaten a checklist, and it deserves a test that would catch its
    return.
    """
    note = Note(owner_id=owner.id, body="packing", display_title="packing")
    db.add(note)
    await db.flush()
    db.add(NoteItem(note_id=note.id, text="socks", position=0))
    await db.commit()

    # A change that says nothing about items must LEAVE them alone — absent means
    # "not telling us", not "empty".
    await _apply_note_items(db, note, {"body": "packing"})
    await db.commit()
    assert (await db.scalar(select(NoteItem.text).where(NoteItem.note_id == note.id))) == "socks"

    # An explicit list replaces them.
    await _apply_note_items(db, note, {"items": [{"text": "charger", "checked": True}]})
    await db.commit()
    rows = (await db.scalars(select(NoteItem).where(NoteItem.note_id == note.id))).all()
    assert [(r.text, r.checked) for r in rows] == [("charger", True)]


async def test_a_note_with_only_items_still_has_a_name(db, owner):
    """The hole that made removing the title unsafe until step 2 closed it."""
    note = Note(owner_id=owner.id, body="", display_title="")
    db.add(note)
    await db.flush()
    db.add(NoteItem(note_id=note.id, text="milk", position=0))
    await db.commit()

    first = await db.scalar(
        select(NoteItem.text).where(NoteItem.note_id == note.id).order_by(NoteItem.position).limit(1)
    )
    note.display_title = derive_display_title(note.body, first)
    await db.commit()

    assert (await db.scalar(select(Note.display_title).where(Note.id == note.id))) == "milk"


async def test_auto_unfurl_stores_a_preview_and_skips_what_is_cached(db, owner, monkeypatch):
    """The background pass, run inline so the assertions are deterministic.

    The network is stubbed — this is about what reaches the DATABASE, not about
    parsing someone's OpenGraph tags (unfurl.py's own tests cover that). What matters
    here is the part only a real database can show: the unique constraint holding, the
    upsert going to the right row, and a second pass not re-fetching.
    """
    note = Note(
        owner_id=owner.id,
        body="read https://example.com/a and https://example.com/b",
        display_title="read https://example.com/a and https://example.com/b",
    )
    db.add(note)
    await db.commit()

    calls: list[str] = []

    async def fake_unfurl(url):
        calls.append(url)
        return {"url": url, "title": f"T {url}", "description": None, "image_url": None, "site_name": "example.com"}

    monkeypatch.setattr("thoughtsync.unfurl_queue.unfurl", fake_unfurl)

    await _unfurl_new_urls(note.id, note.body)
    assert sorted(calls) == ["https://example.com/a", "https://example.com/b"]

    rows = (await db.scalars(select(NoteLinkPreview).where(NoteLinkPreview.note_id == note.id))).all()
    assert {r.url for r in rows} == {"https://example.com/a", "https://example.com/b"}
    assert all(r.title.startswith("T ") for r in rows)

    # A second pass over an unchanged body fetches nothing — the whole reason
    # `schedule` is safe to call on every save.
    calls.clear()
    await _unfurl_new_urls(note.id, note.body)
    assert calls == []


async def test_auto_unfurl_drops_a_preview_whose_url_left_the_body(db, owner, monkeypatch):
    """A slow fetch must not resurrect a link the person deleted mid-flight."""
    note = Note(owner_id=owner.id, body="https://example.com/gone", display_title="x")
    db.add(note)
    await db.commit()

    async def fake_unfurl(url):
        # Simulate the body changing while the request was in the air.
        return {"url": url, "title": "T", "description": None, "image_url": None, "site_name": None}

    monkeypatch.setattr("thoughtsync.unfurl_queue.unfurl", fake_unfurl)
    note.body = "changed my mind"
    await db.commit()

    await _unfurl_new_urls(note.id, "https://example.com/gone")
    rows = (await db.scalars(select(NoteLinkPreview).where(NoteLinkPreview.note_id == note.id))).all()
    assert rows == [], "a preview was stored for a URL the note no longer contains"


async def test_detection_agrees_with_what_gets_stored(db, owner, monkeypatch):
    """The detector and the storage path read the same body the same way."""
    body = "one https://example.com/x. two (https://example.com/y) three"
    assert detect_urls(body) == ["https://example.com/x", "https://example.com/y"]

    note = Note(owner_id=owner.id, body=body, display_title="one")
    db.add(note)
    await db.commit()

    async def fake_unfurl(url):
        return {"url": url, "title": "T", "description": None, "image_url": None, "site_name": None}

    monkeypatch.setattr("thoughtsync.unfurl_queue.unfurl", fake_unfurl)
    await _unfurl_new_urls(note.id, body)

    stored = {
        r for r in (await db.scalars(select(NoteLinkPreview.url).where(NoteLinkPreview.note_id == note.id))).all()
    }
    assert stored == set(detect_urls(body))


async def test_registration_closes_itself_once_an_admin_exists(app_client, db):
    """The gap this removes: registration was open between "my account exists" and
    "I remembered to turn it off", and on a public host that gap starts at DNS.

    Runs against a real database because it is the interaction between two writes —
    the user row and the settings row — inside one transaction.
    """
    # The instance is empty (the fixture truncated it), so this is the first account:
    # allowed unconditionally, and it becomes the admin.
    first = await app_client.post(
        "/api/auth/register",
        json={"email": "owner@example.test", "password": "a-long-enough-password"},
    )
    assert first.status_code == 201
    assert (await first.get_json())["is_admin"] is True

    # …and the door shut behind it.
    async with session_scope() as fresh:
        assert await get_setting(fresh, "allow_registration") is False

    second = await app_client.post(
        "/api/auth/register",
        json={"email": "stranger@example.test", "password": "a-long-enough-password"},
    )
    assert second.status_code == 403

    # Re-opening it deliberately still works — that is how a second person gets in
    # until invites exist.
    async with session_scope() as fresh:
        await set_settings(fresh, {"allow_registration": True})
        await fresh.commit()

    third = await app_client.post(
        "/api/auth/register",
        json={"email": "invited@example.test", "password": "a-long-enough-password"},
    )
    assert third.status_code == 201
    assert (await third.get_json())["is_admin"] is False


async def test_security_settings_are_live_and_bounded(app_client, db):
    """The security values are settings now, not constants — so saving one has to take
    effect without a restart, and a dangerous value has to be refused.

    Real database because the whole point is the round trip: write through the admin
    API, re-read into the cache the throttle consults, observe the new number.
    """
    # An admin to authenticate as. First account, so it is allowed and becomes admin.
    reset_live()
    created = await app_client.post(
        "/api/auth/register",
        json={"email": "admin@example.test", "password": "a-long-enough-password"},
    )
    assert created.status_code == 201

    # Defaults are what the registry says.
    async with session_scope() as fresh:
        await refresh_live(fresh)
    assert live("trusted_proxy_hops") == 1
    assert live("signin_limit_per_account") == 10

    # A value that would disable the protection is REFUSED, not clamped — storing a
    # different number than the one typed is how somebody ends up believing a limit
    # is set to something it is not.
    bad = await app_client.patch("/api/settings", json={"signin_limit_per_account": 0})
    assert bad.status_code == 400
    assert "at least" in (await bad.get_json())["error"]

    # …and so is a hop count that would trust anything a caller sent.
    bad_hops = await app_client.patch("/api/settings", json={"trusted_proxy_hops": 99})
    assert bad_hops.status_code == 400

    # A legitimate change applies to the cache the throttle reads, immediately.
    ok = await app_client.patch(
        "/api/settings", json={"signin_limit_per_account": 3, "trusted_proxy_hops": 2}
    )
    assert ok.status_code == 200
    assert live("signin_limit_per_account") == 3
    assert live("trusted_proxy_hops") == 2

    # And it is persisted, not just cached.
    async with session_scope() as fresh:
        assert await get_setting(fresh, "trusted_proxy_hops") == 2

    reset_live()


async def test_the_security_group_reaches_the_admin_ui(app_client, db):
    """Every security value has to be visible and editable, which is the whole reason
    they moved out of the environment."""
    created = await app_client.post(
        "/api/auth/register",
        json={"email": "admin2@example.test", "password": "a-long-enough-password"},
    )
    assert created.status_code == 201

    resp = await app_client.get("/api/settings")
    assert resp.status_code == 200
    rows = (await resp.get_json())["settings"]
    security = {r["key"]: r for r in rows if r["group"] == "Security"}

    assert set(security) == {
        "trusted_proxy_hops",
        "signin_limit_per_account",
        "signin_limit_per_address",
        "signin_window_minutes",
        "register_limit_per_address",
        "register_window_minutes",
    }
    # The UI renders a number input from these, and it cannot offer a safe range it
    # was never told about.
    for row in security.values():
        assert row["type"] == "int"
        assert row["minimum"] is not None and row["maximum"] is not None
        assert row["description"], f"{row['key']} has no description to explain itself"