inkwell/src/thoughtsync/notes/__init__.py

"""Notes API (the `/api/notes` blueprint).

The bulk of the shared logic lives in cohesive sibling modules — serialization
(`serialize`), #tags (`tags`), recurring reminders (`recurrence`),
small text/query helpers (`helpers`), and export/import (`import_export`). The
route handlers themselves stay here so blueprint registration is in one place, and
`bp` is defined in `_bp` so every module can import it without a cycle.

External callers import from `thoughtsync.notes` (see `__all__`); those names are
re-exported here so the package is a drop-in replacement for the old module."""
from __future__ import annotations

import hashlib
import io
import json
import os
import uuid
import zipfile
from datetime import datetime, timedelta, timezone

from quart import Response, g, jsonify, request, send_file
from sqlalchemy import func, literal_column, select

from ..acl import visible_to_user
from ..auth import login_required
from ..colors import normalize_color
from ..common import coerce_bool, iso, parse_dt
from ..config import Config
from ..db import session_scope
from ..labeling import reconcile_manual_labels, resolve_owned_label_ids
from ..models.label import Label, NoteLabel
from ..models.note import Note
from ..models.note_attachment import NoteAttachment
from ..models.note_link_preview import NoteLinkPreview
from ..models.note_revision import NoteRevision
from ..revisions import should_snapshot
from .checklist import (
    append_item,
    parse_items,
    remove_item,
    set_item_checked,
    set_item_text,
)
from ..responses import json_error, not_found, parse_uuid
from ..retention import purge_note
from ..settings import get_setting
from ..unfurl_queue import schedule as schedule_unfurls
from ..unfurl import UnfurlError, unfurl
from ._bp import bp
from .helpers import (
    ALLOWED_IMAGE_MIMES,
    VALID_FILTERS,
    _attachment_ext,
    _get_owned,
    _header_filename,
    _safe_filename,
    _slugify,
    apply_filter,
    derive_display_title,
    is_empty_note,
    parse_list_items,
)
from .import_export import (
    IMPORT_MAX_ENTRIES,
    _ImportBudget,
    _ImportTooLarge,
    _create_imported_note,
    _keep_spec,
    _native_spec,
    _note_markdown,
    _read_import_specs,
    _usec_to_dt,
)
from .tags import (
    _lift_and_reconcile_tags,
    parse_tags,
)
from .recurrence import REMINDER_RECURRENCES, next_occurrence, normalize_recurrence
from .serialize import _labels_for_notes, _serialize_note, _serialize_notes

__all__ = [
    "bp",
    "derive_display_title",
    "is_empty_note",
    "parse_list_items",
    "parse_tags",
    "normalize_color",
    "normalize_recurrence",
    "next_occurrence",
    "_lift_and_reconcile_tags",
    "_serialize_notes",
    "_safe_filename",
    "_attachment_ext",
    "_header_filename",
    "_slugify",
    "_keep_spec",
    "_native_spec",
    "_usec_to_dt",
]


