"""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 NOTE_COLORS, 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_item import NoteItem
from ..models.note_link_preview import NoteLinkPreview
from ..models.note_revision import NoteRevision
from ..responses import json_error, not_found, parse_uuid
from ..retention import purge_note
from ..settings import get_setting
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 (
    _reconcile_tags,
    parse_tags,
)
from .recurrence import REMINDER_RECURRENCES, next_occurrence, normalize_recurrence
from .serialize import _items_for_notes, _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",
    "_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")
    color = request.args.get("color")
    kind = request.args.get("kind")
    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 color is not None:
            if color not in NOTE_COLORS:
                return json_error("invalid color", 400)
            stmt = stmt.where(Note.color == color)
        if kind is not None:
            if kind not in ("text", "list"):
                return json_error("invalid kind", 400)
            stmt = stmt.where(Note.kind == kind)
        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 title+body (generated tsvector, migration 0005),
            # ranked — so the facet bar's text box searches, not just filters.
            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("/search")
@login_required
async def search_notes():
    q = (request.args.get("q") or "").strip()
    if not q:
        return jsonify({"notes": []})
    async with session_scope() as db:
        tsquery = func.websearch_to_tsquery("english", q)
        # search_vector is a generated column (migration 0005), not mapped on the ORM.
        search_col = literal_column("notes.search_vector")
        stmt = (
            select(Note)
            .where(
                visible_to_user("note", Note.owner_id, Note.id, g.user_id),
                Note.deleted_at.is_(None),
                search_col.op("@@")(tsquery),
            )
            .order_by(func.ts_rank(search_col, tsquery).desc(), Note.updated_at.desc())
            .limit(100)
        )
        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)
        items_map = await _items_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, [])
                items = items_map.get(n.id, [])
                atts = att_by_note.get(n.id, [])
                short = str(n.id)[:8]
                payload["notes"].append(
                    {
                        "id": str(n.id),
                        "title": n.title,
                        "display_title": n.display_title,
                        "body": n.body,
                        "color": n.color,
                        "kind": n.kind,
                        "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],
                        "items": [{"text": it["text"], "checked": it["checked"]} for it in items],
                        "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, items))
                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 {}
    title = data.get("title") if isinstance(data.get("title"), str) else ""
    body = data.get("body") if isinstance(data.get("body"), str) else ""
    kind = data.get("kind") if data.get("kind") in ("text", "list") else "text"
    # A checklist note's "content" is its items, not the body — so it's non-empty
    # when it has a title or at least one item (quick-add can create one in one shot).
    item_texts = parse_list_items(data.get("items")) if kind == "list" else []
    if kind == "list":
        if not (title.strip() or item_texts):
            return json_error("note is empty", 400)
    elif is_empty_note(title, body):
        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)
            )
        )
        clean_title = title.strip() or None
        note = Note(
            owner_id=g.user_id,
            title=clean_title,
            display_title=derive_display_title(clean_title, body),
            body=body,
            kind=kind,
            color=normalize_color(data.get("color")),
            position=int(max_pos) + 1,
        )
        db.add(note)
        await db.flush()  # assign note.id before writing items/links
        for pos, text in enumerate(item_texts):
            db.add(NoteItem(note_id=note.id, text=text, position=pos))
        await _reconcile_tags(db, note)
        await db.commit()
        await db.refresh(note)
        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_title = note.title
        old_body = note.body
        if "title" in data:
            title = data["title"] if isinstance(data["title"], str) else ""
            note.title = title.strip() or None
        if "body" in data and isinstance(data["body"], str):
            note.body = data["body"]
        if "color" in data:
            note.color = normalize_color(data["color"])
        if "kind" in data and data["kind"] in ("text", "list"):
            note.kind = data["kind"]
        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"])
        # Recompute the display name (explicit title, else first body line) whenever
        # the title or body may have changed.
        if "title" in data or "body" in data:
            note.display_title = derive_display_title(note.title, note.body)
        if "body" in data:
            await _reconcile_tags(db, note)
        # Version history: snapshot the PRE-edit title+body whenever either changed.
        if note.title != old_title or note.body != old_body:
            db.add(NoteRevision(note_id=note.id, title=old_title, body=old_body))
        await db.commit()
        await db.refresh(note)
        return jsonify(await _serialize_note(db, note))


def _serialize_revision(rev: NoteRevision) -> dict:
    return {
        "id": str(rev.id),
        "title": rev.title,
        "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.title == rev.title and 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 title/body ripple as a normal edit.
        db.add(NoteRevision(note_id=note.id, title=note.title, body=note.body))
        note.title = rev.title
        note.body = rev.body
        note.display_title = derive_display_title(note.title, note.body)
        await _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))


async def _get_item(db, note: Note, item_id: str) -> NoteItem | None:
    iid = parse_uuid(item_id)
    if iid is None:
        return None
    return await db.scalar(select(NoteItem).where(NoteItem.id == iid, NoteItem.note_id == note.id))


@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()
        max_pos = await db.scalar(
            select(func.coalesce(func.max(NoteItem.position), -1)).where(NoteItem.note_id == note.id)
        )
        db.add(NoteItem(note_id=note.id, text=text, position=int(max_pos) + 1))
        await db.commit()
        return jsonify(await _serialize_note(db, note)), 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 {}
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        item = await _get_item(db, note, item_id)
        if item is None:
            return not_found()
        if "text" in data and isinstance(data["text"], str):
            item.text = data["text"]
        if "checked" in data:
            item.checked = bool(data["checked"])
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@bp.delete("/<note_id>/items/<item_id>")
@login_required
async def delete_item(note_id: str, item_id: str):
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        item = await _get_item(db, note, item_id)
        if item is None:
            return not_found()
        await db.delete(item)
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@bp.post("/<note_id>/items/reorder")
@login_required
async def reorder_items(note_id: str):
    data = await request.get_json(silent=True) or {}
    order = data.get("item_ids")
    if not isinstance(order, list):
        return json_error("item_ids must be a list", 400)
    async with session_scope() as db:
        note = await _get_owned(db, note_id)
        if note is None:
            return not_found()
        existing = {
            str(i.id): i for i in (await db.scalars(select(NoteItem).where(NoteItem.note_id == note.id))).all()
        }
        pos = 0
        for iid in order:
            item = existing.get(str(iid))
            if item is not None:
                item.position = pos
                pos += 1
        await db.commit()
        return jsonify(await _serialize_note(db, note))


@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})