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

import pytest
import pytest_asyncio
from sqlalchemy import func, select, text, update

from inkwell import ratelimit
from inkwell.app import create_app
from inkwell.config import Config
from inkwell.db import dispose_engine, session_scope
from inkwell.models.invite import Invite
from inkwell.models.password_reset import PasswordReset
from inkwell.models.share import Share
from inkwell.models.label import NoteLabel
from inkwell.models.note import Note
from inkwell.models.note_attachment import NoteAttachment
from inkwell.models.user import User
from inkwell.notes.tags import _lift_and_reconcile_tags
from inkwell.settings import get_setting, live, refresh_live, reset_live, set_settings
from inkwell.notes.checklist import parse_items, set_item_checked
from inkwell.notes.helpers import derive_display_title
from inkwell.models.note_link_preview import NoteLinkPreview
from inkwell.models.note_revision import NoteRevision
from inkwell.revisions import REVISION_WINDOW_MINUTES, should_snapshot
from inkwell.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, shares, invites, password_resets, 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("inkwell.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("inkwell.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("inkwell.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, for an admin who would rather open the
    # door than send an invite.
    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


# --- One save path for a note's text (#5164) ---------------------------------
#
# A body edit is a sequence: keep a revision, rename the note, lift #tags, commit,
# queue link previews. It was written out once per route, and the copies drifted —
# restoring a revision skipped the previews. These pin the sequence at each door.


async def _signed_in(app_client, who: str):
    reg = await app_client.post(
        "/api/auth/register",
        json={"email": f"{who}@example.test", "password": "a-long-enough-password"},
    )
    assert reg.status_code == 201


def _record_previews(monkeypatch, module: str) -> list[tuple[str, str]]:
    """Capture what a write path queues for unfurling, instead of fetching."""
    queued: list[tuple[str, str]] = []
    monkeypatch.setattr(f"{module}.schedule_unfurls", lambda nid, body: queued.append((str(nid), body)))
    return queued


async def test_restoring_a_revision_queues_previews_for_its_links(app_client, db, monkeypatch):
    """The bug that motivated the shared path: restore was the one copy of the edit
    sequence without the unfurl step, so a link brought back by a restore stayed a
    bare URL on every surface."""
    await _signed_in(app_client, "restore")
    queued = _record_previews(monkeypatch, "inkwell.notes")

    note = await (await app_client.post("/api/notes", json={"body": "read https://example.com/a"})).get_json()
    await app_client.patch(f"/api/notes/{note['id']}", json={"body": "no link any more"})
    revisions = (await (await app_client.get(f"/api/notes/{note['id']}/revisions")).get_json())["revisions"]
    assert [r["body"] for r in revisions] == ["read https://example.com/a"]

    queued.clear()
    resp = await app_client.post(f"/api/notes/{note['id']}/revisions/{revisions[0]['id']}/restore")
    assert resp.status_code == 200
    assert (await resp.get_json())["body"] == "read https://example.com/a"
    assert queued == [(note["id"], "read https://example.com/a")], "the restored link gets its preview"


async def test_restoring_keeps_the_text_it_replaces(app_client, db):
    """Restore snapshots unconditionally — inside an editing session too — so a
    restore can itself be undone."""
    await _signed_in(app_client, "undo")
    note = await (await app_client.post("/api/notes", json={"body": "first"})).get_json()
    await app_client.patch(f"/api/notes/{note['id']}", json={"body": "second"})
    rev = (await (await app_client.get(f"/api/notes/{note['id']}/revisions")).get_json())["revisions"][0]

    await app_client.post(f"/api/notes/{note['id']}/revisions/{rev['id']}/restore")
    bodies = [r["body"] for r in (await (await app_client.get(f"/api/notes/{note['id']}/revisions")).get_json())["revisions"]]
    assert sorted(bodies) == ["first", "second"]


async def test_each_route_that_writes_text_queues_previews_only_when_it_changed(app_client, db, monkeypatch):
    await _signed_in(app_client, "routes")
    queued = _record_previews(monkeypatch, "inkwell.notes")

    note = await (await app_client.post("/api/notes", json={"body": "https://example.com/new"})).get_json()
    assert queued == [(note["id"], "https://example.com/new")], "create"

    queued.clear()
    await app_client.patch(f"/api/notes/{note['id']}", json={"pinned": True})
    assert queued == [], "a pin is not a text change"

    await app_client.patch(f"/api/notes/{note['id']}", json={"body": "https://example.com/new\n- [ ] milk"})
    assert queued == [(note["id"], "https://example.com/new\n- [ ] milk")], "PATCH"

    queued.clear()
    await app_client.patch(f"/api/notes/{note['id']}/items/0", json={"checked": True})
    assert queued == [(note["id"], "https://example.com/new\n- [x] milk")], "ticking an item"


async def test_a_pushed_note_is_named_tagged_and_queued_like_a_typed_one(app_client, db, monkeypatch):
    await _signed_in(app_client, "push")
    queued = _record_previews(monkeypatch, "inkwell.sync")
    nid = str(uuid.uuid4())

    resp = await app_client.post(
        "/api/sync/push",
        json={"changes": [{"entity": "note", "id": nid, "op": "upsert", "body": "Trip https://example.com/t\n#travel",
                           "edited_at": "2026-01-01T00:00:00Z", "created_at": "2026-01-01T00:00:00Z"}]},
    )
    assert (await resp.get_json())["results"][0]["status"] == "created"

    note = await (await app_client.get(f"/api/notes/{nid}")).get_json()
    assert note["body"] == "Trip https://example.com/t", "the standalone tag was lifted"
    assert [lb["name"] for lb in note["labels"]] == ["travel"]
    assert note["display_title"] == "Trip https://example.com/t"
    assert queued == [(nid, "Trip https://example.com/t")], "queued with the text as stored"


async def test_a_pushed_edit_keeps_the_clients_edit_time_when_a_tag_is_lifted(app_client, db):
    """Last-write-wins compares edit times, so the stored one has to be the client's.
    Lifting a tag rewrites the body in a second flush, and the column's onupdate
    used to stamp the server's clock over the time the client sent."""
    await _signed_in(app_client, "edited")
    nid = str(uuid.uuid4())
    await app_client.post(
        "/api/sync/push",
        json={"changes": [{"entity": "note", "id": nid, "op": "upsert", "body": "milk\n#groceries",
                           "edited_at": "2026-01-01T00:00:00Z", "created_at": "2026-01-01T00:00:00Z"}]},
    )
    note = await (await app_client.get(f"/api/notes/{nid}")).get_json()
    assert datetime.fromisoformat(note["updated_at"]) == datetime(2026, 1, 1, tzinfo=timezone.utc)


# --- attachments as a sync entity (#5168) ---------------------------------------


async def _note_revision(app_client, nid: str) -> int:
    feed = await (await app_client.get("/api/sync/changes?since=0")).get_json()
    return next(n["sync_revision"] for n in feed["notes"] if n["id"] == nid)


async def _pushed_note(app_client, body: str = "with a file") -> str:
    nid = str(uuid.uuid4())
    resp = await app_client.post(
        "/api/sync/push",
        json={"changes": [{"entity": "note", "id": nid, "op": "upsert", "body": body,
                           "edited_at": "2026-01-01T00:00:00Z", "created_at": "2026-01-01T00:00:00Z"}]},
    )
    assert (await resp.get_json())["results"][0]["status"] == "created"
    return nid


async def _put(app_client, aid: str, nid: str, raw: bytes, sha: str | None = None, name: str = "scan.pdf"):
    return await app_client.put(
        f"/api/sync/attachments/{aid}",
        data=raw,
        headers={"Content-Type": "application/pdf"},
        query_string={"note_id": nid, "filename": name, "sha256": sha or hashlib.sha256(raw).hexdigest()},
    )


async def test_an_offline_attachment_uploads_once_under_its_own_id(app_client, db):
    await _signed_in(app_client, "upload")
    nid = await _pushed_note(app_client)
    aid = str(uuid.uuid4())

    assert (await _put(app_client, aid, nid, b"%PDF-1", sha="0" * 64)).status_code == 400, "hash mismatch refused"
    assert (await app_client.get(f"/api/notes/{nid}")).status_code == 200
    assert (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"] == []

    before = await _note_revision(app_client, nid)
    first = await _put(app_client, aid, nid, b"%PDF-1")
    assert first.status_code == 201
    assert await _note_revision(app_client, nid) > before, "other devices hear about it"

    again = await _put(app_client, aid, nid, b"%PDF-1")
    assert again.status_code == 200 and (await again.get_json())["status"] == "exists", "a retried upload is a no-op"

    atts = (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"]
    assert [(a["id"], a["filename"], a["mime"], a["sha256"]) for a in atts] == [
        (aid, "scan.pdf", "application/pdf", hashlib.sha256(b"%PDF-1").hexdigest())
    ]
    other = await _pushed_note(app_client, "another note")
    assert (await _put(app_client, aid, other, b"%PDF-1")).status_code == 409, "one id, one note"


async def test_an_upload_to_a_note_the_caller_does_not_own_is_not_found(app_client, db):
    # Signed in first: registration is open only to the first account.
    await _signed_in(app_client, "intruder")
    stranger = User(email="stranger@example.test", display_name="Stranger")
    db.add(stranger)
    await db.flush()
    theirs = Note(owner_id=stranger.id, body="theirs", display_title="theirs")
    db.add(theirs)
    await db.commit()

    resp = await _put(app_client, str(uuid.uuid4()), str(theirs.id), b"x")
    assert resp.status_code == 404


async def test_a_pushed_attachment_delete_removes_the_row_the_file_and_bumps_the_note(app_client, db):
    await _signed_in(app_client, "remove")
    nid = await _pushed_note(app_client)
    aid = str(uuid.uuid4())
    await _put(app_client, aid, nid, b"bytes")
    stored = (await db.scalar(select(NoteAttachment).where(NoteAttachment.id == uuid.UUID(aid)))).path
    assert (Config.media_root() / stored).is_file()

    before = await _note_revision(app_client, nid)
    resp = await app_client.post(
        "/api/sync/push",
        json={"changes": [{"entity": "attachment", "id": aid, "op": "delete", "edited_at": "2000-01-01T00:00:00Z"}]},
    )
    assert (await resp.get_json())["results"] == [{"id": aid, "entity": "attachment", "status": "applied"}]
    assert (await (await app_client.get(f"/api/notes/{nid}")).get_json())["attachments"] == []
    assert not (Config.media_root() / stored).exists()
    # The edit time above is older than the note's: a removal is not a version
    # competing under last-write-wins, so it applies anyway.
    assert await _note_revision(app_client, nid) > before, "the note re-syncs without it"


async def test_a_child_delete_cannot_reach_another_owners_rows(app_client, db):
    await _signed_in(app_client, "prober")
    stranger = User(email="other@example.test", display_name="Other")
    db.add(stranger)
    await db.flush()
    theirs = Note(owner_id=stranger.id, body="theirs", display_title="theirs")
    db.add(theirs)
    await db.flush()
    att = NoteAttachment(note_id=theirs.id, path="x/y.bin", mime="application/octet-stream", size=1)
    preview = NoteLinkPreview(note_id=theirs.id, url="https://example.com/")
    db.add_all([att, preview])
    await db.commit()

    resp = await app_client.post(
        "/api/sync/push",
        json={"changes": [
            {"entity": "attachment", "id": str(att.id), "op": "delete", "edited_at": "2030-01-01T00:00:00Z"},
            {"entity": "preview", "id": str(preview.id), "op": "delete", "edited_at": "2030-01-01T00:00:00Z"},
            {"entity": "preview", "id": str(uuid.uuid4()), "op": "upsert", "edited_at": "2030-01-01T00:00:00Z"},
        ]},
    )
    statuses = [r["status"] for r in (await resp.get_json())["results"]]
    # Someone else's row answers exactly like a missing one: nothing to learn here.
    assert statuses == ["noop", "noop", "rejected"]
    assert await db.scalar(select(func.count()).select_from(NoteAttachment)) == 1
    assert await db.scalar(select(func.count()).select_from(NoteLinkPreview)) == 1


async def test_a_preview_arriving_or_leaving_moves_its_note_in_the_feed(app_client, db):
    """0031: before it, a preview fetched after the save, or dismissed on the web,
    never reached a device that had already pulled the note."""
    await _signed_in(app_client, "previews")
    nid = await _pushed_note(app_client, "read https://example.com/a")

    before = await _note_revision(app_client, nid)
    async with session_scope() as other:  # as the background unfurl does
        other.add(NoteLinkPreview(note_id=uuid.UUID(nid), url="https://example.com/a", title="A"))
        await other.commit()
    arrived = await _note_revision(app_client, nid)
    assert arrived > before

    preview_id = (await (await app_client.get(f"/api/notes/{nid}")).get_json())["previews"][0]["id"]
    await app_client.delete(f"/api/notes/{nid}/previews/{preview_id}")
    assert await _note_revision(app_client, nid) > arrived


# --- the export format, pinned against the shared fixture (#5170) ---------------


async def test_an_export_writes_the_keys_the_shared_fixture_names(app_client, db):
    """The desktop exports offline with the core's copy of this format, and both
    copies are held to core/testdata/portable.json. This is the server's side."""
    import io
    import json
    import zipfile
    from pathlib import Path

    keys = json.loads(
        (Path(__file__).resolve().parents[1] / "core" / "testdata" / "portable.json").read_text(encoding="utf-8")
    )["export"]
    await _signed_in(app_client, "exporter")
    assert (await app_client.post("/api/labels", json={"name": "travel"})).status_code == 201
    nid = await _pushed_note(app_client, "exported")
    raw = b"%PDF-1"
    assert (await _put(app_client, str(uuid.uuid4()), nid, raw)).status_code == 201

    resp = await app_client.get("/api/notes/export")
    assert resp.status_code == 200
    with zipfile.ZipFile(io.BytesIO(await resp.get_data())) as zf:
        doc = json.loads(zf.read("notes.json"))
        assert sorted(doc) == sorted(keys["document"])
        assert doc["app"] == "inkwell"
        [note] = doc["notes"]
        assert sorted(note) == sorted(keys["note"])
        assert [sorted(lb) for lb in doc["labels"]] == [sorted(keys["label"])]
        [att] = note["attachments"]
        assert sorted(att) == sorted(keys["attachment"])
        assert zf.read(att["file"]) == raw, "the listed file is in the archive"


async def test_a_keep_note_that_is_only_a_photo_imports(app_client, db):
    """It used to be skipped as empty. The core's importer keeps it as well."""
    import io
    import json
    import zipfile

    from werkzeug.datastructures import FileStorage

    await _signed_in(app_client, "keeper")
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        zf.writestr("Takeout/Keep/Photo.json", json.dumps({"textContent": "", "attachments": [{"filePath": "p.png"}]}))
        zf.writestr("Takeout/Keep/p.png", b"png bytes")
        zf.writestr("Takeout/Keep/Empty.json", json.dumps({"textContent": "  "}))
    resp = await app_client.post(
        "/api/notes/import", files={"file": FileStorage(io.BytesIO(buf.getvalue()), filename="takeout.zip")}
    )
    assert resp.status_code == 201
    assert await resp.get_json() == {"source": "keep", "imported": 1, "skipped": 1}
    [note] = (await db.scalars(select(Note))).all()
    [att] = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
    assert att.mime == "image/png"


# ---- invites (#5172) ---------------------------------------------------------------

_PASSWORD = "a-long-enough-password"


async def _admin_with_invite(app_client, **body) -> str:
    """Register the instance's first account (the admin, signed in on `app_client`)
    and have it issue an invite. Returns the token."""
    created = await app_client.post("/api/auth/register", json={"email": "owner@example.test", "password": _PASSWORD})
    assert created.status_code == 201
    resp = await app_client.post("/api/invites", json=body)
    assert resp.status_code == 201, await resp.get_data(as_text=True)
    return (await resp.get_json())["token"]


async def _register(email: str, invite: str | None = None):
    """Register from a separate client, so the admin's session stays where it is."""
    body = {"email": email, "password": _PASSWORD}
    if invite is not None:
        body["invite"] = invite
    return await create_app().test_client().post("/api/auth/register", json=body)


async def test_an_invite_lets_one_person_in_while_registration_stays_closed(app_client, db):
    token = await _admin_with_invite(app_client)

    # Registration closed itself behind the admin; a stranger with no invite is out.
    stranger = await _register("stranger@example.test")
    assert stranger.status_code == 403

    invited = await _register("guest@example.test", token)
    assert invited.status_code == 201, await invited.get_data(as_text=True)
    assert (await invited.get_json())["is_admin"] is False

    # Single use: the same link again, for anyone, is refused with the generic answer.
    again = await _register("someone-else@example.test", token)
    assert again.status_code == 403
    assert (await again.get_json())["error"] == "invalid or expired invite"

    # The admin's list says who used it.
    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
    assert [(i["status"], i["redeemed_by_email"]) for i in listed] == [("redeemed", "guest@example.test")]
    async with session_scope() as fresh:
        assert await get_setting(fresh, "allow_registration") is False


async def test_only_an_admin_handles_invites(app_client, db):
    token = await _admin_with_invite(app_client)
    guest = create_app().test_client()
    joined = await guest.post(
        "/api/auth/register", json={"email": "guest@example.test", "password": _PASSWORD, "invite": token}
    )
    assert joined.status_code == 201
    # Signed in as the guest now, who is not an admin.
    assert (await guest.post("/api/invites", json={})).status_code == 403
    assert (await guest.get("/api/invites")).status_code == 403


async def test_an_invite_pinned_to_an_email_admits_only_that_email(app_client, db):
    token = await _admin_with_invite(app_client, email="Pinned@Example.test")

    wrong = await _register("other@example.test", token)
    assert wrong.status_code == 403
    assert (await wrong.get_json())["error"] == "invalid or expired invite"

    # Any capitalisation of the pinned address, since emails compare case-insensitively.
    right = await _register("PINNED@example.test", token)
    assert right.status_code == 201


async def test_revoked_and_expired_invites_are_refused(app_client, db):
    revoked = await _admin_with_invite(app_client)
    expiring = (await (await app_client.post("/api/invites", json={"days": 1})).get_json())["token"]

    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
    by_status = {i["status"] for i in listed}
    assert by_status == {"pending"}
    # The invite made first is the last in the list (newest first).
    revoked_id = listed[-1]["id"]
    resp = await app_client.delete(f"/api/invites/{revoked_id}")
    assert resp.status_code == 200
    assert (await resp.get_json())["status"] == "revoked"

    async with session_scope() as fresh:
        await fresh.execute(
            update(Invite)
            .where(Invite.id != uuid.UUID(revoked_id))
            .values(expires_at=datetime.now(timezone.utc) - timedelta(minutes=1))
        )
        await fresh.commit()

    assert (await _register("a@example.test", revoked)).status_code == 403
    assert (await _register("b@example.test", expiring)).status_code == 403
    statuses = sorted(i["status"] for i in (await (await app_client.get("/api/invites")).get_json())["invites"])
    assert statuses == ["expired", "revoked"]


async def test_an_invite_lifetime_outside_the_bounds_is_refused(app_client, db):
    await _admin_with_invite(app_client)
    for days in (0, 31, "a week", True):
        resp = await app_client.post("/api/invites", json={"days": days})
        assert resp.status_code == 400, days
    resp = await app_client.post("/api/invites", json={"email": "not-an-address"})
    assert resp.status_code == 400


async def test_a_taken_email_leaves_the_invite_unused(app_client, db):
    """The redemption and the new account are one transaction: an account that can't
    be created must not use up the link."""
    token = await _admin_with_invite(app_client)
    taken = await _register("owner@example.test", token)
    assert taken.status_code == 409
    listed = (await (await app_client.get("/api/invites")).get_json())["invites"]
    assert listed[0]["status"] == "pending"
    assert (await _register("guest@example.test", token)).status_code == 201


async def test_two_people_racing_one_invite_cannot_both_get_in(app_client, db):
    token = await _admin_with_invite(app_client)
    results = await asyncio.gather(
        _register("first@example.test", token),
        _register("second@example.test", token),
    )
    assert sorted(r.status_code for r in results) == [201, 403]
    async with session_scope() as fresh:
        count = await fresh.scalar(select(func.count()).select_from(User))
    assert count == 2, "the admin and exactly one of the two"



# --- Admin-issued password reset links (#5173) ---------------------------------


async def _admin_and_guest(app_client):
    """The admin, signed in on `app_client`, and a second account signed in on a
    client of its own. Returns the guest's client and account id."""
    token = await _admin_with_invite(app_client)
    guest = create_app().test_client()
    joined = await guest.post(
        "/api/auth/register", json={"email": "guest@example.test", "password": _PASSWORD, "invite": token}
    )
    assert joined.status_code == 201
    return guest, (await joined.get_json())["id"]


async def _reset_link(app_client, account_id: str) -> str:
    resp = await app_client.post(f"/api/accounts/{account_id}/reset-link")
    assert resp.status_code == 201, await resp.get_data(as_text=True)
    return (await resp.get_json())["token"]


async def test_a_reset_link_sets_the_password_and_signs_the_account_out_everywhere(app_client, db):
    guest, guest_id = await _admin_and_guest(app_client)
    device = await guest.post("/api/auth/devices", json={"name": "Phone"})
    bearer = {"Authorization": f"Bearer {(await device.get_json())['token']}"}
    assert (await guest.get("/api/auth/me")).status_code == 200
    assert (await create_app().test_client().get("/api/auth/me", headers=bearer)).status_code == 200

    token = await _reset_link(app_client, guest_id)
    browser = create_app().test_client()
    reset = await browser.post("/api/auth/reset-password", json={"token": token, "password": "a-brand-new-password"})
    assert reset.status_code == 200, await reset.get_data(as_text=True)
    # The browser that used the link is signed in as the account it was made for.
    assert (await (await browser.get("/api/auth/me")).get_json())["email"] == "guest@example.test"

    # The session from before the reset is over, and so is the linked device.
    assert (await guest.get("/api/auth/me")).status_code == 401
    assert (await create_app().test_client().get("/api/auth/me", headers=bearer)).status_code == 401

    # The new password signs in; the old one doesn't.
    fresh = create_app().test_client()
    old = await fresh.post("/api/auth/login", json={"email": "guest@example.test", "password": _PASSWORD})
    assert old.status_code == 401
    new = await fresh.post("/api/auth/login", json={"email": "guest@example.test", "password": "a-brand-new-password"})
    assert new.status_code == 200

    # The admin is untouched, and the link works once.
    assert (await app_client.get("/api/auth/me")).status_code == 200
    again = await create_app().test_client().post(
        "/api/auth/reset-password", json={"token": token, "password": "yet-another-password"}
    )
    assert again.status_code == 403
    assert (await again.get_json())["error"] == "invalid or expired reset link"


async def test_only_an_admin_lists_accounts_and_makes_reset_links(app_client, db):
    guest, guest_id = await _admin_and_guest(app_client)
    listed = (await (await app_client.get("/api/accounts")).get_json())["accounts"]
    assert [(a["email"], a["is_admin"]) for a in listed] == [
        ("owner@example.test", True),
        ("guest@example.test", False),
    ]
    assert (await guest.get("/api/accounts")).status_code == 403
    assert (await guest.post(f"/api/accounts/{guest_id}/reset-link")).status_code == 403
    assert (await app_client.post(f"/api/accounts/{uuid.uuid4()}/reset-link")).status_code == 404
    assert (await app_client.post("/api/accounts/not-an-id/reset-link")).status_code == 404


async def test_an_expired_or_superseded_reset_link_is_refused(app_client, db):
    _, guest_id = await _admin_and_guest(app_client)
    first = await _reset_link(app_client, guest_id)
    second = await _reset_link(app_client, guest_id)

    async def use(token: str):
        return await create_app().test_client().post(
            "/api/auth/reset-password", json={"token": token, "password": "a-brand-new-password"}
        )

    # Making a second link closed the first.
    assert (await use(first)).status_code == 403

    async with session_scope() as fresh:
        await fresh.execute(
            update(PasswordReset).values(expires_at=datetime.now(timezone.utc) - timedelta(minutes=1))
        )
        await fresh.commit()
    assert (await use(second)).status_code == 403

    # Nothing changed: the old password still signs in.
    signed = await create_app().test_client().post(
        "/api/auth/login", json={"email": "guest@example.test", "password": _PASSWORD}
    )
    assert signed.status_code == 200


async def test_a_short_password_leaves_the_reset_link_usable(app_client, db):
    _, guest_id = await _admin_and_guest(app_client)
    token = await _reset_link(app_client, guest_id)
    client = create_app().test_client()
    short = await client.post("/api/auth/reset-password", json={"token": token, "password": "short"})
    assert short.status_code == 400
    ok = await client.post("/api/auth/reset-password", json={"token": token, "password": "a-brand-new-password"})
    assert ok.status_code == 200


async def test_revoking_this_device_from_a_web_session_is_a_bad_request(app_client, db):
    """A session-cookie caller holds no device token, so "revoke the one I'm using"
    has nothing to name. Moved here from the unit lane when the session check began
    reading the account (#5173)."""
    await _signed_in(app_client, "web")
    resp = await app_client.delete("/api/auth/devices/self")
    assert resp.status_code == 400


# --- Sharing a note (#5174) ---------------------------------------------------
#
# Owner, a recipient, and a stranger who is on the instance but not shared with.


async def _three_people(app_client):
    """The owner (admin, signed in on `app_client`), a recipient and a stranger, each
    signed in on a client of their own. Returns (recipient, stranger, ids by name)."""
    first = await _admin_with_invite(app_client)
    second = (await (await app_client.post("/api/invites", json={})).get_json())["token"]
    people = {}
    clients = []
    for name, token in (("recipient", first), ("stranger", second)):
        client = create_app().test_client()
        joined = await client.post(
            "/api/auth/register",
            json={"email": f"{name}@example.test", "password": _PASSWORD, "display_name": name.title(), "invite": token},
        )
        assert joined.status_code == 201
        people[name] = (await joined.get_json())["id"]
        clients.append(client)
    return clients[0], clients[1], people


async def _owners_note(app_client, body: str = "a shared thought #idea") -> str:
    resp = await app_client.post("/api/notes", json={"body": body})
    assert resp.status_code == 201, await resp.get_data(as_text=True)
    return (await resp.get_json())["id"]


async def _share(app_client, nid: str, user_id: str, permission: str = "view"):
    return await app_client.post(f"/api/notes/{nid}/shares", json={"user_id": user_id, "permission": permission})


async def _board_ids(client, query: str = "") -> list[str]:
    return [n["id"] for n in (await (await client.get(f"/api/notes?{query}")).get_json())["notes"]]


async def test_a_shared_note_reaches_the_recipient_and_no_one_else(app_client, db):
    recipient, stranger, people = await _three_people(app_client)
    nid = await _owners_note(app_client)

    # The directory is everyone but you.
    members = (await (await app_client.get("/api/users/directory")).get_json())["members"]
    assert sorted(m["email"] for m in members) == ["recipient@example.test", "stranger@example.test"]

    assert (await recipient.get(f"/api/notes/{nid}")).status_code == 404
    shared = await _share(app_client, nid, people["recipient"])
    assert shared.status_code == 201, await shared.get_data(as_text=True)
    assert [(s["member"]["email"], s["permission"]) for s in (await shared.get_json())["shares"]] == [
        ("recipient@example.test", "view")
    ]

    seen = await (await recipient.get(f"/api/notes/{nid}")).get_json()
    assert seen["permission"] == "view"
    assert seen["shared_by"]["display_name"] == "owner"
    # Labels are personal: the owner's #idea doesn't come with the note.
    assert seen["labels"] == []
    assert nid in await _board_ids(recipient)
    assert await _board_ids(recipient, "shared=with_me") == [nid]

    mine = await (await app_client.get(f"/api/notes/{nid}")).get_json()
    assert (mine["permission"], mine["shared"], mine["shared_by"]) == ("owner", True, None)
    assert [lb["name"] for lb in mine["labels"]] == ["idea"]
    assert await _board_ids(app_client, "shared=with_me") == []

    # The stranger, on the same instance, sees nothing of it.
    assert (await stranger.get(f"/api/notes/{nid}")).status_code == 404
    assert nid not in await _board_ids(stranger)


async def test_a_view_share_writes_nothing_and_an_edit_share_writes_only_text(app_client, db):
    recipient, _, people = await _three_people(app_client)
    nid = await _owners_note(app_client, "first line\n- [ ] milk")
    await _share(app_client, nid, people["recipient"], "view")

    # View: every write is a 404, the same answer a stranger gets (#1984).
    writes = [
        recipient.patch(f"/api/notes/{nid}", json={"body": "mine now"}),
        recipient.post(f"/api/notes/{nid}/items", json={"text": "eggs"}),
        recipient.patch(f"/api/notes/{nid}/items/0", json={"checked": True}),
        recipient.post(f"/api/notes/{nid}/trash"),
        recipient.put(f"/api/notes/{nid}/labels", json={"label_ids": []}),
        recipient.get(f"/api/notes/{nid}/shares"),
        recipient.post(f"/api/notes/{nid}/shares", json={"user_id": people["stranger"]}),
    ]
    for pending in writes:
        assert (await pending).status_code == 404

    # Sharing again changes the permission rather than adding a second grant.
    upgraded = await _share(app_client, nid, people["recipient"], "edit")
    assert [s["permission"] for s in (await upgraded.get_json())["shares"]] == ["edit"]

    # Edit: the text and the checklist.
    edited = await recipient.patch(f"/api/notes/{nid}", json={"body": "first line, edited #fromguest\n- [ ] milk"})
    assert edited.status_code == 200, await edited.get_data(as_text=True)
    assert (await edited.get_json())["labels"] == []
    ticked = await recipient.patch(f"/api/notes/{nid}/items/0", json={"checked": True})
    assert ticked.status_code == 200
    assert (await recipient.post(f"/api/notes/{nid}/items", json={"text": "eggs"})).status_code == 201

    # A #tag typed into someone else's note files it under the OWNER's tags.
    owner_tags = [lb["name"] for lb in (await (await app_client.get("/api/labels")).get_json())["labels"]]
    assert "fromguest" in owner_tags
    assert (await (await recipient.get("/api/labels")).get_json())["labels"] == []

    # Everything else stays the owner's.
    assert (await recipient.patch(f"/api/notes/{nid}", json={"pinned": True})).status_code == 403
    assert (await recipient.patch(f"/api/notes/{nid}", json={"body": "x", "archived": True})).status_code == 403
    assert (await recipient.post(f"/api/notes/{nid}/trash")).status_code == 404
    assert (await recipient.get(f"/api/notes/{nid}/revisions")).status_code == 404

    body = (await (await app_client.get(f"/api/notes/{nid}")).get_json())["body"]
    assert body.startswith("first line, edited")
    assert "- [x] milk" in body and "eggs" in body


async def test_unsharing_trashing_and_deleting_each_end_access(app_client, db):
    recipient, _, people = await _three_people(app_client)
    nid = await _owners_note(app_client)
    share_id = (await (await _share(app_client, nid, people["recipient"])).get_json())["shares"][0]["id"]

    left = await app_client.delete(f"/api/notes/{nid}/shares/{share_id}")
    assert left.status_code == 200
    assert (await left.get_json())["shares"] == []
    assert (await recipient.get(f"/api/notes/{nid}")).status_code == 404
    assert (await (await app_client.get(f"/api/notes/{nid}")).get_json())["shared"] is False

    # Trashed by its owner, it leaves the recipient's board without reaching their Trash.
    await _share(app_client, nid, people["recipient"])
    assert (await app_client.post(f"/api/notes/{nid}/trash")).status_code == 200
    assert nid not in await _board_ids(recipient)
    assert await _board_ids(recipient, "filter=trash") == []

    # Deleted for good, it is shared with nobody.
    assert (await app_client.delete(f"/api/notes/{nid}")).status_code == 200
    async with session_scope() as fresh:
        assert await fresh.scalar(select(func.count()).select_from(Share)) == 0


async def test_a_share_names_someone_else_on_this_instance(app_client, db):
    recipient, _, people = await _three_people(app_client)
    nid = await _owners_note(app_client)
    me = (await (await app_client.get("/api/auth/me")).get_json())["id"]
    assert (await _share(app_client, nid, me)).status_code == 400
    assert (await _share(app_client, nid, str(uuid.uuid4()))).status_code == 400
    assert (await _share(app_client, nid, "not-an-id")).status_code == 400
    assert (await _share(app_client, nid, people["recipient"], "admin")).status_code == 400
    # Only the owner shares: a recipient re-sharing gets the stranger's 404.
    await _share(app_client, nid, people["recipient"], "edit")
    assert (await _share(recipient, nid, people["stranger"])).status_code == 404
    # A share someone else's note doesn't hold can't be removed through this one.
    other = await _owners_note(app_client, "another")
    sid = (await (await _share(app_client, nid, people["stranger"])).get_json())["shares"][-1]["id"]
    assert (await app_client.delete(f"/api/notes/{other}/shares/{sid}")).status_code == 404