@bp.get("")
@login_required
async def list_notes():
    filter_name = request.args.get("filter", "active")
    if filter_name not in VALID_FILTERS:
        return json_error("invalid filter", 400)
    # Combinable facet filters (all optional, AND-ed together) — the rich-search /
    # saved-filter lens. Multiple ?label= narrow to notes carrying ALL of them.
    label_params = request.args.getlist("label")
    has_reminder = coerce_bool(request.args.get("has_reminder"))
    has_attachment = coerce_bool(request.args.get("has_attachment"))
    query_text = (request.args.get("q") or "").strip()
    # Optional creation-date range — the "browse by when" / Timeline lens. Both bounds
    # are ISO-8601 instants forming a HALF-OPEN interval [created_after, created_before),
    # so a client can pass local day-boundaries (start-of-day .. start-of-next-day)
    # without off-by-one. `sort=created` orders newest-captured first for a chronological
    # timeline; the default keeps the board's pinned/position/updated order.
    after_param = request.args.get("created_after")
    before_param = request.args.get("created_before")
    sort = request.args.get("sort")
    async with session_scope() as db:
        stmt = select(Note).where(visible_to_user("note", Note.owner_id, Note.id, g.user_id))
        stmt = apply_filter(stmt, filter_name)
        for raw_label in label_params:
            lid = parse_uuid(raw_label)
            if lid is None:
                return json_error("invalid label", 400)
            stmt = stmt.where(Note.id.in_(select(NoteLabel.note_id).where(NoteLabel.label_id == lid)))
        if has_reminder:
            stmt = stmt.where(Note.remind_at.is_not(None))
        if has_attachment:
            stmt = stmt.where(Note.id.in_(select(NoteAttachment.note_id)))
        if after_param:
            after_dt = parse_dt(after_param)
            if after_dt is None:
                return json_error("invalid created_after", 400)
            stmt = stmt.where(Note.created_at >= after_dt)
        if before_param:
            before_dt = parse_dt(before_param)
            if before_dt is None:
                return json_error("invalid created_before", 400)
            stmt = stmt.where(Note.created_at < before_dt)
        if query_text:
            # Full-text match over the note's name + body (generated tsvector,
            # migrations 0005/0026), ranked. This is the ONLY text search now: the
            # separate facet-less `/search` route was removed because landing on it
            # was the one place you could not also narrow by tag (note 2930).
            tsquery = func.websearch_to_tsquery("english", query_text)
            search_col = literal_column("notes.search_vector")
            stmt = stmt.where(search_col.op("@@")(tsquery)).order_by(
                func.ts_rank(search_col, tsquery).desc(), Note.updated_at.desc()
            )
        elif sort == "created":
            stmt = stmt.order_by(Note.created_at.desc())
        else:
            stmt = stmt.order_by(Note.pinned.desc(), Note.position.desc(), Note.updated_at.desc())
        notes = (await db.scalars(stmt)).all()
        return jsonify({"notes": await _serialize_notes(db, notes)})


@bp.get("/reminders")
@login_required
async def list_reminders():
    async with session_scope() as db:
        stmt = (
            select(Note)
            .where(
                visible_to_user("note", Note.owner_id, Note.id, g.user_id),
                Note.deleted_at.is_(None),
                Note.remind_at.is_not(None),
            )
            .order_by(Note.remind_at.asc())
        )
        notes = (await db.scalars(stmt)).all()
        return jsonify({"notes": await _serialize_notes(db, notes)})


