"""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
from datetime import datetime, timedelta, timezone

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.label import NoteLabel
from thoughtsync.models.note import Note
from thoughtsync.models.user import User
from thoughtsync.notes.tags import _lift_and_reconcile_tags
from thoughtsync.settings import get_setting, live, refresh_live, reset_live, set_settings
from thoughtsync.notes.checklist import parse_items, set_item_checked
from thoughtsync.notes.helpers import derive_display_title
from thoughtsync.models.note_link_preview import NoteLinkPreview
from thoughtsync.models.note_revision import NoteRevision
from thoughtsync.revisions import REVISION_WINDOW_MINUTES, should_snapshot
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_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_its_prose_on_both_sides_of_its_list(db, owner):
    """The shape M304 made expressible at all.

    The old model could not hold this: a row had a position in a table and none in the
    text, so a checklist could only ever render AFTER the body. Prose, list, prose is
    the case that proves the storage changed, not just the styling.
    """
    body = "weekend shop\n\n- [ ] milk\n- [x] eggs\n\nback before six"
    note = Note(owner_id=owner.id, body=body, display_title=derive_display_title(body))
    db.add(note)
    await db.commit()

    stored = await db.scalar(select(Note.body).where(Note.id == note.id))
    assert [(i.text, i.checked) for i in parse_items(stored)] == [("milk", False), ("eggs", True)]
    assert stored.splitlines()[0] == "weekend shop"
    assert stored.splitlines()[-1] == "back before six"


async def test_ticking_an_item_is_a_body_edit(db, owner):
    """What replaced `_apply_note_items`: there is no separate thing left to apply.

    The regression that function guarded against — a sync silently eating a checklist
    off a note that also had a body — cannot recur, because there is nothing to delete.
    A pushed body either has the lines or it does not.
    """
    body = "packing\n\n- [ ] socks"
    note = Note(owner_id=owner.id, body=body, display_title="packing")
    db.add(note)
    await db.commit()

    note.body = set_item_checked(note.body, 0, True)
    await db.commit()

    stored = await db.scalar(select(Note.body).where(Note.id == note.id))
    assert stored == "packing\n\n- [x] socks"
    assert parse_items(stored)[0].checked
    # The prose is untouched — a tick rewrites one line, not the note.
    assert stored.splitlines()[0] == "packing"


async def test_a_standalone_tag_leaves_the_body_and_becomes_an_ordinary_label(db, owner):
    """M311. The tag was being shown twice — as text and as a chip — so the text goes.

    `via_tag=False` is the load-bearing half. It is what makes the chip's × appear in
    both editors (they gate it on exactly this), which matters because deleting the
    text is no longer a way to remove the tag: there is no text.
    """
    note = Note(owner_id=owner.id, body="#todo\nreorganize the homepage", display_title="#todo")
    db.add(note)
    await db.flush()
    await _lift_and_reconcile_tags(db, note)
    await db.commit()

    assert note.body == "reorganize the homepage"
    # Re-derived by the lift itself. Every caller sets display_title BEFORE calling,
    # so if the function did not do this the note would be named after a line it had
    # just deleted.
    assert note.display_title == "reorganize the homepage"

    rows = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
    assert len(rows) == 1
    assert rows[0].via_tag is False


async def test_a_tag_moved_onto_its_own_line_graduates_instead_of_vanishing(db, owner):
    """The bug a naive lift has, pinned.

    A tag that is still in prose stays derived. Move it to its own line and it must
    become an ordinary label — NOT be detached for no longer appearing in the body,
    which is what happens if the row is dropped before it is graduated.
    """
    note = Note(owner_id=owner.id, body="call #mom tomorrow", display_title="call #mom tomorrow")
    db.add(note)
    await db.flush()
    await _lift_and_reconcile_tags(db, note)
    await db.commit()

    rows = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
    assert len(rows) == 1
    assert rows[0].via_tag is True
    assert note.body == "call #mom tomorrow", "a tag inside a sentence is left alone"

    note.body = "#mom\ncall tomorrow"
    await _lift_and_reconcile_tags(db, note)
    await db.commit()

    rows = (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()
    assert len(rows) == 1, "the label survived the move"
    assert rows[0].via_tag is False
    assert note.body == "call tomorrow"


async def test_deleting_an_inline_tag_still_detaches_it(db, owner):
    """The old behaviour, unchanged where the text is unchanged. A tag still living in
    prose is still owned by that prose."""
    note = Note(owner_id=owner.id, body="call #mom tomorrow", display_title="call #mom tomorrow")
    db.add(note)
    await db.flush()
    await _lift_and_reconcile_tags(db, note)
    await db.commit()
    assert len((await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all()) == 1

    note.body = "call tomorrow"
    await _lift_and_reconcile_tags(db, note)
    await db.commit()
    assert (await db.scalars(select(NoteLabel).where(NoteLabel.note_id == note.id))).all() == []


async def test_a_note_with_only_a_list_still_has_a_name(db, owner):
    """The hole that made removing the title unsafe, still closed — by a different
    mechanism. There is no item table to fall back to any more; the name comes from
    the first line with its marker stripped, because calling the note "- [ ] milk"
    would show someone the storage instead of the note.
    """
    body = "- [ ] milk\n- [ ] eggs"
    note = Note(owner_id=owner.id, body=body, display_title=derive_display_title(body))
    db.add(note)
    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"


async def _revision_count(db, note_id) -> int:
    rows = (await db.scalars(select(NoteRevision.id).where(NoteRevision.note_id == note_id))).all()
    return len(rows)


async def test_a_session_of_edits_costs_one_revision(db, owner):
    """The change that makes autosave affordable.

    Version history used to snapshot on EVERY body write, so the clients saved as
    rarely as they could — only when an editor closed — and a crash mid-session lost
    everything typed. Durability was paying for history. Now a sitting earns one
    revision no matter how many times it is written, so a client can write whenever
    it likes.
    """
    note = Note(owner_id=owner.id, body="one", display_title="one")
    db.add(note)
    await db.commit()

    # A session's worth of autosaves.
    for text_ in ("one two", "one two three", "one two three four"):
        if await should_snapshot(db, note.id, note.body, text_):
            db.add(NoteRevision(note_id=note.id, body=note.body))
        note.body = text_
        await db.commit()

    assert await _revision_count(db, note.id) == 1

    # And it is the body as it was BEFORE the sitting, not some midpoint — which is
    # what makes one-per-session the useful granularity rather than an arbitrary one.
    kept = (await db.scalars(select(NoteRevision.body).where(NoteRevision.note_id == note.id))).all()
    assert kept == ["one"]


async def test_rewriting_the_same_text_is_not_a_version(db, owner):
    note = Note(owner_id=owner.id, body="unchanged", display_title="unchanged")
    db.add(note)
    await db.commit()

    assert await should_snapshot(db, note.id, note.body, "unchanged") is False
    assert await _revision_count(db, note.id) == 0


async def test_a_later_sitting_earns_its_own_revision(db, owner):
    """The window has to REOPEN, or a note edited daily would keep only its first
    version forever — which would be a worse history than the one we replaced."""
    note = Note(owner_id=owner.id, body="today", display_title="today")
    db.add(note)
    await db.flush()
    # A revision from longer ago than one session: the clock is not mocked, the row
    # is simply written with an older timestamp, which is what the query reads.
    stale = datetime.now(timezone.utc) - timedelta(minutes=REVISION_WINDOW_MINUTES + 1)
    db.add(NoteRevision(note_id=note.id, body="yesterday", created_at=stale))
    await db.commit()

    assert await should_snapshot(db, note.id, note.body, "tomorrow") is True


async def test_a_revision_inside_the_window_blocks_another(db, owner):
    note = Note(owner_id=owner.id, body="draft", display_title="draft")
    db.add(note)
    await db.flush()
    db.add(NoteRevision(note_id=note.id, body="earlier", created_at=datetime.now(timezone.utc)))
    await db.commit()

    assert await should_snapshot(db, note.id, note.body, "draft revised") is False


async def test_renaming_a_tag_onto_an_existing_one_merges_into_the_older(app_client, db):
    """Renaming a tag onto a name another tag holds merges them, and the OLDER row
    is the survivor — whichever side the caller happened to be renaming.

    Runs against a real database because the whole question is about `created_at`
    ordering and the note_labels rows moving, neither of which a unit test sees.

    Age decides, rather than "the one that already held the name", so that renaming
    A→B and renaming B→A land on the same row. If the incumbent won, the survivor
    would depend on which way round someone typed it, and two clients racing the
    same tidy-up would disagree about which id still exists.
    """
    reg = await app_client.post(
        "/api/auth/register",
        json={"email": "tags@example.test", "password": "a-long-enough-password"},
    )
    assert reg.status_code == 201

    # Two separate requests, so two transactions and two distinct `func.now()`s.
    older = await (await app_client.post("/api/labels", json={"name": "grocery"})).get_json()
    newer = await (await app_client.post("/api/labels", json={"name": "errands"})).get_json()

    one = await (await app_client.post("/api/notes", json={"body": "milk"})).get_json()
    two = await (await app_client.post("/api/notes", json={"body": "stamps"})).get_json()
    await app_client.put(f"/api/notes/{one['id']}/labels", json={"label_ids": [older["id"]]})
    await app_client.put(f"/api/notes/{two['id']}/labels", json={"label_ids": [newer["id"]]})

    # Rename the YOUNGER onto the older's name, with different casing — matching is
    # case-insensitive, and the survivor must end up spelled the way we asked.
    resp = await app_client.patch(f"/api/labels/{newer['id']}", json={"name": "Grocery"})
    assert resp.status_code == 200
    survivor = await resp.get_json()

    assert survivor["id"] == older["id"], "the older row is the one that keeps existing"
    assert survivor["name"] == "Grocery", "the survivor takes the spelling that was asked for"

    listing = (await (await app_client.get("/api/labels")).get_json())["labels"]
    assert len(listing) == 1, "the two became one"
    assert listing[0]["id"] == older["id"]
    assert listing[0]["count"] == 2, "it carries every note from both sides"


async def test_the_rename_merge_survivor_does_not_depend_on_the_direction(app_client, db):
    """The mirror of the test above: rename the OLDER onto the younger's name. The
    older still survives — it just changes its name — so the two directions agree."""
    reg = await app_client.post(
        "/api/auth/register",
        json={"email": "tags2@example.test", "password": "a-long-enough-password"},
    )
    assert reg.status_code == 201

    older = await (await app_client.post("/api/labels", json={"name": "grocery"})).get_json()
    newer = await (await app_client.post("/api/labels", json={"name": "errands"})).get_json()

    resp = await app_client.patch(f"/api/labels/{older['id']}", json={"name": "errands"})
    assert resp.status_code == 200
    survivor = await resp.get_json()

    assert survivor["id"] == older["id"], "age wins in this direction too"
    assert survivor["name"] == "errands"
    assert newer["id"] != older["id"]

    listing = (await (await app_client.get("/api/labels")).get_json())["labels"]
    assert [lb["id"] for lb in listing] == [older["id"]]


async def test_creating_a_tag_that_differs_only_in_case_returns_the_existing_one(app_client, db):
    """A case-sensitive match here used to mint "Groceries" beside "groceries". No
    synced client can hold both — their `labels` index is unique on `lower(name)` —
    so the pair was a pull that would fail later, on a phone, with no UI in the path.
    """
    reg = await app_client.post(
        "/api/auth/register",
        json={"email": "tags3@example.test", "password": "a-long-enough-password"},
    )
    assert reg.status_code == 201

    first = await app_client.post("/api/labels", json={"name": "groceries"})
    assert first.status_code == 201
    second = await app_client.post("/api/labels", json={"name": "Groceries"})
    assert second.status_code == 200, "an existing tag is returned, not a second one made"

    assert (await second.get_json())["id"] == (await first.get_json())["id"]
    listing = (await (await app_client.get("/api/labels")).get_json())["labels"]
    assert len(listing) == 1