@bp.post("/<note_id>/reminder/complete")
@login_required
async def complete_reminder(note_id: str):
    """Mark a reminder handled: a recurring reminder advances to its next occurrence;
    a one-off clears its reminder."""
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        if note.remind_at is not None and note.recurrence in REMINDER_RECURRENCES:
            note.remind_at = next_occurrence(note.remind_at, note.recurrence, datetime.now(timezone.utc))
        else:
            note.remind_at = None
            note.recurrence = None
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@bp.post("/<note_id>/reminder/snooze")
@login_required
async def snooze_reminder(note_id: str):
    """Re-fire a reminder a little later — remind_at moves to now + `minutes`."""
    data = await request.get_json(silent=True) or {}
    try:
        minutes = int(data.get("minutes", 10))
    except (ValueError, TypeError):
        minutes = 10
    minutes = max(1, min(minutes, 60 * 24 * 30))  # 1 minute .. 30 days
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        note.remind_at = datetime.now(timezone.utc) + timedelta(minutes=minutes)
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@bp.get("/export")
@login_required
async def export_notes():
    """Download all of the caller's notes as a zip: a machine-readable notes.json,
    a Markdown file per note, and the attachment media. 'Your data is yours.'"""
    async with session_scope() as db:
        notes_list = (
            await db.scalars(
                select(Note).where(Note.owner_id == g.user_id, Note.deleted_at.is_(None)).order_by(Note.created_at)
            )
        ).all()
        ids = [n.id for n in notes_list]
        labels_map = await _labels_for_notes(db, ids)
        att_rows = (
            (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id.in_(ids)))).all() if ids else []
        )
        att_by_note: dict = {}
        for a in att_rows:
            att_by_note.setdefault(a.note_id, []).append(a)
        all_labels = (
            await db.scalars(select(Label).where(Label.owner_id == g.user_id).order_by(Label.name))
        ).all()

        payload: dict = {
            "app": "thoughtsync",
            "version": 1,
            "exported_at": datetime.now(timezone.utc).isoformat(),
            "labels": [{"name": lb.name, "color": lb.color} for lb in all_labels],
            "notes": [],
        }
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
            for n in notes_list:
                labels = labels_map.get(n.id, [])
                atts = att_by_note.get(n.id, [])
                short = str(n.id)[:8]
                payload["notes"].append(
                    {
                        "id": str(n.id),
                        "display_title": n.display_title,
                        "body": n.body,
                        "pinned": n.pinned,
                        "archived": n.archived,
                        "remind_at": n.remind_at.isoformat() if n.remind_at else None,
                        "recurrence": n.recurrence,
                        "created_at": n.created_at.isoformat() if n.created_at else None,
                        "updated_at": n.updated_at.isoformat() if n.updated_at else None,
                        "labels": [lb["name"] for lb in labels],
                        "attachments": [
                            {"file": f"attachments/{short}/{os.path.basename(a.path)}", "mime": a.mime} for a in atts
                        ],
                    }
                )
                zf.writestr(f"notes/{_slugify(n.display_title)}-{short}.md", _note_markdown(n, labels))
                for a in atts:
                    src = Config.media_root() / a.path
                    if src.is_file():
                        zf.write(src, f"attachments/{short}/{os.path.basename(a.path)}")
            zf.writestr("notes.json", json.dumps(payload, indent=2, ensure_ascii=False))
        data = buf.getvalue()
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d")
    return Response(
        data,
        headers={
            "Content-Type": "application/zip",
            "Content-Disposition": f'attachment; filename="thoughtsync-export-{stamp}.zip"',
        },
    )


@bp.post("/import")
@login_required
async def import_notes():
    """Import notes from an uploaded zip — either a ThoughtSync export (round-trip)
    or a Google Keep Takeout archive. Additive: imported notes are appended, never
    overwriting existing ones. Returns a per-run summary."""
    files = await request.files
    upload = files.get("file")
    if upload is None:
        return json_error("no file provided", 400)
    raw = upload.stream.read()
    if not raw:
        return json_error("empty upload", 400)
    try:
        zf = zipfile.ZipFile(io.BytesIO(raw))
    except zipfile.BadZipFile:
        return json_error("that file isn't a valid .zip archive", 400)

    if len(zf.infolist()) > IMPORT_MAX_ENTRIES:
        return json_error("that archive has too many files to import", 413)

    # Bound total decompressed bytes so a zip bomb can't exhaust memory/disk. Any entry
    # (or the cumulative total) blowing the budget aborts the whole import — nothing is
    # committed, since the session rolls back when the exception exits its block.
    budget = _ImportBudget()
    try:
        specs, source = _read_import_specs(zf, budget)
        if not specs:
            return json_error(
                "no importable notes found — expected a ThoughtSync export or a Google Keep Takeout zip", 400
            )

        imported = 0
        skipped = 0
        async with session_scope() as db:
            max_pos = await db.scalar(
                select(func.coalesce(func.max(Note.position), 0)).where(
                    Note.owner_id == g.user_id, Note.deleted_at.is_(None)
                )
            )
            pos = int(max_pos)
            for spec in specs:
                if await _create_imported_note(db, g.user_id, spec, zf, pos + 1, budget):
                    pos += 1
                    imported += 1
                else:
                    skipped += 1
            await db.commit()
    except _ImportTooLarge:
        return json_error("that archive is too large to import", 413)
    return jsonify({"source": source, "imported": imported, "skipped": skipped}), 201


@bp.get("/titles")
@login_required
async def list_titles():
    """Owner's non-trashed notes, keyed by their display NAME.

    Survived the removal of [[wiki-links]] (note 2897) because it was serving two
    different things, and only one of them was linking. This is what the command
    palette lists so someone can jump to a note by name — which is recall, the thing
    this app is actually for. The `[[` autocomplete that also read it is gone.
    """
    async with session_scope() as db:
        rows = (
            await db.scalars(select(Note).where(Note.owner_id == g.user_id, Note.deleted_at.is_(None)))
        ).all()
        return jsonify(
            {"titles": [{"id": str(n.id), "title": n.display_title} for n in rows if n.display_title]}
        )


@bp.post("/reorder")
@login_required
async def reorder_notes():
    data = await request.get_json(silent=True) or {}
    ids = data.get("ids")
    if not isinstance(ids, list):
        return json_error("ids must be a list", 400)
    parsed: list = []
    for rid in ids:
        parsed_id = parse_uuid(rid)
        if parsed_id is None:
            return json_error("invalid id", 400)
        parsed.append(parsed_id)
    async with session_scope() as db:
        owned = {
            n.id: n
            for n in (await db.scalars(select(Note).where(Note.owner_id == g.user_id, Note.id.in_(parsed)))).all()
        }
        total = len(parsed)
        for index, nid in enumerate(parsed):
            note = owned.get(nid)
            if note is not None:
                note.position = total - index
        await db.commit()
        return jsonify({"ok": True})


@bp.post("")
@login_required
async def create_note():
    data = await request.get_json(silent=True) or {}
    body = data.get("body") if isinstance(data.get("body"), str) else ""
    # Items are accepted on ANY note — a checklist is something a note HAS.
    item_texts = parse_list_items(data.get("items"))
    if is_empty_note(body, item_texts):
        return json_error("note is empty", 400)
    async with session_scope() as db:
        # New notes go to the top of the manual order.
        max_pos = await db.scalar(
            select(func.coalesce(func.max(Note.position), 0)).where(
                Note.owner_id == g.user_id, Note.deleted_at.is_(None)
            )
        )
        # Items still arrive separately — a client holds a list, not a blob — but they
        # are folded into the body, which is where a checklist lives now (M304).
        for text in item_texts:
            body = append_item(body, text)
        note = Note(
            owner_id=g.user_id,
            display_title=derive_display_title(body),
            body=body,
            position=int(max_pos) + 1,
        )
        db.add(note)
        await db.flush()  # assign note.id before writing links
        # The FOLDED body: an item can carry a #tag too.
        await _lift_and_reconcile_tags(db, note)
        await db.commit()
        await db.refresh(note)
        # After the commit, never before it: the note is saved and the response is
        # about to go out. Any link previews arrive on a later read.
        schedule_unfurls(note.id, note.body)
        return jsonify(await _serialize_note(db, note)), 201


@bp.get("/<note_id>")
@login_required
async def get_note(note_id: str):
    nid = parse_uuid(note_id)
    if nid is None:
        return not_found()
    async with session_scope() as db:
        note = await db.scalar(
            select(Note).where(Note.id == nid, visible_to_user("note", Note.owner_id, Note.id, g.user_id))
        )
        if note is None:
            return not_found()
        return jsonify(await _serialize_note(db, note))


@bp.patch("/<note_id>")
@login_required
async def update_note(note_id: str):
    data = await request.get_json(silent=True) or {}
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        old_body = note.body
        if "body" in data and isinstance(data["body"], str):
            note.body = data["body"]
        if "pinned" in data:
            note.pinned = bool(data["pinned"])
        if "archived" in data:
            note.archived = bool(data["archived"])
        if "remind_at" in data:
            raw = data["remind_at"]
            if raw in (None, ""):
                note.remind_at = None
                note.recurrence = None  # no reminder → recurrence is moot
            else:
                remind_dt = parse_dt(raw)
                if remind_dt is None:
                    return json_error("invalid remind_at", 400)
                note.remind_at = remind_dt
        if "recurrence" in data:
            note.recurrence = normalize_recurrence(data["recurrence"])
        if "body" in data:
            note.display_title = derive_display_title(note.body)
            await _lift_and_reconcile_tags(db, note)
        # Version history: snapshot the PRE-edit body, once per editing session
        # rather than once per write — see revisions.should_snapshot. Writing often
        # is what lets a client autosave instead of hoarding text until it closes.
        if await should_snapshot(db, note.id, old_body, note.body):
            db.add(NoteRevision(note_id=note.id, body=old_body))
        await db.commit()
        await db.refresh(note)
        if note.body != old_body:
            schedule_unfurls(note.id, note.body)
        return jsonify(await _serialize_note(db, note))


def _serialize_revision(rev: NoteRevision) -> dict:
    return {
        "id": str(rev.id),
        "body": rev.body,
        "created_at": iso(rev.created_at),
    }


@bp.get("/<note_id>/revisions")
@login_required
async def list_revisions(note_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        rows = (
            await db.scalars(
                select(NoteRevision)
                .where(NoteRevision.note_id == note.id)
                .order_by(NoteRevision.created_at.desc())
                .limit(50)
            )
        ).all()
        return jsonify({"revisions": [_serialize_revision(r) for r in rows]})


@bp.post("/<note_id>/revisions/<rev_id>/restore")
@login_required
async def restore_revision(note_id: str, rev_id: str):
    rid = parse_uuid(rev_id)
    if rid is None:
        return not_found()
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        rev = await db.scalar(select(NoteRevision).where(NoteRevision.id == rid, NoteRevision.note_id == note.id))
        if rev is None:
            return not_found()
        if note.body == rev.body:
            return jsonify(await _serialize_note(db, note))  # already at this version — no-op
        # Snapshot the CURRENT state first, so restoring is itself undoable, then apply
        # the revision — with the same body ripple as a normal edit.
        db.add(NoteRevision(note_id=note.id, body=note.body))
        note.body = rev.body
        note.display_title = derive_display_title(note.body)
        await _lift_and_reconcile_tags(db, note)
        await db.commit()
        await db.refresh(note)
        return jsonify(await _serialize_note(db, note))


@bp.put("/<note_id>/labels")
@login_required
async def set_note_labels(note_id: str):
    data = await request.get_json(silent=True) or {}
    raw_ids = data.get("label_ids")
    if not isinstance(raw_ids, list):
        return json_error("label_ids must be a list", 400)
    label_ids: list = []
    for rid in raw_ids:
        parsed_label = parse_uuid(rid)
        if parsed_label is None:
            return json_error("invalid label id", 400)
        label_ids.append(parsed_label)
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        # The picker manages MANUAL labels only; tag-sourced (via_tag=True) rows are
        # governed by the body #tags and survive a picker save (see labeling module).
        owned = await resolve_owned_label_ids(db, label_ids, g.user_id)
        await reconcile_manual_labels(db, note, owned)
        await db.commit()
        return jsonify(await _serialize_note(db, note))


def _item_index(item_id: str) -> int | None:
    """An item's id is its ordinal (see serialize.items_of). Anything else is a stale
    id from a client that has not reloaded, and the answer to those is 404."""
    try:
        index = int(item_id)
    except (TypeError, ValueError):
        return None
    return index if index >= 0 else None


async def _rewrite_body(db, note: Note, body: str):
    """Every item mutation is a body edit, so all of them land here.

    One place means one place that snapshots a revision, re-derives `#tags`, recomputes
    the name and queues link unfurls — rather than three routes each remembering to.
    Deliberately the same sequence the PATCH route runs for a body change, because it
    IS a body change.
    """
    old_body = note.body
    if await should_snapshot(db, note.id, old_body, body):
        db.add(NoteRevision(note_id=note.id, body=old_body))
    note.body = body
    note.display_title = derive_display_title(body)
    await _lift_and_reconcile_tags(db, note)
    await db.commit()
    await db.refresh(note)
    if note.body != old_body:
        schedule_unfurls(note.id, note.body)
    return jsonify(await _serialize_note(db, note))


@bp.post("/<note_id>/items")
@login_required
async def add_item(note_id: str):
    data = await request.get_json(silent=True) or {}
    text = data["text"].strip() if isinstance(data.get("text"), str) else ""
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        response = await _rewrite_body(db, note, append_item(note.body, text))
        return response, 201


@bp.patch("/<note_id>/items/<item_id>")
@login_required
async def update_item(note_id: str, item_id: str):
    data = await request.get_json(silent=True) or {}
    index = _item_index(item_id)
    if index is None:
        return not_found()
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        if index >= len(parse_items(note.body)):
            return not_found()
        body = note.body
        if "text" in data and isinstance(data["text"], str):
            body = set_item_text(body, index, data["text"])
        if "checked" in data:
            body = set_item_checked(body, index, bool(data["checked"]))
        return await _rewrite_body(db, note, body)


@bp.delete("/<note_id>/items/<item_id>")
@login_required
async def delete_item(note_id: str, item_id: str):
    index = _item_index(item_id)
    if index is None:
        return not_found()
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        if index >= len(parse_items(note.body)):
            return not_found()
        return await _rewrite_body(db, note, remove_item(note.body, index))


# The reorder route is gone with M304. Reordering a checklist is moving a line, which
# is something a text editor already does and no client ever called this for — the
# only reference to it in the tree was a test asserting the route existed.


@bp.post("/<note_id>/attachments")
@login_required
async def upload_attachment(note_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        files = await request.files
        upload = files.get("file")
        if upload is None:
            return json_error("no file provided", 400)
        # Any file type is allowed — images render inline, everything else downloads.
        mime = (upload.content_type or "application/octet-stream").split(";")[0].strip().lower()
        filename = _safe_filename(upload.filename)
        ext = _attachment_ext(filename, mime)
        # A native client may supply the attachment's id so an offline-attached file
        # keeps its identity across sync. Re-uploading an id it already has is a no-op.
        form = await request.form
        raw_id = (form.get("id") or "").strip()
        att_id = uuid.uuid4()
        if raw_id:
            parsed_att = parse_uuid(raw_id)
            if parsed_att is None:
                return json_error("invalid attachment id", 400)
            att_id = parsed_att
            existing = await db.scalar(
                select(NoteAttachment).where(NoteAttachment.id == att_id, NoteAttachment.note_id == note.id)
            )
            if existing is not None:
                return jsonify(await _serialize_note(db, note))  # already have this blob
        raw = upload.stream.read()
        max_mb = int(await get_setting(db, "max_attachment_mb"))
        if len(raw) > max_mb * 1024 * 1024:
            return json_error(f"file is too large (max {max_mb} MB)", 413)
        rel = os.path.join(str(note.id), f"{att_id}{ext}")
        dest = Config.media_root() / rel
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(raw)
        db.add(
            NoteAttachment(
                id=att_id,
                note_id=note.id,
                path=rel,
                filename=filename,
                mime=mime,
                size=len(raw),
                sha256=hashlib.sha256(raw).hexdigest(),
            )
        )
        await db.commit()
        return jsonify(await _serialize_note(db, note)), 201


@bp.get("/<note_id>/attachments/<att_id>")
@login_required
async def get_attachment(note_id: str, att_id: str):
    nid = parse_uuid(note_id)
    aid = parse_uuid(att_id)
    if nid is None or aid is None:
        return not_found()
    async with session_scope() as db:
        # Owner OR shared may view (rule 47).
        note = await db.scalar(
            select(Note).where(Note.id == nid, visible_to_user("note", Note.owner_id, Note.id, g.user_id))
        )
        if note is None:
            return not_found()
        att = await db.scalar(select(NoteAttachment).where(NoteAttachment.id == aid, NoteAttachment.note_id == nid))
        if att is None:
            return not_found()
        mime = att.mime
        filename = att.filename or os.path.basename(att.path)
        file_path = Config.media_root() / att.path
    if not file_path.is_file():
        return not_found()
    response = await send_file(str(file_path), mimetype=mime)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["Cache-Control"] = "private, max-age=86400"
    # Only trusted raster image types render inline; everything else — notably
    # image/svg+xml, which can carry script — downloads, so a shared note's attachment
    # can't execute script in a viewer's session (rule 47 = shares).
    disposition = "inline" if mime in ALLOWED_IMAGE_MIMES else "attachment"
    response.headers["Content-Disposition"] = f'{disposition}; filename="{_header_filename(filename)}"'
    return response


@bp.delete("/<note_id>/attachments/<att_id>")
@login_required
async def delete_attachment(note_id: str, att_id: str):
    aid = parse_uuid(att_id)
    if aid is None:
        return not_found()
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        att = await db.scalar(select(NoteAttachment).where(NoteAttachment.id == aid, NoteAttachment.note_id == note.id))
        if att is None:
            return not_found()
        file_path = Config.media_root() / att.path
        await db.delete(att)
        await db.commit()
        result = await _serialize_note(db, note)
    try:
        file_path.unlink(missing_ok=True)
    except OSError:
        pass
    return jsonify(result)


@bp.post("/<note_id>/unfurl")
@login_required
async def unfurl_link(note_id: str):
    """Fetch a link preview for a URL in this note and store it. Opt-in via the
    enable_url_unfurl setting; SSRF-guarded server-side fetch (see unfurl.py)."""
    data = await request.get_json(silent=True) or {}
    url = (data.get("url") or "").strip()
    if not url:
        return json_error("url is required", 400)
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        if not await get_setting(db, "enable_url_unfurl"):
            return json_error("link previews are disabled", 403)
    # Fetch OUTSIDE the DB session — network IO shouldn't hold a connection.
    try:
        preview = await unfurl(url)
    except UnfurlError as e:
        return json_error(str(e), 502)
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        # Keyed by the ORIGINAL pasted url (what the note body contains, so the client
        # matches it) — re-unfurling the same link updates the cached preview in place.
        row = await db.scalar(
            select(NoteLinkPreview).where(NoteLinkPreview.note_id == note.id, NoteLinkPreview.url == url)
        )
        if row is None:
            row = NoteLinkPreview(note_id=note.id, url=url)
            db.add(row)
        row.title = preview["title"]
        row.description = preview["description"]
        row.image_url = preview["image_url"]
        row.site_name = preview["site_name"]
        await db.commit()
        return jsonify(await _serialize_note(db, note)), 201


@bp.delete("/<note_id>/previews/<preview_id>")
@login_required
async def delete_preview(note_id: str, preview_id: str):
    pid = parse_uuid(preview_id)
    if pid is None:
        return not_found()
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        row = await db.scalar(
            select(NoteLinkPreview).where(NoteLinkPreview.id == pid, NoteLinkPreview.note_id == note.id)
        )
        if row is None:
            return not_found()
        await db.delete(row)
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@bp.post("/<note_id>/trash")
@login_required
async def trash_note(note_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        note.deleted_at = datetime.now(timezone.utc)
        await db.commit()
        await db.refresh(note)
        return jsonify(await _serialize_note(db, note))


@bp.post("/<note_id>/restore")
@login_required
async def restore_note(note_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        note.deleted_at = None
        await db.commit()
        await db.refresh(note)
        return jsonify(await _serialize_note(db, note))


@bp.delete("/<note_id>")
@login_required
async def delete_note(note_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        if note.deleted_at is None:
            return json_error("note must be trashed before permanent delete", 409)
        # A tombstone, not a dropped row. Deleting the row outright would leave the
        # server with no record the note ever existed, so a linked device that was
        # offline at the time would keep its copy forever — and push it back the
        # next time it was edited. The delete has to be something clients can LEARN.
        await purge_note(db, note)
        await db.commit()
        return jsonify({"ok": True})