"""FeedEcho — self-hosted RSS feed cross-poster.

Routes feed items to Mastodon accounts or email addresses. Web UI for managing
feeds, accounts, echoes, settings, and viewing post history.

Route convention: a handler that performs blocking I/O (outbound HTTP via the
sync httpx client, SMTP, DNS) is declared ``def``, not ``async def``, so
FastAPI runs it in the threadpool. An ``async def`` handler runs on the single
event loop, where one 30-second feed fetch stalls every other request in the
process. ``tests/test_review_fixes.py`` enforces this.
"""

import csv
import io
import json
import os
import re
import logging
import threading
import time
import uuid
import secrets
import secrets as _secrets
from pathlib import Path
from datetime import datetime, timedelta, timezone
from urllib.parse import urlsplit, urlparse, urlencode

import httpx

from fastapi import FastAPI, Request, Form, HTTPException, Query, UploadFile, File, BackgroundTasks
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
import xml.etree.ElementTree as ElementTree
from xml.sax.saxutils import quoteattr
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.routing import Match
from fastapi.staticfiles import StaticFiles
from jinja2 import Environment, FileSystemLoader, select_autoescape

from database import as_utc_naive, dialect, get_db, init_db, timestamp_str, _column_names
from _version import __version__ as APP_VERSION
import auth
from auth import current_user_id
import security
from feed_parser import discover_feed_url, fetch_feed, SSRFError, validate_outbound_url
import plans
import invites
import import_export
from plans import PlanError
from mastodon import test_connection, post_status, verify_credentials
from bluesky import (
    BlueskyAuthError,
    BlueskyError,
    create_session as bluesky_create_session,
    normalize_handle as bluesky_normalize_handle,
    resolve_pds as bluesky_resolve_pds,
    session_expiry as bluesky_session_expiry,
    test_connection as test_bluesky_connection,
)
from template_engine import render_template, available_variables, validate_template
from microblog import (
    MicroblogAuthError,
    MicroblogError,
    list_destinations as microblog_list_destinations,
    test_connection as test_microblog_connection,
)
from matrix import (
    MatrixAuthError,
    MatrixError,
    MatrixPermissionError,
    connect as matrix_connect,
    normalize_homeserver as matrix_normalize_homeserver,
    normalize_room as matrix_normalize_room,
    test_connection as test_matrix_connection,
)
from discord import (
    DiscordAuthError,
    DiscordError,
    DiscordNotFoundError,
    connect as discord_connect,
    normalize_discord_webhook_url as discord_normalize_webhook_url,
    test_connection as test_discord_connection,
)
import telegram
from telegram import (
    TelegramAuthError,
    TelegramBadRequestError,
    TelegramError,
    TelegramNotFoundError,
    connect as telegram_connect,
    test_connection as test_telegram_connection,
)
from webhook import (
    CLEAR_BODY_SENTINEL as webhook_clear_body_sentinel,
    MAX_HEADERS_TEXT,
    WebhookError,
    dump_headers,
    is_clear_body as webhook_is_clear_body,
    load_headers,
    normalize_body_template as webhook_normalize_body_template,
    normalize_webhook_url as webhook_normalize_url,
    parse_headers as webhook_parse_headers,
    test_connection as test_webhook_connection,
)
import github
import scheduler  # noqa: E402 — module-qualified access (tests monkeypatch through it)
from scheduler import (
    start_scheduler,
    stop_scheduler,
    check_feed,
    process_echo,
    next_free_slot,
)
from oauth import get_authorize_url, exchange_code, verify_state
from email_sender import get_smtp_settings, test_smtp_connection
from utils import is_valid_email
from utils import categories_from_value

import logging_setup

logging_setup.setup_logging()
logger = logging.getLogger("feedecho")
access_logger = logging.getLogger("feedecho.access")

app = FastAPI(title="FeedEcho", version=APP_VERSION)

BASE_DIR = Path(__file__).resolve().parent
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")

jinja = Environment(
    loader=FileSystemLoader(BASE_DIR / "templates"),
    autoescape=select_autoescape(["html"]),
)


def _as_utc_naive(value) -> datetime | None:
    """Normalize a stored timestamp (sqlite TEXT or psycopg datetime) to
    a naive UTC datetime, or None if unparseable/empty.

    Thin alias for ``database.as_utc_naive`` so the dialect rule lives in
    one place; see that docstring for the UTC-writer invariant.
    """
    return as_utc_naive(value)


def _iso_utc(value) -> str:
    """ISO-8601 UTC ('2026-08-24T06:46:00Z') for the browser-side
    local-time conversion; '' when the value is missing/unparseable."""
    dt = _as_utc_naive(value)
    return dt.strftime("%Y-%m-%dT%H:%M:%SZ") if dt else ""


def _utc_text(value) -> str:
    """Plain 'YYYY-MM-DD HH:MM:SS' UTC fallback text (shown when JS is off)."""
    dt = _as_utc_naive(value)
    return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else ""


def _safe_url(value) -> str:
    """A URL safe to place in href, or "" for anything not http(s).

    Item and post URLs come from feed publishers and remote APIs. Autoescaping
    stops HTML injection but does nothing about the scheme, so a feed serving
    `javascript:...` as an item link would run script in the authenticated
    origin on click. Rendering an empty href instead drops the link.
    """
    url = str(value or "").strip()
    if url.lower().startswith(("http://", "https://")):
        return url
    if url:
        # debug, not warning: a feed full of odd schemes would otherwise log a
        # line per URL on every page render. The dropped link is visible in the
        # page itself.
        logger.debug("Refusing to render non-http(s) URL in a link: %.80r", url)
    return ""


jinja.filters["iso_utc"] = _iso_utc
jinja.filters["utc_text"] = _utc_text
jinja.filters["safe_url"] = _safe_url
# Admin template uses `value is trial_pending` to render the card-gated trial
# sentinel as "pending card" rather than a nonsense 1999 date.
jinja.tests["trial_pending"] = lambda value: plans.trial_pending(value)

# Every table that holds one connected destination (one stored credential and
# one outbound posting path). Plan caps and usage counts sum across ALL of
# them, so a new destination type must be added here or it silently escapes
# the quota. Keyed by the echoes.destination_type value it belongs to.
DESTINATION_TABLES_BY_TYPE = {
    "mastodon": "accounts",
    "email": "email_accounts",
    "bluesky": "bluesky_accounts",
    "microblog": "microblog_accounts",
    "matrix": "matrix_accounts",
    "discord": "discord_accounts",
    "telegram": "telegram_accounts",
    "webhook": "webhook_accounts",
    "github": "github_accounts",
}
DESTINATION_TABLES = tuple(DESTINATION_TABLES_BY_TYPE.values())

DESTINATION_LIMITS = {
    "mastodon": 500,
    "bluesky": 300,
    "discord": 2000,
    "telegram": 4096,
    "matrix": None,
    "email": None,
    "microblog": None,
    "webhook": None,
    "github": None,
}

OAUTH_SESSION_COOKIE = "feedecho_oauth_session"
OAUTH_SESSION_MAX_AGE = 10 * 60

# ── Optional shared-secret auth ──────────────────────────────────────────────
# If FEEDECHO_AUTH_TOKEN is set, all requests must include it as either:
#   - Cookie: feedecho_auth=<token>   (set by the login page)
#   - X-Auth-Token: <token>           (for API/programmatic access)
# If the env var is unset, auth is disabled (original behavior).
# (Single-tenant mode only; multi mode replaces this with sessions.)
import settings
from settings import AUTH_TOKEN  # noqa: F401  (re-exported for tests/legacy)

# Paths exempt from auth: health check + static files + OAuth callback.
# Only /oauth/callback must be reachable without a cookie (Mastodon redirects
# the user's browser here). /oauth/connect requires auth so unauthenticated
# users cannot trigger outbound requests to arbitrary instance URLs.
_AUTH_EXEMPT_PATHS = {"/healthz", "/favicon.svg", "/static", "/oauth/callback"}
_AUTH_EXEMPT_PREFIXES = ("/static", "/poke/")

# S10f: index the route table once, splitting literal paths (dict lookup of
# path -> methods) from parameterized ones (need the full matches() scan).
# Rebuilt only when the route list length changes (billing.mount appends at
# startup, before serving). Method-aware so a wrong-method request to a known
# path still falls through to the router's 405, matching the pre-cache behavior.
_route_index_cache: dict = {}


def _route_index(app):
    routes = app.routes
    key = id(app)
    cached = _route_index_cache.get(key)
    if cached is None or cached[0] != len(routes):
        literal: dict = {}
        parameterized: list = []
        for route in routes:
            path = getattr(route, "path", None)
            if not path or getattr(route, "matches", None) is None:
                continue
            methods = getattr(route, "methods", None)
            # Routes with no methods set (Mount for /static, WebSocket routes)
            # must go through the matches() scan, not the literal fast-path:
            # a prefix-matched Mount path would otherwise be dropped entirely
            # and change unauthenticated behaviour from redirect to 404.
            if not methods or "{" in path:
                parameterized.append(route)
                continue
            methods = set(methods)
            # Starlette treats a HEAD request as a match for a GET route.
            if "GET" in methods:
                methods.add("HEAD")
            literal.setdefault(path, set()).update(methods)
        cached = (len(routes), literal, parameterized)
        _route_index_cache[key] = cached
    return cached[1], cached[2]


class RequestIdMiddleware(BaseHTTPMiddleware):
    """Request-id threading + structured access log.

    Outermost middleware: accepts a client X-Request-ID (validated against
    a conservative charset + length cap; anything else is replaced with a
    generated uuid) and attaches it to every log record emitted while the
    request is handled, to the response header, and to one access-log line
    per request.

    The access line's ``ip=`` field is the DERIVED client IP (the rightmost
    X-Forwarded-For entry when the TCP peer is a trusted proxy), not the raw
    TCP peer: behind Caddy the peer is always the proxy container, which
    tells you nothing about who is hitting the box. See auth._client_ip.
    """

    _VALID_ID = re.compile(r"^[A-Za-z0-9._-]{1,64}$")

    async def dispatch(self, request: Request, call_next):
        raw = request.headers.get("x-request-id", "")
        request_id = raw if self._VALID_ID.fullmatch(raw) else uuid.uuid4().hex
        token = logging_setup.set_request_id(request_id)
        start = time.perf_counter()
        try:
            response = await call_next(request)
        except Exception:
            # Log a 500 access line while the id is still attached, then
            # reset scope-safely and re-raise (ServerErrorMiddleware logs
            # the traceback, and its record will carry the id).
            duration_ms = round((time.perf_counter() - start) * 1000, 1)
            client_ip = auth._client_ip(request)
            uid = getattr(request.state, "user_id", None)
            access_logger.error(
                "%s %s 500 %sms ip=%s%s (unhandled exception)",
                request.method,
                request.url.path,
                duration_ms,
                client_ip,
                f" user={uid}" if uid is not None else "",
            )
            logging_setup.reset_request_id(token)
            raise
        # Access log emitted while the request id is still attached to the
        # context, so its own record carries the id too.
        response.headers["X-Request-ID"] = request_id
        duration_ms = round((time.perf_counter() - start) * 1000, 1)
        client_ip = auth._client_ip(request)
        uid = getattr(request.state, "user_id", None)
        path = request.url.path
        log = (
            access_logger.debug
            if path == "/healthz"
            else access_logger.info
        )
        log(
            "%s %s %s %sms ip=%s%s",
            request.method,
            path,
            response.status_code,
            duration_ms,
            client_ip,
            f" user={uid}" if uid is not None else "",
        )
        logging_setup.reset_request_id(token)
        return response


class AuthMiddleware(BaseHTTPMiddleware):
    """Shared-secret auth (single mode) or session auth (multi mode).

    Single mode: FEEDECHO_AUTH_TOKEN unset → no-op; set → cookie/header
    token required. Multi mode: feedecho_session cookie required on all
    paths except the public set. Mode is read per request so tests can
    flip settings.MULTI without reloading the app.

    Beyond allow/deny it records ``request.state.authed`` (and, in multi
    mode, ``request.state.user_id``) so ``render()`` can give a viewer the
    chrome that matches what they can actually reach. That identification is
    best-effort on the PUBLIC pages too: without it a signed-in user lands on
    /about with an anonymous nav and no way back into the app.
    """

    _MULTI_EXEMPT_PATHS = {
        "/healthz",
        "/favicon.svg",
        "/oauth/callback",
        "/register",
        "/login",
        "/logout",
        # Email links are opened unauthenticated.
        "/verify-email",
        "/forgot-password",
        "/reset-password",
        # Public hosted-service disclosure page.
        "/about",
        # Pricing is a public decision page like the legal pages.
        "/pricing",
        # Legal pages are for people deciding whether to sign up.
        "/terms",
        "/privacy",
        # Hosted landing page: public for visitors, dashboard for authed
        # (the route branches on authed).
        "/",
    }
    _MULTI_EXEMPT_PREFIXES = ("/static", "/poke/")

    # Paths a card-pending trial user (registered but checkout unfinished) may
    # still reach so they can finish checkout, log out, or delete the abandoned
    # account. Everything else bounces to /settings#billing until the card is
    # collected (the webhook stamps the real trial_ends_at on the Stripe
    # subscription.created event, which clears the sentinel).
    _PENDING_CARD_ALLOWED = {
        "/settings",
        "/api/billing/checkout",
        "/api/billing/portal",
        "/settings/delete-account",
        "/logout",
    }

    # Exempt paths that still render the app's nav/footer, so the viewer is
    # worth identifying. Deliberately NOT the whole exempt set:
    # /oauth/callback authorizes on the signed OAuth state and its handler
    # documents that request.state.user_id is unset there; /logout must not
    # care who you are; /static, /healthz and /favicon.svg must not pay for a
    # database read. /forgot-password and /reset-password are for people who
    # cannot get in, where anonymous chrome is the honest answer.
    _MULTI_PUBLIC_PAGES = {
        "/login", "/register", "/about", "/verify-email", "/terms", "/privacy",
        "/pricing", "/"
    }

    async def dispatch(self, request: Request, call_next):
        if settings.MULTI:
            return await self._multi(request, call_next)
        return await self._single(request, call_next)

    @staticmethod
    def _path_is_routed(request: Request) -> bool:
        """Whether any registered route matches this request's path/method.

        Unknown paths must reach the router so the 404 handler renders the
        real not-found page. Without this, an unauthenticated browser request
        to a nonexistent URL was silently redirected to /login, so a visitor
        following a bad link could not tell "page doesn't exist" from
        "session expired" (impeccable critique 2026-08-29, P1a).

        S10f: literal (no-path-param) routes are checked against a cached
        path->methods dict first — that is the hot path for most requests. The
        `matches()` scan only runs for parameterized routes, which is a
        handful.
        """
        literal, parameterized = _route_index(request.app)
        methods = literal.get(request.url.path)
        if methods and request.method in methods:
            return True
        scope = request.scope
        for route in parameterized:
            matches = getattr(route, "matches", None)
            if matches is None:
                continue
            try:
                # starlette Route.matches returns a (Match, path_params) tuple.
                if matches(scope)[0] == Match.FULL:
                    return True
            except Exception:  # defensive: never break auth over a route table quirk
                continue
        return False

    @staticmethod
    def _token_matches(request: Request) -> bool:
        """Whether this request carries the shared secret.

        Pure comparison, no database, so it is cheap enough to run on every
        request including the exempt ones — and single mode has no user
        attribution, so knowing "this is the operator" everywhere cannot
        change any authorization decision.
        """
        token = (
            request.cookies.get(auth.AUTH_COOKIE_NAME)
            or request.headers.get("x-auth-token")
        )
        if not token or not settings.AUTH_TOKEN:
            return False
        return (
            # Compare bytes: compare_digest raises TypeError on str arguments
            # containing non-ASCII, which would surface as a 500 from a cookie
            # an attacker (or a confused browser) fully controls.
            _secrets.compare_digest(
                token.encode("utf-8", "surrogatepass"),
                settings.AUTH_TOKEN.encode("utf-8", "surrogatepass"),
            )
        )

    async def _single(self, request: Request, call_next):
        path = request.url.path
        # No shared secret configured: every viewer is the operator — but
        # only when the operator explicitly opted out via
        # FEEDECHO_ALLOW_INSECURE=1. validate_config enforces that pairing at
        # startup; this branch is the defense-in-depth backstop for an ASGI
        # server started with lifespan events disabled (--lifespan off),
        # which would skip validate_config entirely (gate SEC-01). Fail
        # closed rather than fall back to open access.
        if not settings.AUTH_TOKEN and not settings.ALLOW_INSECURE:
            return Response(
                "Authentication misconfigured: FEEDECHO_AUTH_TOKEN is not "
                "set. Set it, or set FEEDECHO_ALLOW_INSECURE=1 to run "
                "without authentication.",
                status_code=500,
            )
        if not settings.AUTH_TOKEN or self._token_matches(request):
            request.state.authed = True
            if path == "/login" and request.method == "GET":
                # Nothing to log into: either no token is configured or this
                # viewer already presented it. A login form here is a dead end.
                return RedirectResponse(url="/", status_code=302)
            return await call_next(request)

        # Health check, static files, the OAuth callback and the login form
        # itself stay reachable without the token.
        if (
            path in _AUTH_EXEMPT_PATHS
            or path.startswith(tuple(_AUTH_EXEMPT_PREFIXES))
            or path == "/login"
        ):
            return await call_next(request)

        # Unknown paths get the real 404 page instead of a login redirect.
        if not self._path_is_routed(request):
            return await call_next(request)

        # Redirect browser requests to login, 401 for API/JSON
        accept = request.headers.get("accept", "")
        if "text/html" in accept and request.method == "GET":
            return RedirectResponse(url="/login", status_code=302)
        return JSONResponse(
            {"detail": "Authentication required"}, status_code=401
        )

    @staticmethod
    def _session_user(request: Request) -> int | None:
        """The user behind a valid session cookie, or None.

        Suspension and session-epoch are enforced per request, not just at
        login: a valid HMAC session for a suspended account, or one issued
        before the last password reset, is rejected.
        """
        token = request.cookies.get("feedecho_session")
        claims = security.read_session(token) if token else None
        if not claims:
            return None
        with get_db() as db:
            row = db.execute(
                "SELECT suspended, session_epoch FROM users WHERE id = ?",
                (claims["user_id"],),
            ).fetchone()
        if (
            row
            and not row["suspended"]
            and row["session_epoch"] == claims.get("epoch", 0)
        ):
            return claims["user_id"]
        return None

    @staticmethod
    def _pending_card(uid: int) -> bool:
        """True when a billing account has registered but not collected a card.

        The register flow writes the TRIAL_PENDING sentinel as trial_ends_at;
        the Stripe subscription webhook replaces it with a real date once the
        card is collected. Until then the account must be gated off the app so
        a bot that abandons checkout cannot use it (posting is paused, but it
        should not be able to poke at feeds/echoes/accounts either).
        """
        if not settings.BILLING_ENABLED:
            return False
        with get_db() as db:
            row = db.execute(
                "SELECT plan, trial_ends_at FROM users WHERE id = ?", (uid,)
            ).fetchone()
        return bool(
            row
            and row["plan"] == "trial"
            and plans.trial_pending(row["trial_ends_at"])
        )

    async def _multi(self, request: Request, call_next):
        path = request.url.path
        if path in self._MULTI_EXEMPT_PATHS or path.startswith(
            tuple(self._MULTI_EXEMPT_PREFIXES)
        ):
            if path in self._MULTI_PUBLIC_PAGES:
                uid = self._session_user(request)
                if uid is not None:
                    request.state.user_id = uid
                    request.state.authed = True
                    if path == "/" and self._pending_card(uid):
                        # The dashboard is the landing page only for anonymous
                        # viewers; a card-pending user must finish checkout
                        # before seeing the app.
                        return RedirectResponse(url="/settings#billing", status_code=302)
                    if path in ("/login", "/register") and request.method == "GET":
                        return RedirectResponse(url="/", status_code=302)
            return await call_next(request)

        uid = self._session_user(request)
        if uid is not None:
            request.state.user_id = uid
            request.state.authed = True
            if self._pending_card(uid) and path not in self._PENDING_CARD_ALLOWED:
                return RedirectResponse(url="/settings#billing", status_code=302)
            return await call_next(request)

        # Unknown paths get the real 404 page instead of a login redirect.
        if not self._path_is_routed(request):
            return await call_next(request)

        accept = request.headers.get("accept", "")
        if "text/html" in accept and request.method == "GET":
            return RedirectResponse(url="/login", status_code=302)
        return JSONResponse(
            {"detail": "Authentication required"}, status_code=401
        )


@app.get("/login", response_class=HTMLResponse)
async def login_page(request: Request):
    return auth.login_page(request)


@app.post("/login")
def login_submit(
    request: Request,
    email: str = Form(""),
    password: str = Form(""),
    token: str = Form(""),
):
    return auth.login_submit(request, email=email, password=password, token=token)


@app.get("/register", response_class=HTMLResponse)
async def register_page(request: Request):
    return auth.register_page(request)


@app.post("/register")
def register_submit(
    request: Request,
    email: str = Form(...),
    password: str = Form(...),
    confirm: str = Form(...),
    invite_code: str = Form(""),
):
    return auth.register_submit(
        request, email=email, password=password, confirm=confirm,
        invite_code=invite_code,
    )


@app.post("/logout")
def logout(request: Request):
    return auth.logout(request)


@app.get("/forgot-password", response_class=HTMLResponse)
async def forgot_page(request: Request):
    return auth.forgot_page(request)


@app.post("/forgot-password")
async def forgot_submit(request: Request, email: str = Form("")):
    return auth.forgot_submit(request, email=email)


@app.get("/reset-password", response_class=HTMLResponse)
async def reset_page(request: Request, token: str = ""):
    return auth.reset_page(request, token=token)


@app.post("/reset-password")
def reset_submit(
    request: Request,
    token: str = Form(""),
    password: str = Form(""),
    confirm: str = Form(""),
):
    return auth.reset_submit(
        request, token=token, password=password, confirm=confirm
    )


# ── S8: Security headers middleware ──────────────────────────────────────────
#
# Progressive CSP: frame-ancestors blocks clickjacking (the S9 half);
# script-src 'unsafe-inline' is a concession to the 42 inline onclick handlers
# and 3 inline <script> blocks that predate this release.  A strict nonce-based
# CSP requires migrating all of those to addEventListener / external files —
# deferred to the A-batch a11y/refactor work.
#
# form-action vs. hosted billing: the Subscribe / register forms POST to
# /api/billing/checkout and /api/billing/portal, which 303-redirect to
# Stripe-hosted pages (checkout.stripe.com / billing.stripe.com).  The
# browser enforces form-action on EVERY hop of a form submission's redirect
# chain — the form's action URL AND each redirect target — so an origin the
# user's browser must be redirected to has to be listed here.  The original
# form-action 'self' (S8, v1.42.0) assumed the 302 was exempt; it is not,
# and every card flow (register -> checkout, Subscribe, Manage billing)
# was silently blocked in the browser until 2026-09-10.  The Stripe origins
# are added only when the billing seam is on, so self-hosted deployments
# keep the strict form-action 'self'.

_CSP_BASE = (
    "default-src 'self'; "
    "script-src 'self' 'unsafe-inline'; "
    "style-src 'self' 'unsafe-inline'; "
    "font-src 'self'; "
    "img-src 'self' data: https: http:; "
    "connect-src 'self'; "
    "frame-ancestors 'none'; "
    "frame-src 'none'; "
    "object-src 'none'; "
    "base-uri 'self'"
)

# Stripe-hosted origins the billing redirect chain navigates to: hosted
# Checkout sessions and Customer Portal sessions.  DO NOT tighten without a
# browser-level test of the checkout redirect (the v1.42.0 regression).
_BILLING_FORM_ACTION_ORIGINS = (
    "https://checkout.stripe.com https://billing.stripe.com"
)


# The accounts-page connect form is a form submission (GET forms count)
# whose redirect chain lands on the user-chosen Mastodon instance's
# /oauth/authorize. The browser enforces form-action on EVERY hop of a
# form submission's redirect chain, and that origin is unbounded, so the
# directive has to allow https: at scheme level. Without it, connecting
# ANY instance was silently blocked client-side (reported 2026-09-10,
# same defect class as the Stripe form-action regression below).
#
# The wide directive is scoped to the pages that render the connect form
# (/accounts, and /oauth/connect for its error banner): form-action exists
# to stop cross-origin form hijacking on credential forms (login, register,
# admin), and those pages have no business posting anywhere off-origin.
# Review-gate finding (Gemini 3.8 Flash, 2026-09-10): the first cut applied
# it to every response.
_FORM_ACTION_WIDE_PAGES = ("/accounts", "/oauth/connect")


def _csp_header(path: str | None = None) -> str:
    """Compose the CSP for a response.

    Reads settings.BILLING_ENABLED per response rather than baking it in at
    import: the flag is env-fixed in production, and per-request composition
    keeps tests able to flip it without reimporting the app.
    """
    wide = path is not None and path.rstrip("/") in _FORM_ACTION_WIDE_PAGES
    form_action = "form-action 'self' https:" if wide else "form-action 'self'"
    if settings.BILLING_ENABLED:
        form_action += " " + _BILLING_FORM_ACTION_ORIGINS
    return f"{_CSP_BASE}; {form_action}"


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Sets security headers on every response (S8)."""

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)
        h = response.headers
        if "X-Content-Type-Options" not in h:
            h["X-Content-Type-Options"] = "nosniff"
        if "X-Frame-Options" not in h:
            h["X-Frame-Options"] = "DENY"
        if "Referrer-Policy" not in h:
            # strict-origin-when-cross-origin, NOT no-referrer: a no-referrer
            # policy makes browsers send "Origin: null" on same-origin form
            # POSTs (WHATWG Fetch), which breaks Origin-based CSRF checks and
            # locked every mobile login out with a bare 403. This still sends
            # only the origin cross-site (no path/query leakage) while keeping
            # a real Origin on same-origin posts.
            h["Referrer-Policy"] = "strict-origin-when-cross-origin"
        if "Permissions-Policy" not in h:
            h["Permissions-Policy"] = (
                "camera=(), microphone=(), geolocation=(), payment=()"
            )
        if "Content-Security-Policy" not in h:
            # The wide form-action (https:) is needed only on the pages that
            # render the Mastodon connect form — see _FORM_ACTION_WIDE_PAGES.
            h["Content-Security-Policy"] = _csp_header(request.url.path)
        # Emit HSTS unconditionally: browsers ignore the header over plain HTTP
        # (RFC 6797), so it is safe on http:// and correct on https://. Gating on
        # request.url.scheme silently fails behind a TLS-terminating reverse
        # proxy (the app then reads scheme=http even though the client is on
        # https) — exactly the deployments that need it. Caddy sets its own HSTS
        # for the hosted service; this covers self-hosters behind their own TLS.
        if "Strict-Transport-Security" not in h:
            h["Strict-Transport-Security"] = "max-age=31536000"
        return response


# ── S9: CSRF Origin check on state-changing methods ──────────────────────────
#
# The session cookie is SameSite=Lax, which already blocks most CSRF in modern
# browsers.  This is the recommended second layer: for unsafe methods, a
# cross-origin Origin (or, as fallback, Referer) is rejected.  "Origin: null"
# is NOT automatically rejected: browsers legitimately send it on a same-origin
# form POST when a referrer policy hides the referrer (e.g. Firefox after a
# POST redirect, or privacy tools forcing no-referrer) — rejecting it outright
# locked every mobile login out with a bare 403.  Fetch Metadata (Sec-Fetch-Site,
# set by the browser, unforgeable cross-site) disambiguates: a null Origin with
# Sec-Fetch-Site: same-origin is a genuine same-origin post and is accepted; a
# null Origin without that evidence falls back to Referer, and fails closed.
# Neither header present → allow (non-browser clients like webhooks, curl, and
# single-mode API scripts).

class CSRFOriginMiddleware(BaseHTTPMiddleware):
    """Origin/Referer check on state-changing requests (S9)."""

    _UNSAFE = frozenset({"POST", "PUT", "PATCH", "DELETE"})

    async def dispatch(self, request: Request, call_next):
        if request.method in self._UNSAFE and not self._same_origin(request):
            # A bare Response(status_code=403) previously returned here with
            # no body and no Content-Type, bypassing forbidden_handler and
            # every other error page in the app — a legitimate user whose
            # form got cross-origin-rejected (see the false-positive-prone
            # heuristics documented above) saw a blank page with zero
            # explanation. Error-handling audit finding 1.4
            # (docs/reviews/2026-09-08-error-handling-audit.md).
            if "text/html" in request.headers.get("accept", ""):
                return render(
                    "error.html", request, status_code=403, code=403,
                    message="Your request could not be verified. Please try again.",
                )
            return JSONResponse({"detail": "Cross-origin request rejected"}, status_code=403)
        return await call_next(request)

    @staticmethod
    def _same_origin(request: Request) -> bool:
        req_host = (request.url.hostname or "").lower()
        origin = request.headers.get("origin")
        if origin is not None and origin.strip().lower() != "null":
            return _host_matches(origin, req_host)
        if origin is not None:
            # origin == "null". Trust the browser's own Fetch Metadata claim
            # that this is same-origin; otherwise fall through to Referer.
            if request.headers.get("sec-fetch-site", "").strip().lower() == "same-origin":
                return True
        referer = request.headers.get("referer")
        if referer is not None:
            return _host_matches(referer, req_host)
        # No Origin, no Referer → non-browser client. An explicit "null"
        # Origin with no same-origin evidence and no Referer fails closed.
        return origin is None


def _host_matches(url_val: str, expect_host: str) -> bool:
    if not url_val or url_val.strip().lower() == "null":
        return False
    if not expect_host:
        # No request Host to compare against (e.g. a malformed HTTP/1.0
        # request): fail closed rather than letting two empty hosts match.
        return False
    try:
        host = (urlsplit(url_val).hostname or "").lower()
    except ValueError:
        return False
    return host == expect_host


app.add_middleware(AuthMiddleware)
app.add_middleware(CSRFOriginMiddleware)
app.add_middleware(SecurityHeadersMiddleware)
# Outermost: request-id threading + access logging wraps everything.
app.add_middleware(RequestIdMiddleware)


def _trial_context(request: Request) -> dict:
    """Multi-mode template context: current user's email, plan, trial state.

    Returns {} in single mode or when no authenticated user is present
    (exempt routes like /login and /register).
    """
    if not settings.MULTI:
        return {}
    uid = getattr(request.state, "user_id", None)
    if uid is None:
        return {}
    with get_db() as db:
        # subscription_status is a billing-overlay column; tests that flip
        # BILLING_ENABLED without the billing module won't have it. Include it
        # in the SELECT only when the column actually exists.
        has_sub_status = settings.BILLING_ENABLED and "subscription_status" in _column_names(
            db, "users"
        )
        if has_sub_status:
            row = db.execute(
                "SELECT email, plan, trial_ends_at, is_admin, email_verified,"
                " subscription_status FROM users WHERE id = ?",
                (uid,),
            ).fetchone()
        else:
            row = db.execute(
                "SELECT email, plan, trial_ends_at, is_admin, email_verified"
                " FROM users WHERE id = ?",
                (uid,),
            ).fetchone()
    if not row:
        return {}
    plan = row["plan"] or "trial"
    # A card-on-file subscription carries 'trialing' during the free trial,
    # which means the user has ALREADY subscribed and will renew automatically
    # — they must not be shown a "Subscribe" CTA.
    subscription_status = row["subscription_status"] if has_sub_status else None
    ctx = {
        "current_user_email": row["email"],
        "plan": plan,
        "is_admin": bool(row["is_admin"]),
        "email_verified": bool(row["email_verified"]),
        "subscription_status": subscription_status,
        # Every page can render an accurate paused-posting banner (the
        # scheduler actually skips this user's feeds, so the banner must
        # agree with reality, not just the dashboard's).
        "posting_paused": plans.posting_paused(plan, row["trial_ends_at"]),
        "plan_limits": settings.PLAN_LIMITS.get(plan) or settings.PLAN_LIMITS["trial"],
        # True only when the hosted deployment has mounted its private billing
        # module — templates gate the "Subscribe"/"Manage" UI on this flag.
        "billing_enabled": settings.BILLING_ENABLED,
    }
    # Trial users see their limits and where they stand against them on the
    # dashboard (the banner lives in dashboard.html). One cheap count query
    # for feeds + destinations only when the plan is a trial; paid/beta
    # never pay for it.
    if plan == "trial":
        with get_db() as db:
            feeds_used = db.execute(
                "SELECT COUNT(*) AS c FROM feeds WHERE user_id = ? AND deleted_at IS NULL",
                (uid,),
            ).fetchone()["c"]
            dest_used = sum(
                db.execute(
                    f"SELECT COUNT(*) AS c FROM {t} WHERE user_id = ?", (uid,)
                ).fetchone()["c"]
                for t in DESTINATION_TABLES
            )
        ctx["trial_usage"] = {
            "feeds": feeds_used,
            "max_feeds": plans.limit_for(plan, "max_feeds"),
            "destinations": dest_used,
            "max_destinations": plans.limit_for(plan, "max_destinations"),
            "min_poll_interval": plans.limit_for(plan, "min_poll_interval"),
            "max_posts_per_hour": plans.limit_for(plan, "max_posts_per_hour"),
        }
    ends = row["trial_ends_at"]
    if ends:
        if plans.trial_pending(ends):
            # Card-gated trial not yet started (checkout unfinished). Posting
            # is paused, but this is NOT "trial ended" — the clock never ran.
            ctx["trial_pending"] = True
        else:
            end = plans.normalize_trial_ends(ends)
            if end is not None:
                now = datetime.now(timezone.utc)
                if end <= now:
                    ctx["trial_expired"] = True
                else:
                    ctx["trial_days_left"] = max(1, (end - now).days)
                    # ISO date for a <time class="local-time"> element; the
                    # browser renders it in the viewer's locale (issue #6).
                    ctx["trial_ends_date"] = end.strftime("%Y-%m-%d")
            else:
                logger.warning("Unparseable trial_ends_at for user %s: %r", uid, ends)
    return ctx


def _user_plan(db, uid: int) -> str:
    """The plan string for a user row ('trial' when unset/missing)."""
    row = db.execute("SELECT plan FROM users WHERE id = ?", (uid,)).fetchone()
    return (row["plan"] if row else None) or "trial"


def _require_reader(db, uid: int) -> None:
    """402 in multi mode when the plan does not include the RSS reader.

    Single mode never raises (no plan gating). Every reader route must call
    this so a future route cannot forget the entitlement check.
    """
    if settings.MULTI and not plans.reader_enabled(_user_plan(db, uid)):
        raise HTTPException(status_code=402, detail="The RSS reader requires a paid plan")


def _check_destination_cap(db, uid: int) -> None:
    """Raise PlanError when one more connected account exceeds the plan.

    The cap counts ALL destination types: every row is a stored credential
    and an outbound-posting path. HTTPException(402) here; PlanError directly
    for non-HTTP callers.
    """
    if not settings.MULTI:
        return
    total = sum(
        db.execute(
            f"SELECT COUNT(*) AS c FROM {t} WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        for t in DESTINATION_TABLES
    )
    plans.check_destination_allowance(total, _user_plan(db, uid))


def render(name: str, request: Request, status_code: int = 200, **kwargs) -> HTMLResponse:
    template = jinja.get_template(name)
    authed = bool(getattr(request.state, "authed", False))
    context = {
        "MULTI": settings.MULTI,
        "authed": authed,
        "project_url": settings.PROJECT_URL,
    }
    if settings.MULTI:
        context.update(_trial_context(request))
    # The running version, in the footer, so whoever is responsible for
    # updating this deployment can see whether it is behind (issue #8).
    # Self-hosted: anyone past the auth gate is the operator. Hosted: admins
    # only — a tenant cannot upgrade the service, and /register is public, so
    # gating on "logged in" would hand the exact version to anyone willing to
    # sign up. Anonymous viewers never see it: the middleware only marks a
    # request authed after the session or token checked out, on public pages
    # included.
    if authed and (bool(context.get("is_admin")) if settings.MULTI else True):
        context["app_version"] = APP_VERSION
    context.update(kwargs)
    return HTMLResponse(
        template.render(request=request, **context), status_code=status_code
    )


def validate_url(url: str) -> str:
    """Validate a URL has http or https scheme and is safe for outbound requests.

    Combines scheme check with SSRF protection (blocks private IPs,
    internal hostnames, non-http schemes, embedded credentials).
    Malformed URLs that make urllib.parse.urlsplit itself raise ValueError
    (unclosed IPv6 brackets like "https://[::1") are re-raised as 400s:
    previously they escaped as an unhandled 500 (review-gate finding,
    2026-09-10; the same ValueError path is reachable from the manual
    add-account form).
    """
    if not re.match(r"^https?://", url):
        raise HTTPException(status_code=400, detail="URL must start with http:// or https://")
    try:
        validate_outbound_url(url)
    except SSRFError as e:
        # SSRFError subclasses ValueError, so this clause MUST come first:
        # an SSRF rejection should read as one, not as generic malformedness.
        raise HTTPException(status_code=400, detail=str(e))
    except ValueError as e:
        raise HTTPException(status_code=400, detail=f"URL is malformed: {e}") from e
    return url.rstrip("/")


from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    # uvicorn's configure_logging() runs at server start — AFTER app import
    # under `python app.py` or uvicorn.run() — and resets uvicorn.access back
    # to INFO, re-enabling full-request-line (query-string token) logging.
    # Re-silence it here so the launch path cannot matter. Docker CMDs also
    # pass --no-access-log; this is the catch-all for every other path.
    logging.getLogger("uvicorn.access").setLevel(logging.CRITICAL)
    settings.validate_config()
    _assert_billing_mounted(app)
    init_db()
    _bootstrap_admin()
    _revalidate_stored_templates()
    start_scheduler()
    logger.info("FeedEcho started")
    yield
    stop_scheduler()


def _assert_billing_mounted(app: FastAPI) -> None:
    """Fail fast if billing is advertised but its checkout route is absent.

    When FEEDECHO_BILLING_ENABLED=1, registration 302-redirects into
    /api/billing/checkout (auth.py) and every signup lands there. If the
    hosted billing module failed to mount — or the wrong entrypoint was
    served — that redirect hits a 404 and every new user is stranded
    silently at a paused trial with no error and no alert. The route path
    is already hard-coded in auth.py, so checking it here adds no new
    coupling; it just turns that silent failure into a startup refusal.
    """
    if not settings.BILLING_ENABLED:
        return
    paths = {getattr(r, "path", None) for r in app.routes}
    if "/api/billing/checkout" not in paths:
        raise RuntimeError(
            "FEEDECHO_BILLING_ENABLED=1 but /api/billing/checkout is not "
            "registered — the billing module did not mount. Refusing to start "
            "(serve the hosted entrypoint, or disable billing)."
        )


def _bootstrap_admin() -> None:
    """Promote the user named by FEEDECHO_ADMIN_EMAIL (idempotent).

    The hosted deployment has no other bootstrap path: the first admin is
    promoted from the environment, then manages the rest from the admin
    dashboard. No-op when unset or in single mode. Email comparison is
    case-insensitive (registration normalizes to lowercase). Note: while
    the env var is set, the named account is re-promoted on every startup
    — an intentional recovery hatch, so a dashboard demotion of that
    account is only permanent after removing the env var.
    """
    if not settings.MULTI or not settings.ADMIN_EMAIL:
        return
    with get_db() as db:
        row = db.execute(
            "SELECT id, is_admin FROM users WHERE LOWER(email) = LOWER(?)",
            (settings.ADMIN_EMAIL,),
        ).fetchone()
        if row is None:
            logger.warning(
                "FEEDECHO_ADMIN_EMAIL matches no registered user yet"
            )
            return
        if row["is_admin"]:
            return
        db.execute(
            "UPDATE users SET is_admin = 1 WHERE id = ?", (row["id"],)
        )
    logger.info("Promoted FEEDECHO_ADMIN_EMAIL user (id %s) to admin", row["id"])


app.router.lifespan_context = lifespan


# ── Helpers ──────────────────────────────────────────────────────────────────

def _get_smtp_settings(mask_password: bool = False, user_id: int = 1):
    """Load SMTP settings as a flat dict for templates.

    If mask_password is True, replaces the SMTP password with a placeholder
    so it's never sent to the browser. Used on settings/accounts pages.
    """
    with get_db() as db:
        rows = db.execute(
            "SELECT key, value FROM settings WHERE key LIKE 'smtp_%' AND user_id = ?",
            (user_id,),
        ).fetchall()
    if not rows:
        return {}
    smtp = {row["key"]: row["value"] for row in rows}
    if mask_password and smtp.get("smtp_password"):
        smtp["smtp_password"] = "********"
    return smtp


def _dependent_echo_count_tx(db, user_id: int, destination_type: str, account_id: int) -> int:
    """Live echoes still pointing at this destination, on a caller-owned connection.

    Deleting a destination out from under an echo leaves it aimed at a row
    that no longer exists, and delivery then fails at run time with nothing
    said at delete time. The Bluesky delete guarded against this from the
    start; Mastodon and email did not.

    Every delete route runs this check and the row DELETE on the same `db`
    inside one `with get_db() as db:` block, so the count and the delete are
    atomic — a concurrent request can't create a new echo against this
    destination in between and slip past the check.
    """
    return db.execute(
        """
        SELECT COUNT(*) AS c FROM echoes
         WHERE destination_type = ?
           AND destination_id = ?
           AND deleted_at IS NULL
           AND user_id = ?
        """,
        (destination_type, account_id, user_id),
    ).fetchone()["c"]


def _dependent_echo_count(user_id: int, destination_type: str, account_id: int) -> int:
    """Live echoes still pointing at this destination.

    Convenience wrapper over :func:`_dependent_echo_count_tx` that opens its
    own connection. Callers that must keep the check and a subsequent delete
    atomic (all of the destination-delete routes) should call
    ``_dependent_echo_count_tx`` directly inside their own ``get_db()`` block
    instead of this function.
    """
    with get_db() as db:
        return _dependent_echo_count_tx(db, user_id, destination_type, account_id)


def _render_oauth_error(request: Request, message: str) -> HTMLResponse:
    """Render a standalone OAuth failure page.

    /oauth/callback is in the multi-mode exempt path set, so the request has
    no authenticated user attached and anything that calls current_user_id()
    (like _render_accounts_error) raises 401 instead of showing the error.
    """
    return render(
        "error.html",
        request,
        status_code=400,
        code="OAuth failed",
        message=f"{message} Return to the accounts page to try again.",
    )


def _webhook_url_display(url: str) -> str:
    """Origin only: scheme + host (+ port), nothing else.

    Webhook URLs carry secrets in more places than the query string — Slack
    puts the token in the PATH (/services/T.../B.../xxxx) — so the whole URL
    beyond the origin is treated as sensitive and never rendered.
    """
    from urllib.parse import urlsplit

    parts = urlsplit(url or "")
    host = parts.hostname or ""
    if parts.port:
        return f"{parts.scheme}://{host}:{parts.port}"
    return f"{parts.scheme}://{host}"


def _mask_webhook_urls(rows, type_col: str = "destination_type", url_col: str = "instance"):
    """Return rows as dicts with webhook URLs stripped of query/fragment.

    Webhook URLs may embed a secret query token; the history/dashboard
    instance column must never render it. Non-webhook rows pass through
    unchanged (dictified for uniform template access).
    """
    out = []
    for row in rows:
        as_dict = dict(row)
        if as_dict.get(type_col) == "webhook":
            as_dict[url_col] = _webhook_url_display(as_dict.get(url_col, ""))
        out.append(as_dict)
    return out


def _get_all_accounts(user_id: int = 1):
    """Fetch Mastodon, email, Bluesky, micro.blog, Matrix, Discord, and
    webhook accounts for one user."""
    with get_db() as db:
        mastodon = db.execute(
            "SELECT id, name, username, instance, booster_enabled, created_at FROM accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        email = db.execute(
            "SELECT id, name, email, created_at FROM email_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        bluesky = db.execute(
            "SELECT id, name, handle, did, pds, created_at FROM bluesky_accounts"
            " WHERE user_id = ? ORDER BY handle",
            (user_id,),
        ).fetchall()
        microblog = db.execute(
            "SELECT id, name, uid, created_at FROM microblog_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        matrix_rows = db.execute(
            "SELECT id, name, homeserver, room_id, room_alias, matrix_user_id,"
            " created_at FROM matrix_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        # webhook_url is the credential — never selected for display.
        discord_rows = db.execute(
            "SELECT id, name, channel_id, created_at FROM discord_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        # bot_token is the credential — never selected for display.
        telegram_rows = db.execute(
            "SELECT id, name, chat_id, created_at FROM telegram_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
        # Webhook rows: URL shown without query/fragment (secret tokens live
        # there) and headers shown by NAME only — values are credentials.
        webhook_rows = []
        for row in db.execute(
            "SELECT id, name, url, headers, body_template, created_at FROM webhook_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall():
            header_names = ", ".join(load_headers(row["headers"]).keys())
            has_body = bool((row["body_template"] if "body_template" in row.keys() else "") or "")
            webhook_rows.append(
                {
                    "id": row["id"],
                    "name": row["name"],
                    "url_display": _webhook_url_display(row["url"]),
                    "header_names": header_names,
                    "has_body": has_body,
                    "created_at": row["created_at"],
                }
            )
        # GitHub rows: the token is the credential — never selected for
        # display, so the accounts table can only ever show repo + branch.
        github_rows = db.execute(
            "SELECT id, name, repo, branch, created_at FROM github_accounts"
            " WHERE user_id = ? ORDER BY name",
            (user_id,),
        ).fetchall()
    return mastodon, email, bluesky, microblog, matrix_rows, discord_rows, telegram_rows, webhook_rows, github_rows


def _shout_destinations(user_id: int):
    """One option per connected account, encoded ``type:id`` for the reader's
    Compose/Shout picker. Labels carry the type so a destination is unambiguous.
    Includes per-destination character/grapheme limit.

    GitHub is deliberately excluded: a Compose post has no per-echo file path
    template, so there is no filename to write to. That deviation is documented
    in the README's GitHub section; the picker does not offer a broken option.
    """
    (mastodon, email, bluesky, microblog, matrix_rows, discord_rows, telegram_rows, webhook_rows, _github_rows) = _get_all_accounts(user_id)
    out = []
    for row in mastodon:
        out.append({"value": f"mastodon:{row['id']}", "label": f"Mastodon · {row['name']}", "limit": DESTINATION_LIMITS.get("mastodon")})
    for row in email:
        out.append({"value": f"email:{row['id']}", "label": f"Email · {row['name']}", "limit": DESTINATION_LIMITS.get("email")})
    for row in bluesky:
        out.append({"value": f"bluesky:{row['id']}", "label": f"Bluesky · {row['name']}", "limit": DESTINATION_LIMITS.get("bluesky")})
    for row in microblog:
        out.append({"value": f"microblog:{row['id']}", "label": f"Micro.blog · {row['name']}", "limit": DESTINATION_LIMITS.get("microblog")})
    for row in matrix_rows:
        out.append({"value": f"matrix:{row['id']}", "label": f"Matrix · {row['name']}", "limit": DESTINATION_LIMITS.get("matrix")})
    for row in discord_rows:
        out.append({"value": f"discord:{row['id']}", "label": f"Discord · {row['name']}", "limit": DESTINATION_LIMITS.get("discord")})
    for row in telegram_rows:
        out.append({"value": f"telegram:{row['id']}", "label": f"Telegram · {row['name']}", "limit": DESTINATION_LIMITS.get("telegram")})
    for row in webhook_rows:
        out.append({"value": f"webhook:{row['id']}", "label": f"Webhook · {row['name']}", "limit": DESTINATION_LIMITS.get("webhook")})
    return out


def _render_accounts_error(request: Request, message: str) -> HTMLResponse:
    """Render the accounts page with an error banner."""
    uid = current_user_id(request)
    (
        mastodon_accounts,
        email_accounts,
        bluesky_accounts,
        microblog_accounts,
        matrix_accounts,
        discord_accounts,
        telegram_accounts,
        webhook_accounts,
        github_accounts,
    ) = _get_all_accounts(uid)
    smtp_settings = _get_smtp_settings(mask_password=True, user_id=uid)
    return render(
        "accounts.html",
        request,
        mastodon_accounts=mastodon_accounts,
        email_accounts=email_accounts,
        bluesky_accounts=bluesky_accounts,
        microblog_accounts=microblog_accounts,
        matrix_accounts=matrix_accounts,
        discord_accounts=discord_accounts,
        telegram_accounts=telegram_accounts,
        webhook_accounts=webhook_accounts,
        github_accounts=github_accounts,
        smtp_configured=bool(smtp_settings.get("smtp_host")),
        smtp_settings=smtp_settings,
        booster_configured=bool(settings.BOOSTER_URL and settings.BOOSTER_TOKEN),
        error=message,
    )


# ── Pages ───────────────────────────────────────────────────────────────────

@app.get("/", response_class=HTMLResponse)
async def dashboard(request: Request):
    # Hosted: / is the public landing page for visitors; signed-in users get
    # the dashboard. Single mode has no marketing surface — / is the app.
    if settings.MULTI and not bool(getattr(request.state, "authed", False)):
        return render("landing.html", request)
    uid = current_user_id(request)
    with get_db() as db:
        mastodon_accounts = db.execute(
            "SELECT COUNT(*) as c FROM accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        email_accounts = db.execute(
            "SELECT COUNT(*) as c FROM email_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        bluesky_accounts = db.execute(
            "SELECT COUNT(*) as c FROM bluesky_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        microblog_accounts = db.execute(
            "SELECT COUNT(*) as c FROM microblog_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        matrix_accounts = db.execute(
            "SELECT COUNT(*) as c FROM matrix_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        discord_accounts = db.execute(
            "SELECT COUNT(*) as c FROM discord_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        telegram_accounts = db.execute(
            "SELECT COUNT(*) as c FROM telegram_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        webhook_accounts = db.execute(
            "SELECT COUNT(*) as c FROM webhook_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        github_accounts = db.execute(
            "SELECT COUNT(*) as c FROM github_accounts WHERE user_id = ?", (uid,)
        ).fetchone()["c"]
        feeds = db.execute(
            "SELECT * FROM feeds WHERE deleted_at IS NULL AND user_id = ? ORDER BY name",
            (uid,),
        ).fetchall()
        echoes = db.execute("""
            SELECT e.*, f.name as feed_name,
                   CASE
                     WHEN e.destination_type = 'mastodon' THEN '@' || a.username || '@' || REPLACE(a.instance, 'https://', '')
                     WHEN e.destination_type = 'email' THEN ea.name || ' (' || ea.email || ')'
                     WHEN e.destination_type = 'bluesky' THEN '@' || b.handle
                     WHEN e.destination_type = 'microblog' THEN mb.name
                     WHEN e.destination_type = 'matrix' THEN mx.name
                     WHEN e.destination_type = 'discord' THEN dc.name
                     WHEN e.destination_type = 'telegram' THEN tg.name
                     WHEN e.destination_type = 'webhook' THEN wh.name
                     WHEN e.destination_type = 'github' THEN gh.name
                   END as destination_name
            FROM echoes e
            JOIN feeds f ON e.feed_id = f.id
            LEFT JOIN accounts a ON e.destination_type = 'mastodon' AND e.destination_id = a.id AND a.user_id = e.user_id
            LEFT JOIN email_accounts ea ON e.destination_type = 'email' AND e.destination_id = ea.id AND ea.user_id = e.user_id
            LEFT JOIN bluesky_accounts b ON e.destination_type = 'bluesky' AND e.destination_id = b.id AND b.user_id = e.user_id
            LEFT JOIN microblog_accounts mb ON e.destination_type = 'microblog' AND e.destination_id = mb.id AND mb.user_id = e.user_id
            LEFT JOIN matrix_accounts mx ON e.destination_type = 'matrix' AND e.destination_id = mx.id AND mx.user_id = e.user_id
            LEFT JOIN discord_accounts dc ON e.destination_type = 'discord' AND e.destination_id = dc.id AND dc.user_id = e.user_id
            LEFT JOIN telegram_accounts tg ON e.destination_type = 'telegram' AND e.destination_id = tg.id AND tg.user_id = e.user_id
            LEFT JOIN webhook_accounts wh ON e.destination_type = 'webhook' AND e.destination_id = wh.id AND wh.user_id = e.user_id
            LEFT JOIN github_accounts gh ON e.destination_type = 'github' AND e.destination_id = gh.id AND gh.user_id = e.user_id
            WHERE e.deleted_at IS NULL AND e.user_id = ?
            ORDER BY e.created_at DESC
        """, (uid,)).fetchall()
        recent_posts = db.execute("""
            SELECT pi.*, f.name as feed_name, e.destination_type as destination_type,
                   CASE
                     WHEN e.destination_type = 'mastodon' THEN '@' || a.username || '@' || REPLACE(a.instance, 'https://', '')
                     WHEN e.destination_type = 'email' THEN ea.name || ' (' || ea.email || ')'
                     WHEN e.destination_type = 'bluesky' THEN '@' || b.handle
                     WHEN e.destination_type = 'microblog' THEN mb.name
                     WHEN e.destination_type = 'matrix' THEN mx.name
                     WHEN e.destination_type = 'discord' THEN dc.name
                     WHEN e.destination_type = 'telegram' THEN tg.name
                     WHEN e.destination_type = 'webhook' THEN wh.name
                     WHEN e.destination_type = 'github' THEN gh.name
                   END as account_name,
                   CASE
                     WHEN e.destination_type = 'mastodon' THEN a.instance
                     WHEN e.destination_type = 'email' THEN ea.email
                     WHEN e.destination_type = 'bluesky' THEN b.pds
                     WHEN e.destination_type = 'microblog' THEN mb.uid
                     WHEN e.destination_type = 'matrix' THEN mx.homeserver
                     WHEN e.destination_type = 'discord' THEN dc.channel_id
                     WHEN e.destination_type = 'telegram' THEN tg.chat_id
                     WHEN e.destination_type = 'webhook' THEN wh.url
                     WHEN e.destination_type = 'github' THEN gh.repo
                   END as instance
            FROM posted_items pi
            JOIN echoes e ON pi.echo_id = e.id
            JOIN feeds f ON e.feed_id = f.id
            LEFT JOIN accounts a ON e.destination_type = 'mastodon' AND e.destination_id = a.id AND a.user_id = e.user_id
            LEFT JOIN email_accounts ea ON e.destination_type = 'email' AND e.destination_id = ea.id AND ea.user_id = e.user_id
            LEFT JOIN bluesky_accounts b ON e.destination_type = 'bluesky' AND e.destination_id = b.id AND b.user_id = e.user_id
            LEFT JOIN microblog_accounts mb ON e.destination_type = 'microblog' AND e.destination_id = mb.id AND mb.user_id = e.user_id
            LEFT JOIN matrix_accounts mx ON e.destination_type = 'matrix' AND e.destination_id = mx.id AND mx.user_id = e.user_id
            LEFT JOIN discord_accounts dc ON e.destination_type = 'discord' AND e.destination_id = dc.id AND dc.user_id = e.user_id
            LEFT JOIN telegram_accounts tg ON e.destination_type = 'telegram' AND e.destination_id = tg.id AND tg.user_id = e.user_id
            LEFT JOIN webhook_accounts wh ON e.destination_type = 'webhook' AND e.destination_id = wh.id AND wh.user_id = e.user_id
            LEFT JOIN github_accounts gh ON e.destination_type = 'github' AND e.destination_id = gh.id AND gh.user_id = e.user_id
            WHERE e.user_id = ?
            ORDER BY pi.posted_at DESC
            LIMIT 20
        """, (uid,)).fetchall()
        # Webhook URLs can carry a secret query token: mask before render.
        recent_posts = _mask_webhook_urls(recent_posts)
        stats = {
            "accounts": (
                mastodon_accounts
                + email_accounts
                + bluesky_accounts
                + microblog_accounts
                + matrix_accounts
                + discord_accounts
                + telegram_accounts
                + webhook_accounts
                + github_accounts
            ),
            "feeds": len(feeds),
            "echoes": len(echoes),
            "active_echoes": sum(1 for e in echoes if e["enabled"]),
            "total_posts": db.execute(
                "SELECT COUNT(*) AS n FROM posted_items pi JOIN echoes e ON pi.echo_id = e.id"
                " WHERE pi.status = 'success' AND e.user_id = ?",
                (uid,),
            ).fetchone()["n"],
            "failed_posts": db.execute(
                "SELECT COUNT(*) AS n FROM posted_items pi JOIN echoes e ON pi.echo_id = e.id"
                " WHERE pi.status IN ('failed', 'gave_up') AND e.user_id = ?",
                (uid,),
            ).fetchone()["n"],
        }
    return render("dashboard.html", request, feeds=feeds, echoes=echoes,
                  recent_posts=recent_posts, stats=stats)


# ── Email verification ────────────────────────────────────────────────────────

@app.get("/verify-email", response_class=HTMLResponse)
async def verify_email(token: str = "", request: Request = None):
    from auth import _require_multi
    from verification import consume_token

    _require_multi()
    uid = consume_token(token, "verify") if token else None
    if uid is None:
        return render(
            "error.html", request, status_code=400, code=400,
            message="This verification link is invalid or has expired.",
        )
    with get_db() as db:
        db.execute(
            "UPDATE users SET email_verified = 1 WHERE id = ?", (uid,)
        )
    logger.info("User %s verified their email", uid)
    return RedirectResponse(url="/?verified=1", status_code=302)


@app.post("/resend-verification")
def resend_verification(request: Request):
    from auth import _require_multi
    from verification import issue_token, resend_allowed

    _require_multi()
    uid = current_user_id(request)
    with get_db() as db:
        row = db.execute(
            "SELECT email, email_verified FROM users WHERE id = ?", (uid,)
        ).fetchone()
    if not row or row["email_verified"]:
        return RedirectResponse(url="/", status_code=302)
    if not resend_allowed(uid, "verify"):
        return render(
            "error.html", request, status_code=400, code=400,
            message="Too many verification emails. Wait a while before trying again.",
        )
    try:
        from email_sender import send_system_email

        token = issue_token(uid, "verify")
        link = auth._absolute_link(f"/verify-email?token={token}")
        if not link:
            return render(
                "error.html", request, status_code=500, code=500,
                message="This deployment cannot send email links yet "
                        "(FEEDECHO_BASE_URL is not configured).",
            )
        send_system_email(
            row["email"],
            "Verify your FeedEcho account",
            f"Verify your email address by opening this link:\n\n{link}\n\n"
            "This link expires in 24 hours.",
        )
    except Exception as exc:  # noqa: BLE001 — resend failure surfaces in the banner
        logger.warning("Resend verification email for user %s failed: %s", uid, exc)
        return render(
            "error.html", request, status_code=400, code=400,
            message="System email is not configured yet. Try again later.",
        )
    return RedirectResponse(url="/?verification_sent=1", status_code=302)


# ── Admin ─────────────────────────────────────────────────────────────────────

def _admin_uid_or_none(request: Request) -> int | None:
    """The admin's user id, or None when the caller is not an admin.

    404 in single mode (admin routes don't exist there); None in multi
    mode when the authenticated user lacks the role.
    """
    from auth import _require_multi

    _require_multi()
    uid = current_user_id(request)
    return uid if auth.is_admin(uid) else None


def _require_admin(request: Request) -> int:
    """The admin's user id or HTTP 403, rendered by the 403 exception
    handler (HTML for browsers, JSON for API clients). Replaces all 11
    guard copies (audit finding 3.4), including GET /admin.
    """
    uid = _admin_uid_or_none(request)
    if uid is None:
        raise HTTPException(status_code=403, detail="Admin access required")
    return uid


def _get_user_or_404(db, user_id: int, columns: str = "id"):
    """One users row or HTTP 404. Replaces the 6 lookup copies (3.5)."""
    row = db.execute(f"SELECT {columns} FROM users WHERE id = ?", (user_id,)).fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="User not found")
    return row


def _error(request: Request, status: int, message: str):
    """render(error.html, ...) with the repeated status/code kwargs (3.6).

    Not yet wired into the ~30 remaining render("error.html") call sites —
    kept for the follow-up pass that migrates them.
    """
    return render("error.html", request, status_code=status, code=status, message=message)


def _upsert_settings(db, uid: int, values: dict[str, str]) -> None:
    """Batch upsert into settings, replacing the 3-copy loop (3.7)."""
    for key, value in values.items():
        db.execute(
            "INSERT INTO settings (user_id, key, value) VALUES (?, ?, ?)"
            " ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value",
            (uid, key, value),
        )


def _admin_usage(db) -> list:
    """Per-tenant usage for the admin page's support view.

    One row per user: resource counts, delivery outcomes in the last 24h
    and 7d, and the most recent delivery error. Set-based (one query for
    all tenants) so the admin page stays one round trip.

    Trial-paused/suspended owners show explicitly — the #1 "nothing posts"
    report is a paused trial, and the admin row should say so at a glance.
    """
    day_ago = (datetime.now(timezone.utc) - timedelta(days=1)).strftime(
        "%Y-%m-%d %H:%M:%S"
    )
    week_ago = (datetime.now(timezone.utc) - timedelta(days=7)).strftime(
        "%Y-%m-%d %H:%M:%S"
    )
    # Time bounds are Python-computed UTC strings bound as parameters:
    # datetime('now', ...) is sqlite-only, and CURRENT_TIMESTAMP on Postgres
    # resolves in the session time zone (see database.as_utc_naive invariant).
    rows = db.execute("""
        SELECT
            u.id AS user_id,
            (SELECT COUNT(*) FROM feeds f
              WHERE f.user_id = u.id AND f.deleted_at IS NULL) AS feeds,
            (SELECT COUNT(*) FROM feeds f
              WHERE f.user_id = u.id AND f.deleted_at IS NULL AND f.paused = 1) AS feeds_paused,
            (SELECT COUNT(*) FROM echoes e
              WHERE e.user_id = u.id AND e.deleted_at IS NULL) AS echoes,
            (SELECT COUNT(*) FROM echoes e
              WHERE e.user_id = u.id AND e.deleted_at IS NULL AND e.enabled = 1) AS echoes_enabled,
            (SELECT COUNT(*) FROM accounts a
              WHERE a.user_id = u.id) +
            (SELECT COUNT(*) FROM email_accounts ea
              WHERE ea.user_id = u.id) +
            (SELECT COUNT(*) FROM bluesky_accounts b
              WHERE b.user_id = u.id) +
            (SELECT COUNT(*) FROM microblog_accounts m
              WHERE m.user_id = u.id) +
            (SELECT COUNT(*) FROM matrix_accounts mx
              WHERE mx.user_id = u.id) +
            (SELECT COUNT(*) FROM discord_accounts dc
              WHERE dc.user_id = u.id) +
            (SELECT COUNT(*) FROM telegram_accounts tg
              WHERE tg.user_id = u.id) +
            (SELECT COUNT(*) FROM webhook_accounts wh
              WHERE wh.user_id = u.id) +
            (SELECT COUNT(*) FROM github_accounts gh
              WHERE gh.user_id = u.id) AS destinations,
            (SELECT COUNT(*) FROM posted_items pi
              JOIN echoes e2 ON pi.echo_id = e2.id
              WHERE e2.user_id = u.id AND pi.status = 'success'
                AND pi.posted_at >= ?) AS posts_24h,
            (SELECT COUNT(*) FROM posted_items pi
              JOIN echoes e2 ON pi.echo_id = e2.id
              WHERE e2.user_id = u.id AND pi.status = 'success'
                AND pi.posted_at >= ?) AS posts_7d,
            (SELECT COUNT(*) FROM posted_items pi
              JOIN echoes e2 ON pi.echo_id = e2.id
              WHERE e2.user_id = u.id AND pi.status IN ('failed', 'gave_up')
                AND pi.posted_at >= ?) AS failures_7d,
            (SELECT COUNT(*) FROM posted_items pi
              JOIN echoes e2 ON pi.echo_id = e2.id
              WHERE e2.user_id = u.id AND pi.status = 'queued') AS queued_now,
            (SELECT pi.error_message FROM posted_items pi
              JOIN echoes e2 ON pi.echo_id = e2.id
              WHERE e2.user_id = u.id AND pi.status IN ('failed', 'gave_up')
              ORDER BY pi.posted_at DESC LIMIT 1) AS last_error
        FROM users u
        WHERE u.email != 'local'
    """, (day_ago, week_ago, week_ago)).fetchall()
    return rows


def _admin_stats(db) -> dict:
    rows = db.execute(
        "SELECT COUNT(*) AS n,"
        " SUM(CASE WHEN is_admin = 1 THEN 1 ELSE 0 END) AS admins,"
        " SUM(CASE WHEN suspended = 1 THEN 1 ELSE 0 END) AS suspended,"
        " SUM(CASE WHEN email_verified = 1 THEN 1 ELSE 0 END) AS verified,"
        " SUM(CASE WHEN plan = 'trial' AND trial_ends_at > CURRENT_TIMESTAMP"
        " THEN 1 ELSE 0 END) AS active_trials"
        " FROM users WHERE email != 'local'"
    ).fetchone()
    return {k: (rows[k] or 0) for k in ("n", "admins", "suspended", "verified", "active_trials")}


def _admin_lock(db) -> None:
    """Serialize last-admin guard+write across concurrent requests.

    On Postgres, pg_advisory_xact_lock is transaction-scoped (released at
    commit/rollback) and re-entrant within a transaction, so calling it from
    the guard and again nowhere else is safe; two concurrent admin actions
    serialize here, closing the count-then-act race between guard and UPDATE.
    SQLite has no such primitive; it is single-writer (busy_timeout 30s) and
    single mode has one tenant, so a no-op is correct there.
    """
    if dialect() == "postgres":
        db.execute("SELECT pg_advisory_xact_lock(824213001)")


def _admin_guard_last_admin(db, user_id: int, column: str) -> str | None:
    """Error message if an action would leave zero admins, else None.

    column is 'suspended' (guard: last ACTIVE admin), 'is_admin'
    (guard: last admin bit), or 'delete' (guard: deleting an admin).
    Single-connection check + write keeps this race-free per request
    transaction; on Postgres the advisory lock in _admin_lock also
    serializes concurrent requests against each other.
    """
    _admin_lock(db)
    if column in ("is_admin", "delete"):
        # Demoting or deleting: preserve at least one admin bit, whatever its state.
        count = db.execute(
            "SELECT COUNT(*) AS n FROM users WHERE is_admin = 1"
        ).fetchone()["n"]
        if (count or 0) <= 1:
            return "Cannot delete the last admin account" if column == "delete" else "Cannot demote the last admin account"
    else:
        # Suspending an admin: keep at least one ACTIVE admin.
        row = db.execute(
            "SELECT is_admin, suspended FROM users WHERE id = ?", (user_id,)
        ).fetchone()
        if row and row["is_admin"] and not row["suspended"]:
            active = db.execute(
                "SELECT COUNT(*) AS n FROM users"
                " WHERE is_admin = 1 AND suspended = 0"
            ).fetchone()["n"]
            if (active or 0) <= 1:
                return "Cannot suspend the last active admin account"
    return None


_VERIFIED_FILTERS = ("yes", "no")


def _admin_filter_params(plan, verified):
    """Normalize the admin user-list filter inputs (None-safe).

    Unknown or invalid values fall back to 'no filter' instead of erroring:
    a stale or hand-edited URL (or a stray multipart field) must never 500
    the admin page. Plan keys are matched EXACTLY — custom
    FEEDECHO_PLAN_LIMITS keys may be case-sensitive, and the per-row plan
    dropdown posts the raw key — while `verified` accepts only yes/no.
    """
    p = plan.strip() if isinstance(plan, str) else ""
    v = verified.strip().lower() if isinstance(verified, str) else ""
    plan_f = p if p in settings.PLAN_LIMITS else ""
    ver_f = v if v in _VERIFIED_FILTERS else ""
    return plan_f, ver_f


def _admin_filter_qs(plan_f, ver_f):
    """Query string ('?plan=trial&verified=no') carrying the active filters.

    urlencode handles plan keys with reserved characters (custom
    FEEDECHO_PLAN_LIMITS names are arbitrary strings).
    """
    params = {}
    if plan_f:
        params["plan"] = plan_f
    if ver_f:
        params["verified"] = ver_f
    return ("?" + urlencode(params)) if params else ""


async def _admin_filter_qs_from_form(request):
    """Filter query string to re-apply after an admin action POST.

    The per-row action forms carry the current filters as hidden inputs so
    a suspend/delete inside a filtered view lands back on that view.
    """
    try:
        form = await request.form()
    except Exception:  # noqa: BLE001 — a broken body must never 500 the redirect
        form = {}
    plan_f, ver_f = _admin_filter_params(
        form.get("filter_plan"), form.get("filter_verified")
    )
    return _admin_filter_qs(plan_f, ver_f)


@app.get("/admin", response_class=HTMLResponse)
async def admin_page(request: Request):
    uid = _require_admin(request)
    plan_f, ver_f = _admin_filter_params(
        request.query_params.get("plan"), request.query_params.get("verified")
    )
    with get_db() as db:
        sql = """
            SELECT id, email, plan, trial_ends_at, email_verified,
                   suspended, is_admin, created_at
              FROM users
             WHERE email != 'local'
        """
        params = []
        if plan_f:
            sql += " AND plan = ?"
            params.append(plan_f)
        if ver_f:
            sql += " AND email_verified = ?"
            params.append(1 if ver_f == "yes" else 0)
        sql += " ORDER BY created_at DESC"
        users = db.execute(sql, params).fetchall()
        stats = _admin_stats(db)
        usage_rows = _admin_usage(db)
        usage_by_user = {row["user_id"]: row for row in usage_rows}
        smtp_rows = db.execute(
            "SELECT key, value FROM system_settings WHERE key LIKE 'smtp_%'"
        ).fetchall()
        invite_rows = invites.list_codes(db)
    smtp = {row["key"]: row["value"] for row in smtp_rows}
    if "smtp_password" in smtp and smtp["smtp_password"]:
        smtp["smtp_password"] = "\u2022\u2022\u2022\u2022\u2022\u2022 (stored)"
    else:
        smtp["smtp_password"] = ""
    smtp["configured"] = bool(smtp.get("smtp_host") and smtp.get("smtp_port"))
    return render("admin.html", request, users=users, stats=stats, smtp=smtp,
                  plan_names=sorted(settings.PLAN_LIMITS.keys()),
                  invite_codes=invite_rows,
                  invites_required=settings.INVITES_REQUIRED,
                  usage_by_user=usage_by_user,
                  filter_plan=plan_f, filter_verified=ver_f)


_SMTP_FORM_KEYS = (
    "smtp_host", "smtp_port", "smtp_username", "smtp_password",
    "smtp_from_email", "smtp_from_name", "smtp_use_tls",
)


@app.post("/admin/email")
async def admin_email_save(request: Request):
    uid = _require_admin(request)
    form = await request.form()

    # Validate before storing anything: a bad port or control characters
    # in header fields must fail here, not at send time.
    port_raw = (form.get("smtp_port") or "").strip()
    try:
        port = int(port_raw)
        if not 1 <= port <= 65535:
            raise ValueError
    except (ValueError, TypeError):
        return render("error.html", request, status_code=400,
                      code=400, message="SMTP port must be a number between 1 and 65535")

    from_email = (form.get("smtp_from_email") or "").strip()
    # Bare-host sender addresses (feedecho@localhost) are legitimate for a
    # local relay, matching the per-tenant SMTP route's rule; both shapes
    # still reject whitespace and control characters.
    if from_email and not is_valid_email(from_email, require_domain_dot=False):
        return render("error.html", request, status_code=400,
                      code=400, message="From address is not a valid email address")

    updates: dict[str, str] = {}
    for key in _SMTP_FORM_KEYS:
        if key == "smtp_password":
            pw = form.get(key) or ""
            # No strip: SMTP passwords may legitimately contain leading or
            # trailing whitespace. Blank (or the masked sentinel, which the
            # page never actually renders into an input) keeps the stored
            # value.
            if pw and pw != "\u2022\u2022\u2022\u2022\u2022\u2022 (stored)":
                updates[key] = pw
            continue
        value = (form.get(key) or "").strip()
        if key in ("smtp_host", "smtp_username", "smtp_from_name") and re.search(
            r"[\r\n]", value
        ):
            return render("error.html", request, status_code=400,
                          code=400, message=f"Invalid characters in {key}")
        updates[key] = value
    updates["smtp_use_tls"] = "1" if form.get("smtp_use_tls") else "0"
    with get_db() as db:
        for key, value in updates.items():
            db.execute(
                "INSERT INTO system_settings (key, value) VALUES (?, ?)"
                " ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                (key, value),
            )
    logger.info("Admin %s updated system email settings", uid)
    return RedirectResponse(url="/admin#email", status_code=302)


@app.post("/admin/email/test")
def admin_email_test(request: Request):
    uid = _require_admin(request)
    with get_db() as db:
        row = db.execute(
            "SELECT email FROM users WHERE id = ?", (uid,)
        ).fetchone()
    target = row["email"] if row else ""
    from email_sender import test_system_smtp_connection

    ok, message = test_system_smtp_connection(to_email=target)
    logger.info("Admin %s system email test: %s (%s)", uid, "ok" if ok else "failed", message)
    return render(
        "admin_email_result.html", request,
        ok=ok, message=message, target=target,
    )


@app.post("/admin/users/{user_id}/suspend")
async def admin_suspend(user_id: int, request: Request):
    """Set suspended=1 (atomic target state, idempotent)."""
    uid = _require_admin(request)
    if user_id == uid:
        return render("error.html", request, status_code=400,
                      code=400, message="You cannot suspend your own account")
    with get_db() as db:
        row = _get_user_or_404(db, user_id)
        guard = _admin_guard_last_admin(db, user_id, "suspended")
        if guard:
            return render("error.html", request, status_code=400,
                          code=400, message=guard)
        db.execute(
            "UPDATE users SET suspended = 1 WHERE id = ? AND suspended = 0",
            (user_id,),
        )
    logger.info("Admin %s suspended user %s", uid, user_id)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/unsuspend")
async def admin_unsuspend(user_id: int, request: Request):
    uid = _require_admin(request)
    with get_db() as db:
        row = _get_user_or_404(db, user_id)
        db.execute(
            "UPDATE users SET suspended = 0 WHERE id = ? AND suspended = 1",
            (user_id,),
        )
    logger.info("Admin %s unsuspended user %s", uid, user_id)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/promote")
async def admin_promote(user_id: int, request: Request):
    uid = _require_admin(request)
    with get_db() as db:
        row = _get_user_or_404(db, user_id)
        db.execute(
            "UPDATE users SET is_admin = 1 WHERE id = ? AND is_admin = 0",
            (user_id,),
        )
    logger.info("Admin %s promoted user %s", uid, user_id)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/demote")
async def admin_demote(user_id: int, request: Request):
    uid = _require_admin(request)
    if user_id == uid:
        return render("error.html", request, status_code=400,
                      code=400, message="You cannot demote your own account")
    with get_db() as db:
        row = _get_user_or_404(db, user_id)
        guard = _admin_guard_last_admin(db, user_id, "is_admin")
        if guard:
            return render("error.html", request, status_code=400,
                          code=400, message=guard)
        db.execute(
            "UPDATE users SET is_admin = 0 WHERE id = ? AND is_admin = 1",
            (user_id,),
        )
    logger.info("Admin %s demoted user %s", uid, user_id)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/plan")
async def admin_set_plan(user_id: int, request: Request):
    """Move a user between plans (trial / beta / paid).

    Moving off 'trial' un-pauses posting immediately (the scheduler gate
    reads plan + trial_ends_at per tick). Moving TO 'trial' does not change
    trial_ends_at: a user whose expiry is already in the past is paused the
    moment the plan flips to trial — that is why extend-trial refuses
    non-trial users and why this route is a deliberate admin action.
    """
    uid = _require_admin(request)
    form = await request.form()
    plan = (form.get("plan") or "").strip()
    if plan not in settings.PLAN_LIMITS:
        return render("error.html", request, status_code=400,
                      code=400, message=f"Unknown plan: {plan!r}")
    with get_db() as db:
        row = _get_user_or_404(db, user_id)
        db.execute("UPDATE users SET plan = ? WHERE id = ?", (plan, user_id))
    logger.info("Admin %s set user %s plan to %s", uid, user_id, plan)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/extend-trial")
async def admin_extend_trial(user_id: int, request: Request):
    """Extend (or restart) a user's trial by N days from now.

    Sets plan back to 'trial' and trial_ends_at to now + days, so an expired
    trial resumes posting the moment the operator grants more time.
    """
    uid = _require_admin(request)
    form = await request.form()
    try:
        days = int(form.get("days") or 0)
    except ValueError:
        days = 0
    if not 1 <= days <= 365:
        return render("error.html", request, status_code=400,
                      code=400, message="Trial extension must be 1-365 days")
    with get_db() as db:
        row = _get_user_or_404(db, user_id, columns="id, plan")
        if row["plan"] != "trial":
            # Extending a paid/beta user's "trial" would silently downgrade
            # them to trial limits and pause them when it lapses. The plan
            # dropdown is the tool for that; this button only touches trials.
            return render("error.html", request, status_code=400,
                          code=400,
                          message=f"{row['plan']} plans have no trial to extend — set the plan instead")
        new_end = (
            datetime.now(timezone.utc) + timedelta(days=days)
        ).strftime("%Y-%m-%d %H:%M:%S")
        db.execute(
            "UPDATE users SET trial_ends_at = ? WHERE id = ?",
            (new_end, user_id),
        )
    logger.info("Admin %s extended user %s trial by %d days (to %s)", uid, user_id, days, new_end)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/users/{user_id}/delete")
async def admin_delete_user(user_id: int, request: Request):
    """Hard-delete a user account and every row it owns (admin action).

    The spam-cleanup counterpart to Suspend: suspension stops posting but
    leaves the row, so the same throwaway address can be re-registered and
    the user list keeps growing. This reuses the self-serve account
    deletion machinery (D4):

    - `_hard_delete_user` removes every owned row (feeds, echoes,
      destinations, settings, queued/sent history) plus the user row, with
      child rows scoped through their parent ids.
    - The billing hook (`cancel_subscription_for_account_deletion`) runs
      first via `_account_deletion_hooks` and VETOES (AccountDeletionAbort)
      when Stripe cannot be reached, so a paying customer is never deleted
      with a live, still-charging subscription.
    - Guard: self-delete refusal. That is also the last-admin protection:
      the caller is an admin, so deleting another admin always leaves at
      least one admin (themselves); the final admin can only be removed by
      their own delete, which is refused.
    - Safety: the form must contain the fixed token DELETE (confirm_text),
      because an admin-delete is irreversible. The typed-email pattern was
      tried first (v1.53.0) and is hostile on mobile: autocorrect/autofill
      mangles random spam addresses, so the confirmation never matched. A
      fixed token is keyboard-proof; the account identity is shown in the
      input placeholder and the JS confirm dialog instead.
    """
    uid = _require_admin(request)
    form = await request.form()
    confirm_text = (form.get("confirm_text") or "").strip()
    with get_db() as db:
        row = _get_user_or_404(db, user_id, columns="id, email, is_admin")
        if user_id == uid:
            # The operator's own row: refusing here IS the self-delete refusal
            return render("error.html", request, status_code=400,
                          code=400, message="You cannot delete your own account")
        if row["is_admin"]:
            guard = _admin_guard_last_admin(db, user_id, "delete")
            if guard:
                return render("error.html", request, status_code=400,
                              code=400, message=guard)
        if confirm_text.strip().upper() != "DELETE":
            return render(
                "error.html", request, status_code=400, code=400,
                message=(
                    "Confirmation failed: type DELETE (in capitals) to "
                    "delete this account"
                ),
            )
        email = row["email"]

    for hook in _account_deletion_hooks:
        try:
            hook(user_id)
        except AccountDeletionAbort as exc:
            # Fail closed: never orphan a live subscription (same
            # contract as the self-serve path).
            return render("error.html", request, status_code=400,
                          code=400, message=str(exc))
        except Exception:  # noqa: BLE001 — best-effort hooks must not block
            logger.exception("account-deletion hook failed for user %s", user_id)

    with get_db() as db:
        _hard_delete_user(db, user_id)
    logger.info("Admin %s DELETED user %s (%s) and all their data", uid, user_id, email)
    return RedirectResponse(url="/admin" + await _admin_filter_qs_from_form(request),
                            status_code=302)


@app.post("/admin/invites/generate")
async def admin_generate_invites(request: Request):
    """Mint 1-50 fresh invite codes."""
    uid = _require_admin(request)
    form = await request.form()
    try:
        count = int(form.get("count") or 1)
    except ValueError:
        count = 1
    count = max(1, min(count, 50))
    with get_db() as db:
        codes = invites.create_codes(db, count, uid)
    logger.info("Admin %s generated %d invite codes", uid, len(codes))
    return RedirectResponse(url="/admin?invites_created=" + str(len(codes)), status_code=302)


@app.post("/admin/invites/revoke")
async def admin_revoke_invite(request: Request):
    """Revoke an unused invite code (used codes are historical records)."""
    uid = _require_admin(request)
    form = await request.form()
    code = (form.get("code") or "").strip()
    with get_db() as db:
        ok = invites.revoke(db, code, uid)
        if ok:
            logger.info("Admin %s revoked an invite code", uid)
    return RedirectResponse(url="/admin", status_code=302)


@app.get("/feeds", response_class=HTMLResponse)
async def feeds_page(
    request: Request,
    add: str = "",
    name: str = "",
):
    uid = current_user_id(request)
    with get_db() as db:
        folders = db.execute(
             "SELECT * FROM folders WHERE user_id = ? ORDER BY position, name",
             (uid,),
        ).fetchall()
        feeds = db.execute(
            "SELECT * FROM feeds WHERE deleted_at IS NULL AND user_id = ? ORDER BY name",
            (uid,),
        ).fetchall()
        feed_echoes = {}
        for f in feeds:
            feed_echoes[f["id"]] = db.execute(
                "SELECT COUNT(*) as c FROM echoes WHERE feed_id = ? AND deleted_at IS NULL AND user_id = ?",
                (f["id"], uid),
            ).fetchone()["c"]
    return render(
        "feeds.html",
        request,
        feeds=feeds,
        folders=folders,
        feed_echoes=feed_echoes,
        add_url=add or "",
        add_name=name or "",
    )


@app.get("/queue", response_class=HTMLResponse)
async def queue_page(request: Request):
    """Curation desk queue: human-scheduled posts waiting for release (Phase 3)."""
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        rows = db.execute(
            """
            SELECT q.*, f.name AS feed_name
              FROM queued_posts q
              LEFT JOIN feeds f ON q.feed_id = f.id
             WHERE q.user_id = ? AND q.status IN ('queued', 'sending', 'failed', 'draft')
             ORDER BY q.scheduled_at ASC
            """,
            (uid,),
        ).fetchall()

        # Destination labels map
        dests = _shout_destinations(uid)
        dest_labels = {d["value"]: d["label"] for d in dests}

        # Formatted list
        posts = []
        for r in rows:
            posts.append({
                "id": r["id"],
                "feed_name": r["feed_name"] or "Unknown feed",
                "destination_label": dest_labels.get(f"{r['destination_type']}:{r['destination_id']}", f"{r['destination_type']}:{r['destination_id']}"),
                "destination_type": r["destination_type"],
                "destination_id": r["destination_id"],
                "content": r["content"],
                "scheduled_at": r["scheduled_at"],
                "status": r["status"],
                "attempt_count": r["attempt_count"],
                "error_message": r["error_message"],
            })

    return render("queue.html", request, posts=posts)


@app.post("/api/queue/{post_id}/post-now")
def queue_post_now(request: Request, post_id: int):
    uid = current_user_id(request)
    now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    claim_token = secrets.token_urlsafe(16)
    with get_db() as db:
        _require_reader(db, uid)
        # Atomically transition from queued to sending to prevent races with flush_queue
        claimed = db.execute(
            """
            UPDATE queued_posts
               SET status = 'sending', claim_token = ?, claimed_at = ?
             WHERE id = ? AND user_id = ? AND status IN ('queued', 'draft')
            """,
            (claim_token, now_utc, post_id, uid),
        )
        if claimed.rowcount != 1:
            raise HTTPException(status_code=409, detail="Post is already sending or has been processed")

        row = db.execute(
            """
            SELECT q.*, f.name AS feed_name
              FROM queued_posts q
              LEFT JOIN feeds f ON q.feed_id = f.id
             WHERE q.id = ? AND q.user_id = ?
            """,
            (post_id, uid),
        ).fetchone()

        # Reuse the echo_id stored at enqueue time if available (idempotency)
        if row["echo_id"]:
            echo = db.execute("SELECT * FROM echoes WHERE id = ?", (row["echo_id"],)).fetchone()
            if echo:
                echo_id = echo["id"]
            else:
                echo_id = db.execute(
                    """INSERT INTO echoes (feed_id, destination_type, destination_id, template,
                                           visibility, attach_image, enabled, one_shot, deleted_at, user_id)
                    VALUES (?, ?, ?, '{{ title }}', ?, ?, 0, 1, ?, ?) RETURNING id""",
                    (row["feed_id"], row["destination_type"], row["destination_id"],
                     row["visibility"] or "public", row["attach_image"], now_utc, uid),
                ).fetchone()["id"]
                echo = db.execute("SELECT * FROM echoes WHERE id = ?", (echo_id,)).fetchone()
        else:
            echo_id = db.execute(
                """INSERT INTO echoes (feed_id, destination_type, destination_id, template,
                                       visibility, attach_image, enabled, one_shot, deleted_at, user_id)
                VALUES (?, ?, ?, '{{ title }}', ?, ?, 0, 1, ?, ?) RETURNING id""",
                (row["feed_id"], row["destination_type"], row["destination_id"],
                 row["visibility"] or "public", row["attach_image"], now_utc, uid),
            ).fetchone()["id"]
            echo = db.execute("SELECT * FROM echoes WHERE id = ?", (echo_id,)).fetchone()
            # Persist the echo_id so future operations reuse it
            db.execute("UPDATE queued_posts SET echo_id = ? WHERE id = ?", (echo_id, post_id))

    item = {
        "id": row["item_id"],
        "title": "",
        "link": "",
        "summary": "",
        "content": "",
        "content_text": "",
        "content_link": "",
        "content_html": "",
        "author": "",
        "date": "",
        "image_url": "",
        "image_alt": row["image_alt"] or "",
    }
    if row["feed_item_id"]:
        with get_db() as db:
            fi = db.execute("SELECT * FROM feed_items WHERE id = ?", (row["feed_item_id"],)).fetchone()
            if fi:
                item.update({
                    "title": fi["title"] or "",
                    "link": fi["link"] or "",
                    "summary": fi["summary"] or "",
                    "content": fi["content"] or "",
                    "content_html": (fi["content_html"] if "content_html" in fi.keys() else "") or "",
                    "categories": categories_from_value(fi["categories"]) if "categories" in fi.keys() else {},
                    "image_url": fi["image_url"] or "",
                })

    try:
        ok = process_echo(
            echo,
            item,
            feed_name=row["feed_name"] or "",
            override_content=row["content"],
        )
    except Exception as exc:
        with get_db() as db:
            db.execute(
                """
                UPDATE queued_posts
                   SET status = 'failed', error_message = ?, attempt_count = attempt_count + 1
                 WHERE id = ? AND claim_token = ?
                """,
                (f"Dispatch exception: {exc}", post_id, claim_token),
            )
        return {"success": False, "status": "failed", "error": str(exc)}

    with get_db() as db:
        pi = db.execute(
            "SELECT id, status, post_url, error_message FROM posted_items WHERE echo_id = ? AND item_id = ?",
            (echo_id, row["item_id"]),
        ).fetchone()
        final_status = "sent" if (pi and pi["status"] == "success") else "failed"
        err_msg = pi["error_message"] if pi else "Unknown dispatch error"
        pi_id = pi["id"] if pi else None
        db.execute(
            """
            UPDATE queued_posts
               SET status = ?, posted_item_id = ?, error_message = ?,
                   attempt_count = attempt_count + 1
             WHERE id = ? AND claim_token = ?
            """,
            (final_status, pi_id, err_msg if final_status == "failed" else None, post_id, claim_token),
        )

    is_success = (final_status == "sent")
    return {"success": is_success, "status": final_status, "post_url": pi["post_url"] if pi else None}


@app.post("/api/queue/{post_id}/cancel")
def queue_cancel(request: Request, post_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT id, status, echo_id FROM queued_posts WHERE id = ? AND user_id = ?",
            (post_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Queued post not found")
        # draft rows (auto-queue staging) cancel like queued ones: they are
        # still awaiting their first dispatch attempt.
        if row["status"] not in ("queued", "failed", "draft"):
            raise HTTPException(status_code=400, detail="Cannot cancel post that is already sending or completed")
        # Delete the orphaned one-shot echo and any posted_items row, atomically
        # with the status check (same transaction).
        if row["echo_id"]:
            db.execute("DELETE FROM posted_items WHERE echo_id = ?", (row["echo_id"],))
            db.execute("DELETE FROM echoes WHERE id = ?", (row["echo_id"],))
        db.execute(
            "UPDATE queued_posts SET status = 'cancelled' WHERE id = ? AND user_id = ?",
            (post_id, uid),
        )
    return RedirectResponse(url="/queue", status_code=303)


@app.post("/api/queue/{post_id}/retry")
def queue_retry(request: Request, post_id: int):
    """Reset a failed queued post back to 'queued' for re-dispatch."""
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT id, status, error_message, echo_id FROM queued_posts WHERE id = ? AND user_id = ?",
            (post_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Queued post not found")
        if row["status"] != "failed":
            raise HTTPException(status_code=400, detail="Can only retry failed posts")
        # Clear the stale posted_items row from the failed attempt so _claim_post
        # can create a fresh pending row on the next dispatch attempt.
        if row["echo_id"]:
            db.execute(
                "DELETE FROM posted_items WHERE echo_id = ?",
                (row["echo_id"],),
            )
        db.execute(
            "UPDATE queued_posts SET status = 'queued', error_message = NULL, attempt_count = 0 WHERE id = ? AND user_id = ?",
            (post_id, uid),
        )
    return RedirectResponse(url="/queue", status_code=303)


@app.post("/api/queue/{post_id}/edit")
def queue_edit(
    request: Request,
    post_id: int,
    content: str = Form(...),
    scheduled_at: str = Form(...),
):
    uid = current_user_id(request)
    content = content.strip()
    if not content:
        raise HTTPException(status_code=400, detail="Content cannot be empty")
    scheduled_at = scheduled_at.strip()
    norm_dt = as_utc_naive(scheduled_at)
    if not norm_dt:
        raise HTTPException(status_code=400, detail="Invalid scheduled_at datetime")
    normalized_scheduled_at = norm_dt.strftime("%Y-%m-%d %H:%M:%S")

    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT id, status FROM queued_posts WHERE id = ? AND user_id = ?",
            (post_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Queued post not found")
        if row["status"] not in ("queued", "draft"):
            raise HTTPException(status_code=400, detail="Cannot edit post that is already sending or completed")
        db.execute(
            "UPDATE queued_posts SET content = ?, scheduled_at = ? WHERE id = ? AND user_id = ? AND status IN ('queued', 'draft')",
            (content, normalized_scheduled_at, post_id, uid),
        )
    return RedirectResponse(url="/queue", status_code=303)


@app.post("/api/queue/{post_id}/reorder")
def queue_reorder(request: Request, post_id: int, direction: str = Form(...)):
    """Move a queued item up or down by swapping scheduled_at with adjacent item."""
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        cur = db.execute(
            "SELECT * FROM queued_posts WHERE id = ? AND user_id = ? AND status = 'queued'",
            (post_id, uid),
        ).fetchone()
        if not cur:
            raise HTTPException(status_code=404, detail="Queued post not found")

        cur_dest = f"{cur['destination_type']}:{cur['destination_id']}"
        if direction == "up":
            adj = db.execute(
                """
                SELECT * FROM queued_posts
                 WHERE user_id = ? AND destination_type = ? AND destination_id = ?
                   AND status = 'queued' AND scheduled_at < ?
                 ORDER BY scheduled_at DESC LIMIT 1
                """,
                (uid, cur["destination_type"], cur["destination_id"], cur["scheduled_at"]),
            ).fetchone()
        else:
            adj = db.execute(
                """
                SELECT * FROM queued_posts
                 WHERE user_id = ? AND destination_type = ? AND destination_id = ?
                   AND status = 'queued' AND scheduled_at > ?
                 ORDER BY scheduled_at ASC LIMIT 1
                """,
                (uid, cur["destination_type"], cur["destination_id"], cur["scheduled_at"]),
            ).fetchone()

        if adj:
            t1, t2 = cur["scheduled_at"], adj["scheduled_at"]
            db.execute("UPDATE queued_posts SET scheduled_at = ? WHERE id = ?", (t2, cur["id"]))
            db.execute("UPDATE queued_posts SET scheduled_at = ? WHERE id = ?", (t1, adj["id"]))

    return RedirectResponse(url="/queue", status_code=303)


@app.get("/accounts", response_class=HTMLResponse)
async def accounts_page(request: Request):
    uid = current_user_id(request)
    (
        mastodon_accounts,
        email_accounts,
        bluesky_accounts,
        microblog_accounts,
        matrix_accounts,
        discord_accounts,
        telegram_accounts,
        webhook_accounts,
        github_accounts,
    ) = _get_all_accounts(uid)
    smtp_settings = _get_smtp_settings(mask_password=True, user_id=uid)
    smtp_configured = bool(smtp_settings.get("smtp_host"))
    return render("accounts.html", request,
                  mastodon_accounts=mastodon_accounts,
                  email_accounts=email_accounts,
                  bluesky_accounts=bluesky_accounts,
                  microblog_accounts=microblog_accounts,
                  matrix_accounts=matrix_accounts,
                  discord_accounts=discord_accounts,
                  telegram_accounts=telegram_accounts,
                  webhook_accounts=webhook_accounts,
                  github_accounts=github_accounts,
                  smtp_configured=smtp_configured,
                  smtp_settings=smtp_settings,
                  booster_configured=bool(settings.BOOSTER_URL and settings.BOOSTER_TOKEN))


def _echoes_page_context(db, uid: int) -> dict:
    """Everything the echoes page (and any error re-render of it) needs.

    One builder because /api/echoes/{id}/booster re-renders the page on
    refusal — the page's queries must not drift between the two entry
    points.
    """
    echoes = db.execute("""
        SELECT e.*, f.name as feed_name,
               CASE
                 WHEN e.destination_type = 'mastodon' THEN '@' || a.username || '@' || REPLACE(a.instance, 'https://', '')
                 WHEN e.destination_type = 'email' THEN ea.name || ' (' || ea.email || ')'
                 WHEN e.destination_type = 'bluesky' THEN '@' || b.handle
                 WHEN e.destination_type = 'microblog' THEN mb.name
                 WHEN e.destination_type = 'matrix' THEN mx.name
                 WHEN e.destination_type = 'discord' THEN dc.name
                 WHEN e.destination_type = 'telegram' THEN tg.name
                 WHEN e.destination_type = 'webhook' THEN wh.name
                 WHEN e.destination_type = 'github' THEN gh.name
               END as destination_name,
               a.booster_enabled as dest_booster_enabled
        FROM echoes e
        JOIN feeds f ON e.feed_id = f.id
        LEFT JOIN accounts a ON e.destination_type = 'mastodon' AND e.destination_id = a.id AND a.user_id = e.user_id
        LEFT JOIN email_accounts ea ON e.destination_type = 'email' AND e.destination_id = ea.id AND ea.user_id = e.user_id
        LEFT JOIN bluesky_accounts b ON e.destination_type = 'bluesky' AND e.destination_id = b.id AND b.user_id = e.user_id
        LEFT JOIN microblog_accounts mb ON e.destination_type = 'microblog' AND e.destination_id = mb.id AND mb.user_id = e.user_id
        LEFT JOIN matrix_accounts mx ON e.destination_type = 'matrix' AND e.destination_id = mx.id AND mx.user_id = e.user_id
        LEFT JOIN discord_accounts dc ON e.destination_type = 'discord' AND e.destination_id = dc.id AND dc.user_id = e.user_id
        LEFT JOIN telegram_accounts tg ON e.destination_type = 'telegram' AND e.destination_id = tg.id AND tg.user_id = e.user_id
        LEFT JOIN webhook_accounts wh ON e.destination_type = 'webhook' AND e.destination_id = wh.id AND wh.user_id = e.user_id
        LEFT JOIN github_accounts gh ON e.destination_type = 'github' AND e.destination_id = gh.id AND gh.user_id = e.user_id
        WHERE e.deleted_at IS NULL AND e.user_id = ?
        ORDER BY e.created_at DESC
    """, (uid,)).fetchall()
    feeds = db.execute(
        "SELECT * FROM feeds WHERE deleted_at IS NULL AND user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    mastodon_accounts = db.execute(
        "SELECT id, name, username, instance FROM accounts WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    email_accounts = db.execute(
        "SELECT id, name, email FROM email_accounts WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    bluesky_accounts = db.execute(
        "SELECT id, name, handle FROM bluesky_accounts WHERE user_id = ? ORDER BY handle",
        (uid,),
    ).fetchall()
    microblog_accounts = db.execute(
        "SELECT id, name, uid FROM microblog_accounts WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    matrix_accounts = db.execute(
        "SELECT id, name, room_id, room_alias FROM matrix_accounts"
        " WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    discord_accounts = db.execute(
        "SELECT id, name, channel_id FROM discord_accounts"
        " WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    telegram_accounts = db.execute(
        "SELECT id, name, chat_id FROM telegram_accounts"
        " WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    webhook_accounts = db.execute(
        "SELECT id, name FROM webhook_accounts"
        " WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    github_accounts = db.execute(
        "SELECT id, name, repo, branch FROM github_accounts"
        " WHERE user_id = ? ORDER BY name",
        (uid,),
    ).fetchall()
    return {
        "echoes": echoes,
        "feeds": feeds,
        "mastodon_accounts": mastodon_accounts,
        "email_accounts": email_accounts,
        "bluesky_accounts": bluesky_accounts,
        "microblog_accounts": microblog_accounts,
        "matrix_accounts": matrix_accounts,
        "discord_accounts": discord_accounts,
        "telegram_accounts": telegram_accounts,
        "webhook_accounts": webhook_accounts,
        "github_accounts": github_accounts,
    }


def _render_echoes_error(request: Request, message: str) -> HTMLResponse:
    """Render the echoes page with an error banner (mirrors _render_accounts_error)."""
    uid = current_user_id(request)
    with get_db() as db:
        ctx = _echoes_page_context(db, uid)
    return render(
        "echoes.html",
        request,
        **ctx,
        template_vars=available_variables(),
        preselect_feed_id=0,
        return_to="/echoes",
        booster_configured=bool(settings.BOOSTER_URL and settings.BOOSTER_TOKEN),
        error=message,
    )


@app.get("/echoes", response_class=HTMLResponse)
async def echoes_page(request: Request, feed: str = "", from_: str = "echoes"):
    uid = current_user_id(request)
    with get_db() as db:
        ctx = _echoes_page_context(db, uid)
    preselect_feed_id = _filter_int(feed)
    return_to = "/reader" if from_ == "reader" else "/echoes"
    return render("echoes.html", request,
                  booster_configured=bool(settings.BOOSTER_URL and settings.BOOSTER_TOKEN),
                  template_vars=available_variables(),
                  preselect_feed_id=preselect_feed_id,
                  return_to=return_to,
                  **ctx)


# Post history: one label per destination type, the instance/handle detail
# behind it, and the joins that resolve both across the five destination
# tables. Held as constants because /history renders them AND its filter
# dropdowns are built from them — inlining the SQL twice let the page and its
# filter options drift apart.
_HISTORY_ACCOUNT_LABEL_SQL = """CASE
                     WHEN e.destination_type = 'mastodon' THEN '@' || a.username || '@' || REPLACE(a.instance, 'https://', '')
                     WHEN e.destination_type = 'email' THEN ea.name
                     WHEN e.destination_type = 'bluesky' THEN '@' || b.handle
                     WHEN e.destination_type = 'microblog' THEN mb.name
                     WHEN e.destination_type = 'matrix' THEN mx.name
                     WHEN e.destination_type = 'discord' THEN dc.name
                     WHEN e.destination_type = 'telegram' THEN tg.name
                     WHEN e.destination_type = 'webhook' THEN wh.name
                     WHEN e.destination_type = 'github' THEN gh.name
                   END"""

_HISTORY_ACCOUNT_INSTANCE_SQL = """CASE
                     WHEN e.destination_type = 'mastodon' THEN a.instance
                     WHEN e.destination_type = 'email' THEN ea.email
                     WHEN e.destination_type = 'bluesky' THEN b.pds
                     WHEN e.destination_type = 'microblog' THEN mb.uid
                     WHEN e.destination_type = 'matrix' THEN mx.homeserver
                     WHEN e.destination_type = 'discord' THEN dc.channel_id
                     WHEN e.destination_type = 'telegram' THEN tg.chat_id
                     WHEN e.destination_type = 'webhook' THEN wh.url
                     WHEN e.destination_type = 'github' THEN gh.repo
                   END"""

_HISTORY_DESTINATION_JOINS_SQL = """
            LEFT JOIN accounts a ON e.destination_type = 'mastodon' AND e.destination_id = a.id AND a.user_id = e.user_id
            LEFT JOIN email_accounts ea ON e.destination_type = 'email' AND e.destination_id = ea.id AND ea.user_id = e.user_id
            LEFT JOIN bluesky_accounts b ON e.destination_type = 'bluesky' AND e.destination_id = b.id AND b.user_id = e.user_id
            LEFT JOIN microblog_accounts mb ON e.destination_type = 'microblog' AND e.destination_id = mb.id AND mb.user_id = e.user_id
            LEFT JOIN matrix_accounts mx ON e.destination_type = 'matrix' AND e.destination_id = mx.id AND mx.user_id = e.user_id
            LEFT JOIN discord_accounts dc ON e.destination_type = 'discord' AND e.destination_id = dc.id AND dc.user_id = e.user_id
            LEFT JOIN telegram_accounts tg ON e.destination_type = 'telegram' AND e.destination_id = tg.id AND tg.user_id = e.user_id
            LEFT JOIN webhook_accounts wh ON e.destination_type = 'webhook' AND e.destination_id = wh.id AND wh.user_id = e.user_id
            LEFT JOIN github_accounts gh ON e.destination_type = 'github' AND e.destination_id = gh.id AND gh.user_id = e.user_id"""


def _filter_int(value: str) -> int | None:
    """A query-string integer, or None when the value is not one.

    ``str.isdigit()`` is not a safe pre-check: it is True for characters
    ``int()`` rejects ("²", "①"), so gating on it 500s the page on
    ``?feed=²`` — exactly the malformed input the filters promise to ignore.
    """
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _parse_destination_filter(value: str) -> tuple[str | None, int | None]:
    """Parse a ``"<type>:<id>"`` destination filter from the query string.

    Anything malformed — unknown type, non-numeric id, empty value — is
    dropped and the page renders unfiltered rather than 400ing: a stale
    bookmark or a destination that has since been removed should still show
    history, not an error.
    """
    dest_type, _, raw_id = (value or "").partition(":")
    dest_id = _filter_int(raw_id)
    if dest_type in DESTINATION_TABLES_BY_TYPE and dest_id is not None:
        return dest_type, dest_id
    return None, None


@app.get("/history", response_class=HTMLResponse)
async def history_page(request: Request, feed: str = "", account: str = "", item_id: str = ""):
    """Post history, newest first, optionally filtered by feed, destination, and item_id.

    Filters are applied in SQL, before the LIMIT (issue #16).
    """
    uid = current_user_id(request)
    feed_id = _filter_int(feed)
    dest_type, dest_id = _parse_destination_filter(account)
    item_id_filter = item_id.strip() if item_id else ""

    where = ["e.user_id = ?"]
    params: list = [uid]
    if feed_id is not None:
        where.append("e.feed_id = ?")
        params.append(feed_id)
    if dest_type is not None:
        where.append("e.destination_type = ?")
        where.append("e.destination_id = ?")
        params.extend([dest_type, dest_id])
    if item_id_filter:
        where.append("pi.item_id = ?")
        params.append(item_id_filter)

    with get_db() as db:
        posts = db.execute(f"""
            SELECT pi.*, f.name as feed_name, e.destination_type as destination_type,
                   {_HISTORY_ACCOUNT_LABEL_SQL} as account_name,
                   {_HISTORY_ACCOUNT_INSTANCE_SQL} as instance
            FROM posted_items pi
            JOIN echoes e ON pi.echo_id = e.id
            JOIN feeds f ON e.feed_id = f.id
            {_HISTORY_DESTINATION_JOINS_SQL}
            WHERE {" AND ".join(where)}
            ORDER BY pi.posted_at DESC
            LIMIT 100
        """, tuple(params)).fetchall()
        # Webhook URLs can carry a secret query token: mask before render.
        posts = _mask_webhook_urls(posts)
        feed_options = db.execute("""
            SELECT DISTINCT f.id as id, f.name as name
            FROM posted_items pi
            JOIN echoes e ON pi.echo_id = e.id
            JOIN feeds f ON e.feed_id = f.id
            WHERE e.user_id = ?
            ORDER BY f.name
        """, (uid,)).fetchall()
        destination_rows = db.execute(f"""
            SELECT DISTINCT e.destination_type as destination_type,
                   e.destination_id as destination_id,
                   {_HISTORY_ACCOUNT_LABEL_SQL} as account_name
            FROM posted_items pi
            JOIN echoes e ON pi.echo_id = e.id
            {_HISTORY_DESTINATION_JOINS_SQL}
            WHERE e.user_id = ?
            ORDER BY 3
        """, (uid,)).fetchall()
    # A destination that has been disconnected leaves history rows behind with
    # nothing left to join to, so label those by type and id instead of
    # dropping them — otherwise their rows become unfilterable.
    destination_options = [
        {
            "value": f"{row['destination_type']}:{row['destination_id']}",
            "label": row["account_name"]
            or f"{row['destination_type']} #{row['destination_id']} (removed)",
        }
        for row in destination_rows
    ]
    return render(
        "history.html",
        request,
        posts=posts,
        feed_options=feed_options,
        destination_options=destination_options,
        filters={
            "feed": str(feed_id) if feed_id is not None else "",
            "account": f"{dest_type}:{dest_id}" if dest_type is not None else "",
            "item_id": item_id_filter,
        },
        filtered=feed_id is not None or dest_type is not None or bool(item_id_filter),
    )


# ── Reader keyset pagination & search helpers ──────────────────────────────
READER_PAGE_SIZE = 50
_CURSOR_RE = re.compile(r"^[0-9 :.\-]*\|\d+$")


class _SavedSearchCountsCache:
    """Per-user, 60s TTL cache of saved-search item counts.

    Invalidation is keyed by max_item_id (see .get()) rather than pure
    time, so a stale entry never outlives the item that would change it by
    more than the TTL. Every route that mutates saved searches, feeds, or
    read-state must call .invalidate(uid) — wrapped in a class (instead of
    the previous bare module-level dict) so every one of those call
    sites reads as an obvious, greppable method name rather than a
    dict.pop() that looks like ordinary housekeeping and is easy to miss
    in a diff. Design-patterns audit finding 2.8.
    """

    def __init__(self) -> None:
        self._store: dict[int, tuple[int, float, dict[int, int]]] = {}

    def get(self, uid: int, max_item_id: int, now: float, ttl: int = 60) -> dict[int, int] | None:
        cached = self._store.get(uid)
        if cached and cached[0] == max_item_id and (now - cached[1]) < ttl:
            return cached[2]
        return None

    def set(self, uid: int, max_item_id: int, now: float, counts: dict[int, int]) -> None:
        self._store[uid] = (max_item_id, now, counts)

    def invalidate(self, uid: int) -> None:
        self._store.pop(uid, None)


_saved_search_counts_cache = _SavedSearchCountsCache()


def _reader_keyset(cursor: str | None) -> tuple[str, list]:
    """SQL predicate + params selecting rows strictly AFTER `cursor` in reader order.

    Cursor format: "<published_at or ''>|<id>". Two cases, because NULL-dated
    rows sort as one block after all dated rows:
      - cursor row HAS a date: dated rows older than it, or same date + lower id,
        plus every NULL-dated row.
      - cursor row has NO date: only NULL-dated rows with a lower id.
    """
    if not cursor or not _CURSOR_RE.match(cursor):
        return "", []
    raw_pub, sep, raw_id = cursor.partition("|")
    try:
        item_id = int(raw_id)
        if not (0 < item_id <= 9223372036854775807):
            return "", []
    except ValueError:
        return "", []

    pub = raw_pub.strip()
    if pub:
        # Validate that pub parses as a real datetime before passing to Postgres,
        # otherwise PG raises InvalidDatetimeFormat / DatetimeFieldOverflow (500).
        try:
            datetime.strptime(pub, "%Y-%m-%d %H:%M:%S")
        except ValueError:
            return "", []
        sql = (
            "( i.published_at IS NULL"
            "  OR i.published_at < ?"
            "  OR (i.published_at = ? AND i.id < ?) )"
        )
        return sql, [pub, pub, item_id]
    else:
        sql = "( i.published_at IS NULL AND i.id < ? )"
        return sql, [item_id]


def _parse_reader_query(q: str):
    """Split a reader search query into operator filters and bare text terms.

    Returns ``(filters, terms)``. Operators (lowercased) are:
      is:starred | is:unread | is:read
      feed:NAME   (substring match on the feed name)
      folder:NAME (substring match on the folder name)
      in:title | in:body  (scope the free-text terms)
    Anything without a recognized ``op:value`` shape becomes a bare term.
    """
    filters = []
    terms = []
    for token in q.split():
        op, sep, val = token.partition(":")
        if sep and op.lower() in ("is", "feed", "folder", "in") and val:
            filters.append((op.lower(), val.lower()))
        else:
            terms.append(token.lower())
    return filters, terms


def _escape_like(term: str) -> str:
    """Escape LIKE wildcards so user text matches literally.

    Patterns built from this are used with ``LIKE ... ESCAPE '!'`` on both
    dialects (Postgres defaults to backslash, sqlite has none — so we pick an
    explicit, dialect-neutral escape character).
    """
    return term.replace("!", "!!").replace("%", "!%").replace("_", "!_")


@app.get("/reader", response_class=HTMLResponse)
async def reader_page(
    request: Request,
    feed: str = "",
    folder: str = "",
    view: str = "unread",
    q: str = "",
    fulltext: str = "",
    after: str = "",
    saved: str = "",
):
    """Consolidated RSS reading surface (issue #11).

    Server-rendered list of stored feed items, newest first, scoped to the
    viewer's feeds. ``view`` is one of all|unread|starred|today; ``feed``
    narrows to a single feed; ``folder`` narrows to feeds in a folder;
    ``saved`` loads a saved search query;
    ``fulltext`` renders the full article body inline instead of the summary
    teaser; ``after`` is a keyset cursor ("<published_at>|<id>") for
    pagination. Read/star toggles and the per-feed reading switch live in the
    reader API routes below.
    """
    uid = current_user_id(request)
    feed_id = _filter_int(feed)
    folder_id = _filter_int(folder)
    saved_id = _filter_int(saved)
    view = view if view in ("all", "unread", "starred", "today") else "unread"
    q = (q or "").strip()
    after = (after or "").strip()
    with get_db() as db:
        _require_reader(db, uid)
        saved_searches = db.execute(
            "SELECT * FROM saved_searches WHERE user_id = ? ORDER BY position, name",
            (uid,),
        ).fetchall()

        # If saved search ID provided and no query override, load query from stored search
        active_saved_search = None
        if saved_id is not None:
            for s in saved_searches:
                if s["id"] == saved_id:
                    active_saved_search = s
                    if not q:
                        q = s["query"]
                    break

        folders = db.execute(
            "SELECT * FROM folders WHERE user_id = ? ORDER BY position, name",
            (uid,),
        ).fetchall()

        feeds = db.execute(
            """
            SELECT f.*,
                   (SELECT COUNT(*) FROM feed_items i
                     WHERE i.feed_id = f.id AND i.is_read = 0) AS unread,
                   (SELECT COUNT(*) FROM echoes e
                     WHERE e.feed_id = f.id AND e.deleted_at IS NULL
                       AND e.enabled = 1) AS echo_count
              FROM feeds f
             WHERE f.deleted_at IS NULL AND f.user_id = ? AND f.read_enabled = 1
             ORDER BY f.name
            """,
            (uid,),
        ).fetchall()

        where = ["f.user_id = ?", "f.read_enabled = 1", "f.deleted_at IS NULL"]
        params: list = [uid]
        if feed_id is not None:
            where.append("i.feed_id = ?")
            params.append(feed_id)
        if folder_id is not None:
            where.append("f.folder_id = ?")
            params.append(folder_id)
        if q:
            # Full-text search with operators (issue #11, Tier 1). The
            # unread/starred view filter is skipped so a search surfaces
            # everything cached, read or not. (% and _ act as LIKE wildcards.)
            filters, terms = _parse_reader_query(q)
            text_scope = None  # None = title+body, else "title" or "body"
            for op, val in filters:
                if op == "is":
                    if val == "starred":
                        where.append("i.starred = 1")
                    elif val == "unread":
                        where.append("i.is_read = 0")
                    elif val == "read":
                        where.append("i.is_read = 1")
                elif op == "feed":
                    where.append("LOWER(f.name) LIKE ? ESCAPE '!'")
                    params.append(f"%{_escape_like(val)}%")
                elif op == "folder":
                    where.append("LOWER(fo.name) LIKE ? ESCAPE '!'")
                    params.append(f"%{_escape_like(val)}%")
                elif op == "in":
                    if val in ("title", "body"):
                        text_scope = val
            if terms:
                like = f"%{_escape_like(' '.join(terms))}%"
                if text_scope == "title":
                    where.append("LOWER(i.title) LIKE ? ESCAPE '!'")
                    params.append(like)
                elif text_scope == "body":
                    where.append(
                        "(LOWER(i.content) LIKE ? ESCAPE '!'"
                        " OR LOWER(i.summary) LIKE ? ESCAPE '!')"
                    )
                    params.extend([like, like])
                else:
                    where.append(
                        "(LOWER(i.title) LIKE ? ESCAPE '!'"
                        " OR LOWER(i.content) LIKE ? ESCAPE '!'"
                        " OR LOWER(i.summary) LIKE ? ESCAPE '!')"
                    )
                    params.extend([like, like, like])
        elif view == "unread":
            where.append("i.is_read = 0")
        elif view == "starred":
            where.append("i.starred = 1")
        elif view == "today":
            # Last-24h across all feeds, read or not (a daily entry point).
            cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
            where.append("i.published_at >= ?")
            params.append(cutoff.strftime("%Y-%m-%d %H:%M:%S"))

        if after:
            cursor_sql, cursor_params = _reader_keyset(after)
            if cursor_sql:
                where.append(cursor_sql)
                params.extend(cursor_params)

        # Per-feed keyword mutes (NewsBlur "training-lite"): hide items whose
        # title/body match any muted keyword on their feed, at query time.
        mutes = db.execute(
            "SELECT id, mute_keywords FROM feeds WHERE user_id = ? AND read_enabled = 1"
            " AND mute_keywords IS NOT NULL AND mute_keywords != ''",
            (uid,),
        ).fetchall()
        for mf in mutes:
            for kw in [k.strip() for k in (mf["mute_keywords"] or "").split(",") if k.strip()]:
                like = f"%{_escape_like(kw.lower())}%"
                where.append(
                    "NOT (i.feed_id = ? AND (LOWER(COALESCE(i.title, '')) LIKE ? ESCAPE '!'"
                    " OR LOWER(COALESCE(i.content, '')) LIKE ? ESCAPE '!'"
                    " OR LOWER(COALESCE(i.summary, '')) LIKE ? ESCAPE '!'))"
                )
                params.extend([mf["id"], like, like, like])

        raw_items = db.execute(
            f"""
            SELECT i.*, f.name AS feed_name,
                   (SELECT COUNT(*) FROM posted_items p
                     JOIN echoes e ON p.echo_id = e.id
                    WHERE e.one_shot = 1 AND e.feed_id = i.feed_id
                      AND p.item_id = i.item_id AND p.status = 'success') AS shouted
              FROM feed_items i
              JOIN feeds f ON i.feed_id = f.id
              LEFT JOIN folders fo ON fo.id = f.folder_id
             WHERE {" AND ".join(where)}
             ORDER BY (i.published_at IS NULL), i.published_at DESC, i.id DESC
             LIMIT ?
            """,
            tuple(params) + (READER_PAGE_SIZE + 1,),
        ).fetchall()

        has_more = len(raw_items) > READER_PAGE_SIZE
        raw_page_items = raw_items[:READER_PAGE_SIZE]

        # Delivery badges: compute aggregate status specifically for the items on this page
        # (avoid scanning entire user posted_items history on every reader page load).
        if raw_page_items:
            pairs = [(it["item_id"], it["feed_id"]) for it in raw_page_items]
            pair_clauses = " OR ".join("(p.item_id = ? AND e.feed_id = ?)" for _ in pairs)
            pair_params = [uid]
            for it_id, f_id in pairs:
                pair_params.extend([it_id, f_id])

            delivery_rows = db.execute(
                f"""
                SELECT p.item_id, e.feed_id,
                       SUM(CASE WHEN p.status = 'success'  THEN 1 ELSE 0 END) AS n_sent,
                       SUM(CASE WHEN p.status = 'filtered' THEN 1 ELSE 0 END) AS n_filtered,
                       SUM(CASE WHEN p.status IN ('failed','gave_up') THEN 1 ELSE 0 END) AS n_failed,
                       SUM(CASE WHEN p.status = 'queued'   THEN 1 ELSE 0 END) AS n_queued,
                       MAX(CASE WHEN p.status = 'filtered' THEN p.error_message ELSE NULL END) AS filter_reason
                  FROM posted_items p
                  JOIN echoes e ON e.id = p.echo_id
                 WHERE e.user_id = ? AND e.one_shot = 0 AND ({pair_clauses})
                 GROUP BY p.item_id, e.feed_id
                """,
                tuple(pair_params),
            ).fetchall()

            delivery_map = {(r["item_id"], r["feed_id"]): r for r in delivery_rows}
            items = []
            for it in raw_page_items:
                d = dict(it)
                deliv = delivery_map.get((it["item_id"], it["feed_id"]))
                if deliv:
                    d["n_sent"] = deliv["n_sent"]
                    d["n_filtered"] = deliv["n_filtered"]
                    d["n_failed"] = deliv["n_failed"]
                    d["n_queued"] = deliv["n_queued"]
                    d["filter_reason"] = deliv["filter_reason"]
                else:
                    d["n_sent"] = 0
                    d["n_filtered"] = 0
                    d["n_failed"] = 0
                    d["n_queued"] = 0
                    d["filter_reason"] = None
                items.append(d)
        else:
            items = []
        next_cursor = ""
        if has_more and items:
            last_item = items[-1]
            last_pub = timestamp_str(last_item["published_at"]) if last_item["published_at"] else ""
            next_cursor = f"{last_pub}|{last_item['id']}"

        # Baseline for the "N new" poll: the newest item id across ALL the
        # user's read-enabled feeds (independent of the current view/filter),
        # so the client counts only items that arrive after this page load —
        # not already-read items the current view happens to hide.
        max_row = db.execute(
            "SELECT COALESCE(MAX(i.id), 0) AS m FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE f.user_id = ? AND f.read_enabled = 1 AND f.deleted_at IS NULL",
            (uid,),
        ).fetchone()
        max_item_id = max_row["m"] if max_row else 0

        # Compute counts for saved searches (with in-process 60s cache keyed by (uid, max_item_id))
        saved_search_counts: dict[int, int] = {}
        if saved_searches:
            now_time = time.time()
            cached = _saved_search_counts_cache.get(uid, max_item_id, now_time)
            if cached is not None:
                saved_search_counts = cached
            else:
                # Pre-load mute keywords once for all saved-search count queries
                mutes = db.execute(
                    "SELECT id, mute_keywords FROM feeds WHERE user_id = ? AND read_enabled = 1"
                    " AND mute_keywords IS NOT NULL AND mute_keywords != ''",
                    (uid,),
                ).fetchall()

                for s in saved_searches:
                    s_filters, s_terms = _parse_reader_query(s["query"])
                    # Mirror the /reader route's own read-state handling for a
                    # non-empty query (see the "unread/starred view filter is
                    # skipped" comment above): no implicit unread-only default,
                    # only an explicit is:read/is:unread/is:starred filters the
                    # count. Saved-search queries are never empty (enforced at
                    # creation), so this always matches what opening the saved
                    # search actually shows.
                    s_where = ["f.user_id = ?", "f.read_enabled = 1", "f.deleted_at IS NULL"]
                    s_params: list = [uid]
                    s_text_scope = None
                    for op, val in s_filters:
                        if op == "is":
                            if val == "starred":
                                s_where.append("i.starred = 1")
                            elif val == "unread":
                                s_where.append("i.is_read = 0")
                            elif val == "read":
                                s_where.append("i.is_read = 1")
                        elif op == "feed":
                            s_where.append("LOWER(f.name) LIKE ? ESCAPE '!'")
                            s_params.append(f"%{_escape_like(val)}%")
                        elif op == "folder":
                            s_where.append("LOWER(fo.name) LIKE ? ESCAPE '!'")
                            s_params.append(f"%{_escape_like(val)}%")
                        elif op == "in":
                            if val in ("title", "body"):
                                s_text_scope = val
                    if s_terms:
                        s_like = f"%{_escape_like(' '.join(s_terms))}%"
                        if s_text_scope == "title":
                            s_where.append("LOWER(i.title) LIKE ? ESCAPE '!'")
                            s_params.append(s_like)
                        elif s_text_scope == "body":
                            s_where.append("(LOWER(i.content) LIKE ? ESCAPE '!' OR LOWER(i.summary) LIKE ? ESCAPE '!')")
                            s_params.extend([s_like, s_like])
                        else:
                            s_where.append(
                                "(LOWER(i.title) LIKE ? ESCAPE '!' OR LOWER(i.content) LIKE ? ESCAPE '!'"
                                " OR LOWER(i.summary) LIKE ? ESCAPE '!')"
                            )
                            s_params.extend([s_like, s_like, s_like])

                    # Apply feed mute keywords so saved-search counts match reader visibility
                    for mf in mutes:
                        for kw in [k.strip() for k in (mf["mute_keywords"] or "").split(",") if k.strip()]:
                            like = f"%{_escape_like(kw.lower())}%"
                            s_where.append(
                                "NOT (i.feed_id = ? AND (LOWER(COALESCE(i.title, '')) LIKE ? ESCAPE '!'"
                                " OR LOWER(COALESCE(i.content, '')) LIKE ? ESCAPE '!'"
                                " OR LOWER(COALESCE(i.summary, '')) LIKE ? ESCAPE '!'))"
                            )
                            s_params.extend([mf["id"], like, like, like])

                    cnt_row = db.execute(
                        f"""
                        SELECT COUNT(*) AS c
                          FROM feed_items i
                          JOIN feeds f ON i.feed_id = f.id
                          LEFT JOIN folders fo ON fo.id = f.folder_id
                         WHERE {" AND ".join(s_where)}
                        """,
                        tuple(s_params),
                    ).fetchone()
                    saved_search_counts[s["id"]] = cnt_row["c"] if cnt_row else 0
                _saved_search_counts_cache.set(uid, max_item_id, now_time, saved_search_counts)

    # Progressive enhancement: fetch request with X-Requested-With returns the
    # items fragment alone (for infinite scroll / load more).
    if request.headers.get("X-Requested-With") == "fetch":
        resp = render(
            "reader_items.html",
            request,
            items=items,
            view=view,
            fulltext=bool(fulltext),
            next_cursor=next_cursor,
            shout_destinations=_shout_destinations(uid),
            default_template="{{ title }} {{ link }}",
            template_vars=available_variables(),
        )
        if next_cursor:
            resp.headers["X-Next-Cursor"] = next_cursor
        return resp

    return render(
        "reader.html",
        request,
        feeds=feeds,
        folders=folders,
        items=items,
        next_cursor=next_cursor,
        max_item_id=max_row["m"] if max_row else 0,
        current_feed=str(feed_id) if feed_id is not None else "",
        current_folder=str(folder_id) if folder_id is not None else "",
        saved_searches=saved_searches,
        saved_search_counts=saved_search_counts,
        current_saved_search=str(saved_id) if saved_id is not None else "",
        active_saved_search=active_saved_search,
        view=view,
        q=q,
        fulltext=bool(fulltext),
        shout_destinations=_shout_destinations(uid),
        default_template="{{ title }} {{ link }}",
        template_vars=available_variables(),
    )


@app.get("/howto", response_class=HTMLResponse)
async def howto_page(request: Request):
    return render("howto.html", request)


@app.get("/about", response_class=HTMLResponse)
async def about_page(request: Request):
    from auth import _require_multi

    _require_multi()  # hosted-service disclosure; 404 in self-hosted mode
    return render("about.html", request)


@app.get("/terms", response_class=HTMLResponse)
async def terms_page(request: Request):
    from auth import _require_multi

    # Hosted-service legal page; 404 in self-hosted mode (no contract there).
    _require_multi()
    return render("terms.html", request)


@app.get("/privacy", response_class=HTMLResponse)
async def privacy_page(request: Request):
    from auth import _require_multi

    _require_multi()  # hosted-service legal page; 404 in self-hosted mode
    return render("privacy.html", request)


@app.get("/pricing", response_class=HTMLResponse)
async def pricing_page(request: Request):
    from auth import _require_multi

    _require_multi()  # hosted-service page; 404 in self-hosted mode
    return render("pricing.html", request)


def _render_settings(request: Request, **extra) -> HTMLResponse:
    """Build the settings page context and render it (used by the GET page
    and by the import handler's error/success render)."""
    uid = current_user_id(request)
    smtp_settings = _get_smtp_settings(mask_password=True, user_id=uid)
    smtp_configured = bool(smtp_settings.get("smtp_host"))
    with get_db() as db:
        rows = db.execute(
            """SELECT key, value FROM settings
               WHERE key IN ('retry_max_attempts', 'retry_backoff_minutes',
                             'notify_failure_threshold', 'notify_email')
                 AND user_id = ?""",
            (uid,),
        ).fetchall()
        alt_rows = db.execute(
            """SELECT key, value FROM settings
               WHERE key IN ('alt_text_ai_enabled', 'alt_text_ai_base_url',
                             'alt_text_ai_model', 'alt_text_ai_api_key')
                 AND user_id = ?""",
            (uid,),
        ).fetchall()
    retry_notify = {r["key"]: r["value"] for r in rows}
    alt_text_settings = {r["key"]: r["value"] for r in alt_rows}
    alt_text_configured = bool(alt_text_settings.get("alt_text_ai_base_url"))
    # Mask the API key before sending to the browser
    if alt_text_settings.get("alt_text_ai_api_key"):
        alt_text_settings["alt_text_ai_api_key"] = "********"
    return render("settings.html", request,
                  smtp_settings=smtp_settings,
                  smtp_configured=smtp_configured,
                  retry_notify=retry_notify,
                  alt_text_settings=alt_text_settings,
                  alt_text_configured=alt_text_configured,
                  **extra)


@app.get("/settings", response_class=HTMLResponse)
async def settings_page(request: Request):
    return _render_settings(request)


# ── API: Import / Export ─────────────────────────────────────────────────────

@app.get("/api/export")
def export_all(request: Request):
    """Download the user's feeds, accounts and echoes as a JSON document.

    Contains credentials — auth-gated like every other route; the caller must
    store the file accordingly.
    """
    uid = current_user_id(request)
    with get_db() as db:
        payload = import_export.build_export(db, uid)
    data = json.dumps(payload, indent=2, ensure_ascii=False)
    filename = "feedecho-export-" + datetime.now(timezone.utc).strftime("%Y%m%d") + ".json"
    return Response(
        content=data,
        media_type="application/json",
        headers={"Content-Disposition": f'attachment; filename="{filename}"'},
    )


@app.post("/api/import")
async def import_all(request: Request, file: UploadFile = File(...)):
    """Import feeds, accounts and echoes from an export document.

    Quota overruns (multi mode) and malformed documents render the settings
    page with an error banner; a valid import renders a success banner with the
    counts. Nothing is written on error — the import runs in one transaction.
    """
    uid = current_user_id(request)
    raw = await file.read()
    if len(raw) > 5_000_000:
        return _render_settings(
            request, import_error="Import file is too large (5 MB maximum)."
        )
    try:
        payload = json.loads(raw.decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        return _render_settings(request, import_error="That file is not valid JSON.")
    with get_db() as db:
        try:
            summary = import_export.import_data(db, uid, payload)
        except import_export.ExportError as exc:
            return _render_settings(request, import_error=str(exc))
    parts = [
        f"Imported {summary['added_feeds']} feed"
        + ("s" if summary["added_feeds"] != 1 else ""),
        f"{summary['added_accounts']} account"
        + ("s" if summary["added_accounts"] != 1 else ""),
        f"{summary['added_echoes']} echo"
        + ("s" if summary["added_echoes"] != 1 else ""),
    ]
    if summary["existing_feeds"] or summary["existing_accounts"] or summary["existing_echoes"]:
        parts.append(
            f"{summary['existing_feeds'] + summary['existing_accounts'] + summary['existing_echoes']}"
            " already present (skipped)"
        )
    if summary.get("skipped_accounts"):
        parts.append(
            f"{summary['skipped_accounts']} account"
            + ("s" if summary["skipped_accounts"] != 1 else "")
            + " skipped (no stored credential)"
        )
    if summary["skipped_echoes"]:
        parts.append(
            f"{summary['skipped_echoes']} echo"
            + ("s" if summary["skipped_echoes"] != 1 else "")
            + " skipped (missing feed or account)"
        )
    return _render_settings(request, import_success=", ".join(parts) + ".")


# ── Account deletion (D4) ───────────────────────────────────────────────────

# Overlay-registered hooks run before a tenant is hard-deleted. The OSS repo
# knows nothing about Stripe; the hosted billing module registers a hook that
# cancels the user's subscription so deleting an account never leaves an
# orphaned charge. A hook may raise AccountDeletionAbort to VETO the deletion
# (e.g. it could not cancel a live subscription); any other exception is
# logged and the deletion proceeds (best-effort hooks).
_account_deletion_hooks: list = []


class AccountDeletionAbort(Exception):
    """A pre-deletion hook refused to let the account be deleted."""


def register_account_deletion_hook(fn) -> None:
    _account_deletion_hooks.append(fn)


# Overlay-registered GUARDS that the abandoned-signup sweep consults before it
# deletes a card-pending account (scheduler.cleanup_pending_accounts). Same
# idea as the deletion hooks but for a background job: hosted billing
# registers one that verifies with Stripe that the account has no live
# subscription, because a user whose Checkout webhook never landed still
# LOOKS card-pending locally while paying in Stripe. A guard raises
# AccountDeletionAbort to veto; any OTHER exception also skips the account
# (fail closed — the sweep must never delete on an unverified state).
_pending_cleanup_guards: list = []


def register_pending_cleanup_guard(fn) -> None:
    _pending_cleanup_guards.append(fn)


# Per-user throttle on the deletion password check: a hijacked session (stolen
# cookie) must not get unlimited online guesses at the account password to
# authorize irreversible deletion. Keyed by uid, not IP — the attacker may
# already hold a valid session from a different address. This only delays the
# deletion confirmation; login and every other route are unaffected.
_delete_attempts: dict[int, list[float]] = {}
_delete_lock = threading.Lock()
_MAX_DELETE_ATTEMPTS = 5
_DELETE_WINDOW_SECONDS = 10 * 60


def _delete_throttled(uid: int) -> bool:
    """True when `uid` has exhausted its deletion-confirmation attempts."""
    now = time.time()
    with _delete_lock:
        bucket = [t for t in _delete_attempts.get(uid, []) if now - t < _DELETE_WINDOW_SECONDS]
        if bucket:
            _delete_attempts[uid] = bucket
        else:
            _delete_attempts.pop(uid, None)
        return len(bucket) >= _MAX_DELETE_ATTEMPTS


def _record_delete_failure(uid: int) -> None:
    now = time.time()
    with _delete_lock:
        bucket = [t for t in _delete_attempts.get(uid, []) if now - t < _DELETE_WINDOW_SECONDS]
        bucket.append(now)
        _delete_attempts[uid] = bucket


def _clear_delete_failures(uid: int) -> None:
    with _delete_lock:
        _delete_attempts.pop(uid, None)


def _hard_delete_user(db, uid: int) -> None:
    """Delete every row owned by `uid`, then the user row itself.

    Child rows without a user_id (feed_items, digest_items, drip_items,
    posted_items) are scoped through their parent's id so no tenant data
    survives. Runs inside the caller's transaction. Credential columns are
    encrypted at rest (S3), so dropping the rows destroys the credentials.
    """
    # Rows keyed off the user's feeds (feed_items is the only child of feeds).
    db.execute(
        "DELETE FROM feed_items WHERE feed_id IN (SELECT id FROM feeds WHERE user_id = ?)",
        (uid,),
    )
    # Rows keyed off the user's echoes.
    for child in ("digest_items", "drip_items", "posted_items"):
        db.execute(
            f"DELETE FROM {child} WHERE echo_id IN"
            " (SELECT id FROM echoes WHERE user_id = ?)",
            (uid,),
        )
    # Parent tables carrying user_id directly.
    for table in (
        "echoes",
        "feeds",
        "folders",
        "saved_searches",
        "queued_posts",
        "queue_settings",
        "accounts",
        "email_accounts",
        "bluesky_accounts",
        "microblog_accounts",
        "matrix_accounts",
        "discord_accounts",
        "telegram_accounts",
        "webhook_accounts",
        "github_accounts",
    ):
        db.execute(f"DELETE FROM {table} WHERE user_id = ?", (uid,))
    db.execute("DELETE FROM settings WHERE user_id = ?", (uid,))
    # Auth/flow tables referencing the user id without a user_id column.
    db.execute("DELETE FROM oauth_states WHERE user_id = ?", (uid,))
    db.execute("DELETE FROM email_tokens WHERE user_id = ?", (uid,))
    db.execute("DELETE FROM users WHERE id = ?", (uid,))
    # Clean up the in-process saved-search count cache for this deleted user
    _saved_search_counts_cache.invalidate(uid)


@app.post("/settings/delete-account")
def delete_account_submit(request: Request, password: str = Form("")):
    from auth import _require_multi

    _require_multi()  # self-hosted has no account to delete
    uid = current_user_id(request)
    if _delete_throttled(uid):
        return _render_settings(
            request, delete_error="Too many attempts. Try again later."
        )
    with get_db() as db:
        user = db.execute(
            "SELECT email, password_hash FROM users WHERE id = ?", (uid,)
        ).fetchone()
    if not user or not security.verify_password(password, user["password_hash"]):
        _record_delete_failure(uid)
        return _render_settings(request, delete_error="Incorrect password.")
    _clear_delete_failures(uid)
    email = user["email"]
    for hook in _account_deletion_hooks:
        try:
            hook(uid)
        except AccountDeletionAbort as exc:
            # A hook (e.g. billing) refused to let this account go. Fail closed:
            # do not delete the account and orphan a live subscription.
            return _render_settings(request, delete_error=str(exc))
        except Exception:  # noqa: BLE001 — best-effort hooks must not block
            logger.exception("account-deletion hook failed for user %s", uid)
    with get_db() as db:
        _hard_delete_user(db, uid)
    logger.info("User %s (%s) deleted their account", uid, email)
    response = RedirectResponse(url="/login?deleted=1", status_code=303)
    response.delete_cookie(auth.COOKIE_NAME)
    response.delete_cookie(auth.AUTH_COOKIE_NAME)
    return response


@app.get("/healthz")
async def healthz():
    logger.debug("health check")
    return {"status": "ok"}


# ── API: Mastodon Accounts ──────────────────────────────────────────────────

@app.post("/api/accounts")
async def add_account(
    request: Request,
    name: str = Form(...),
    username: str = Form(""),
    instance: str = Form(...),
    access_token: str = Form(...),
):
    uid = current_user_id(request)
    instance = validate_url(instance)
    with get_db() as db:
        # Cap NEW rows only: re-adding an existing account is an upsert that
        # changes nothing the cap measures.
        existing = db.execute(
            "SELECT id FROM accounts"
            " WHERE user_id = ? AND instance = ? AND username = ?",
            (uid, instance, username or name),
        ).fetchone()
        if existing is None:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        db.execute(
            """
            INSERT INTO accounts (name, username, instance, access_token, user_id)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id, instance, username) DO UPDATE SET
                name = excluded.name,
                access_token = excluded.access_token
            """,
            (name, username or name, instance, security.encrypt_secret(access_token), uid),
        )
    return RedirectResponse(url="/accounts", status_code=303)


@app.post("/api/accounts/{account_id}/test")
def test_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Account not found")
    success, message = test_connection(account["instance"], security.decrypt_secret(account["access_token"]))
    return {"success": success, "message": message}


@app.post("/api/accounts/{account_id}/booster")
async def toggle_booster(request: Request, account_id: int):
    """Toggle FeedBooster for one Mastodon destination: echoed posts to this
    account get announced by @feedbooster@feedecho.net."""
    uid = current_user_id(request)
    if not (settings.BOOSTER_URL and settings.BOOSTER_TOKEN):
        return _render_accounts_error(request, "FeedBooster is not configured on this server.")
    with get_db() as db:
        row = db.execute(
            "SELECT booster_enabled FROM accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Account not found")
        db.execute(
            "UPDATE accounts SET booster_enabled = ? WHERE id = ? AND user_id = ?",
            (0 if row["booster_enabled"] else 1, account_id, uid),
        )
    return RedirectResponse(url="/accounts", status_code=303)


@app.post("/api/accounts/{account_id}/delete")
async def delete_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "mastodon", account_id):
            return _render_accounts_error(
                request,
                "This Mastodon account is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM accounts WHERE id = ? AND user_id = ?", (account_id, uid)
        )
    return RedirectResponse(url="/accounts", status_code=303)


# ── API: Email Accounts ─────────────────────────────────────────────────────

@app.post("/api/email-accounts")
async def add_email_account(
    request: Request,
    name: str = Form(...),
    email: str = Form(...),
):
    uid = current_user_id(request)
    # Recipients become message headers: reject control characters and
    # whitespace here rather than at send time (same rule as the SMTP
    # settings route below).
    email = email.strip()
    if not is_valid_email(email, require_domain_dot=False):
        return _render_accounts_error(
            request, "Enter a valid email address"
        )
    with get_db() as db:
        existing = db.execute(
            "SELECT id FROM email_accounts WHERE user_id = ? AND email = ?",
            (uid, email),
        ).fetchone()
        if existing is None:
            # Cap counts NEW rows only: re-saving an existing address is an
            # upsert that changes nothing the cap measures.
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        db.execute(
            "INSERT INTO email_accounts (name, email, user_id) VALUES (?, ?, ?)"
            " ON CONFLICT(user_id, email) DO UPDATE SET name = excluded.name",
            (name, email, uid),
        )
    return RedirectResponse(url="/accounts?status=email_added", status_code=303)


@app.post("/api/email-accounts/{account_id}/delete")
async def delete_email_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "email", account_id):
            return _render_accounts_error(
                request,
                "This email address is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM email_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts", status_code=303)


# ── API: Bluesky Accounts ───────────────────────────────────────────────────

@app.post("/api/bluesky-accounts")
def add_bluesky_account(
    request: Request,
    name: str = Form(...),
    handle: str = Form(...),
    app_password: str = Form(...),
):
    """Resolve the handle, verify the app password, and store the account.

    Synchronous route (threadpool-offloaded): it performs blocking DNS,
    HTTPS, and SQLite work that must not run on the event loop.
    """
    try:
        handle = bluesky_normalize_handle(handle)
    except ValueError as e:
        return _render_accounts_error(request, str(e))

    display_name = name.strip()[:100] or handle

    try:
        did, pds = bluesky_resolve_pds(handle)
        session = bluesky_create_session(pds, handle, app_password)
    except BlueskyAuthError:
        return _render_accounts_error(
            request,
            "Bluesky rejected the app password. Check the handle and app password and try again.",
        )
    except BlueskyError as e:
        return _render_accounts_error(request, str(e))
    except Exception:
        logger.exception("Bluesky account verification failed for %s", handle)
        return _render_accounts_error(
            request, "Could not verify the Bluesky account. Try again."
        )

    with get_db() as db:
        uid = current_user_id(request)
        try:
            _check_destination_cap(db, uid)
        except PlanError as e:
            return _render_accounts_error(request, str(e))
        db.execute(
            """
            INSERT INTO bluesky_accounts (
                name, handle, app_password, did, pds,
                access_jwt, refresh_jwt, session_expires_at, user_id
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(user_id, handle) DO UPDATE SET
                name = excluded.name,
                app_password = excluded.app_password,
                did = excluded.did,
                pds = excluded.pds,
                access_jwt = excluded.access_jwt,
                refresh_jwt = excluded.refresh_jwt,
                session_expires_at = excluded.session_expires_at
            """,
            (
                display_name,
                handle,
                security.encrypt_secret(app_password),
                session["did"],
                pds,
                security.encrypt_secret(session["access_jwt"]),
                security.encrypt_secret(session["refresh_jwt"]),
                bluesky_session_expiry(session["access_jwt"]),
                uid,
            ),
        )
    return RedirectResponse(url="/accounts?status=bluesky_connected", status_code=303)


@app.post("/api/bluesky-accounts/{account_id}/test")
def test_bluesky_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM bluesky_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Bluesky account not found")
    success, message = test_bluesky_connection(
        account["handle"], security.decrypt_secret(account["app_password"])
    )
    return {"success": success, "message": message}


@app.post("/api/bluesky-accounts/{account_id}/delete")
def delete_bluesky_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "bluesky", account_id):
            return _render_accounts_error(
                request,
                "This Bluesky account is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM bluesky_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=bluesky_deleted", status_code=303)


# ── API: Micro.blog Accounts ────────────────────────────────────────────────

@app.post("/api/microblog-accounts")
def add_microblog_account(
    request: Request,
    token: str = Form(...),
):
    """Verify a Micro.blog app token and store one row per blog it can post to.

    Synchronous route (threadpool-offloaded): it performs blocking HTTPS
    and SQLite work that must not run on the event loop.
    """
    token = token.strip()
    if not token:
        return _render_accounts_error(request, "Enter a Micro.blog app token.")

    try:
        blogs = microblog_list_destinations(token)
    except MicroblogAuthError as e:
        return _render_accounts_error(request, str(e))
    except MicroblogError as e:
        logger.warning("Micro.blog connect failed: %s", e)
        return _render_accounts_error(
            request, "Could not reach micro.blog to verify the token. Try again."
        )
    except Exception:
        logger.exception("Micro.blog account verification failed")
        return _render_accounts_error(
            request, "Could not verify the Micro.blog token. Try again."
        )

    with get_db() as db:
        uid = current_user_id(request)
        # The cap must account for EVERY NEW row this connect will write: a
        # token covering N blogs inserts N rows, not one. Rows the upsert
        # would only update (blogs already connected to this user) don't
        # change what the cap measures, so re-connecting an existing token —
        # e.g. to rotate it — must not be blocked at the cap. Mirrors the
        # email route's "Cap counts NEW rows only" rule.
        existing_uids = {
            row["uid"]
            for row in db.execute(
                "SELECT uid FROM microblog_accounts WHERE user_id = ?", (uid,)
            ).fetchall()
        }
        new_blog_count = sum(1 for b in blogs if b["uid"] not in existing_uids)
        current_total = sum(
            db.execute(
                f"SELECT COUNT(*) AS c FROM {t} WHERE user_id = ?", (uid,)
            ).fetchone()["c"]
            for t in DESTINATION_TABLES
        )
        plan = _user_plan(db, uid)
        cap = plans.limit_for(plan, "max_destinations")
        if cap and new_blog_count and current_total + new_blog_count > cap:
            room = max(0, cap - current_total)
            return _render_accounts_error(
                request,
                f"Your plan allows {cap} connected accounts and this token covers "
                f"{len(blogs)} blog{'s' if len(blogs) != 1 else ''} "
                f"({new_blog_count} new, {room} would fit). "
                "Upgrade or disconnect accounts to connect them all.",
            )
        for blog in blogs:
            db.execute(
                """
                INSERT INTO microblog_accounts (name, uid, token, user_id)
                VALUES (?, ?, ?, ?)
                ON CONFLICT(user_id, uid) DO UPDATE SET
                    name = excluded.name,
                    token = excluded.token
                """,
                (blog["name"], blog["uid"], security.encrypt_secret(token), uid),
            )
    return RedirectResponse(url="/accounts?status=microblog_connected", status_code=303)


@app.post("/api/microblog-accounts/{account_id}/test")
def test_microblog_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM microblog_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Micro.blog account not found")
    success, message = test_microblog_connection(security.decrypt_secret(account["token"]))
    return {"success": success, "message": message}


@app.post("/api/microblog-accounts/{account_id}/delete")
def delete_microblog_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "microblog", account_id):
            return _render_accounts_error(
                request,
                "This micro.blog account is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM microblog_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=microblog_deleted", status_code=303)


# ── API: Matrix Accounts ────────────────────────────────────────────────────

@app.post("/api/matrix-accounts")
def add_matrix_account(
    request: Request,
    homeserver: str = Form(...),
    access_token: str = Form(...),
    room: str = Form(...),
    name: str = Form(""),
):
    """Verify a Matrix token + room and store one account row for that room.

    Synchronous route (threadpool-offloaded): well-known discovery, whoami,
    room resolution, and the membership check are all blocking HTTPS calls.
    """
    access_token = access_token.strip()
    if not access_token:
        return _render_accounts_error(request, "Enter a Matrix access token.")

    try:
        homeserver = matrix_normalize_homeserver(homeserver)
        room_input = matrix_normalize_room(room)
    except ValueError as e:
        return _render_accounts_error(request, str(e))

    try:
        info = matrix_connect(homeserver, access_token, room_input)
    except MatrixAuthError as e:
        return _render_accounts_error(request, str(e))
    except MatrixPermissionError as e:
        return _render_accounts_error(request, str(e))
    except SSRFError as e:
        return _render_accounts_error(request, f"Blocked homeserver URL: {e}")
    except MatrixError as e:
        logger.warning("Matrix connect failed for %s: %s", homeserver, e)
        return _render_accounts_error(request, str(e))
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    except Exception:
        logger.exception("Matrix account verification failed for %s", homeserver)
        return _render_accounts_error(
            request, "Could not verify the Matrix account. Try again."
        )

    # Default display name is what the user recognises the room by: the alias
    # they typed, else the room ID.
    display_name = name.strip()[:100] or info["room_alias"] or info["room_id"]

    with get_db() as db:
        uid = current_user_id(request)
        # Cap counts NEW rows only: reconnecting the same room (to rotate the
        # token) updates in place and must not be blocked at the cap.
        existing = db.execute(
            "SELECT id FROM matrix_accounts"
            " WHERE user_id = ? AND homeserver = ? AND room_id = ?",
            (uid, homeserver, info["room_id"]),
        ).fetchone()
        if not existing:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        db.execute(
            """
            INSERT INTO matrix_accounts (
                name, homeserver, base_url, access_token, matrix_user_id,
                room_id, room_alias, user_id
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(user_id, homeserver, room_id) DO UPDATE SET
                name = excluded.name,
                base_url = excluded.base_url,
                access_token = excluded.access_token,
                matrix_user_id = excluded.matrix_user_id,
                room_alias = excluded.room_alias
            """,
            (
                display_name,
                homeserver,
                info["base_url"],
                security.encrypt_secret(access_token),
                info["user_id"],
                info["room_id"],
                info["room_alias"],
                uid,
            ),
        )
    return RedirectResponse(url="/accounts?status=matrix_connected", status_code=303)


@app.post("/api/matrix-accounts/{account_id}/test")
def test_matrix_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM matrix_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Matrix account not found")
    # Prefer the stored resolved base URL; fall back to the typed homeserver
    # for rows written before discovery ran (or when well-known was down).
    base = account["base_url"] or account["homeserver"]
    success, message = test_matrix_connection(
        base, security.decrypt_secret(account["access_token"]), account["room_id"]
    )
    return {"success": success, "message": message}


@app.post("/api/matrix-accounts/{account_id}/delete")
def delete_matrix_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "matrix", account_id):
            return _render_accounts_error(
                request,
                "This Matrix room is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM matrix_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=matrix_deleted", status_code=303)


# ── API: Discord Accounts ───────────────────────────────────────────────────

@app.post("/api/discord-accounts")
def add_discord_account(
    request: Request,
    webhook_url: str = Form(...),
    name: str = Form(""),
):
    """Verify a Discord webhook URL and store one account row for it.

    Synchronous route (threadpool-offloaded): the verification GET is a
    blocking HTTPS call. Reconnecting the same webhook URL updates the stored
    row (name refresh) instead of duplicating it.
    """
    webhook_url = webhook_url.strip()
    if not webhook_url:
        return _render_accounts_error(request, "Paste the Discord webhook URL.")

    try:
        info = discord_connect(webhook_url)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    except DiscordAuthError as e:
        return _render_accounts_error(request, str(e))
    except DiscordNotFoundError as e:
        return _render_accounts_error(request, str(e))
    except DiscordError as e:
        logger.warning("Discord connect failed: %s", e)
        return _render_accounts_error(request, str(e))
    except Exception:
        logger.exception("Discord account verification failed")
        return _render_accounts_error(
            request, "Could not verify the Discord webhook. Try again."
        )

    # Default display name is the webhook's own name on Discord; the optional
    # user-supplied name wins when given.
    display_name = name.strip()[:100] or info["name"] or "Discord webhook"

    with get_db() as db:
        uid = current_user_id(request)
        # The webhook URL is encrypted at rest (non-deterministic Fernet), so
        # dedup uses a deterministic digest of the plaintext URL; the unique
        # index on (user_id, webhook_url_hash) makes the upsert atomic.
        webhook_url_hash = security.hash_secret(info["webhook_url"])
        existing = db.execute(
            "SELECT id FROM discord_accounts"
            " WHERE user_id = ? AND webhook_url_hash = ?",
            (uid, webhook_url_hash),
        ).fetchone()
        if not existing:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        db.execute(
            """
            INSERT INTO discord_accounts
                (name, webhook_url, webhook_url_hash, channel_id, user_id)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id, webhook_url_hash) DO UPDATE SET
                name = excluded.name,
                channel_id = excluded.channel_id,
                webhook_url = excluded.webhook_url
            """,
            (display_name, security.encrypt_secret(info["webhook_url"]),
             webhook_url_hash, info["channel_id"], uid),
        )
    return RedirectResponse(url="/accounts?status=discord_connected", status_code=303)


@app.post("/api/discord-accounts/{account_id}/test")
def test_discord_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM discord_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Discord account not found")
    success, message = test_discord_connection(security.decrypt_secret(account["webhook_url"]))
    return {"success": success, "message": message}


@app.post("/api/discord-accounts/{account_id}/delete")
def delete_discord_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "discord", account_id):
            return _render_accounts_error(
                request,
                "This Discord webhook is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM discord_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=discord_deleted", status_code=303)


@app.post("/api/telegram-accounts")
def add_telegram_account(
    request: Request,
    bot_token: str = Form(...),
    chat_id: str = Form(...),
    name: str = Form(""),
):
    """Verify a Telegram bot token + chat pair and store one account row.

    Synchronous route (threadpool-offloaded): the verification calls are
    blocking HTTPS. Reconnecting the same chat updates the stored row
    (token rotation, name refresh) instead of duplicating it.
    """
    try:
        info = telegram_connect(bot_token, chat_id)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    except TelegramAuthError as e:
        return _render_accounts_error(request, str(e))
    except TelegramNotFoundError as e:
        return _render_accounts_error(request, str(e))
    except TelegramError as e:
        logger.warning("Telegram connect failed: %s", e)
        return _render_accounts_error(request, str(e))
    except Exception:
        logger.exception("Telegram account verification failed")
        return _render_accounts_error(
            request, "Could not verify the Telegram bot. Try again."
        )

    display_name = name.strip()[:100] or info["name"]

    with get_db() as db:
        uid = current_user_id(request)
        existing = db.execute(
            "SELECT id FROM telegram_accounts"
            " WHERE user_id = ? AND chat_id = ?",
            (uid, info["chat_id"]),
        ).fetchone()
        if not existing:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        db.execute(
            """
            INSERT INTO telegram_accounts
                (name, bot_token, chat_id, user_id)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(user_id, chat_id) DO UPDATE SET
                name = excluded.name,
                bot_token = excluded.bot_token
            """,
            (display_name, security.encrypt_secret(info["bot_token"]),
             info["chat_id"], uid),
        )
    return RedirectResponse(url="/accounts?status=telegram_connected", status_code=303)


@app.post("/api/telegram-accounts/{account_id}/test")
def test_telegram_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM telegram_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Telegram account not found")
    success, message = test_telegram_connection(
        security.decrypt_secret(account["bot_token"]), account["chat_id"]
    )
    return {"success": success, "message": message}


@app.post("/api/telegram-accounts/{account_id}/delete")
def delete_telegram_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "telegram", account_id):
            return _render_accounts_error(
                request,
                "This Telegram account is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM telegram_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=telegram_deleted", status_code=303)


# ── API: Webhook Accounts ───────────────────────────────────────────────────

@app.post("/api/webhook-accounts")
def add_webhook_account(
    request: Request,
    url: str = Form(...),
    name: str = Form(""),
    headers_text: str = Form(""),
    body_template: str = Form(""),
):
    """Validate a generic webhook endpoint and store one account row for it.

    Synchronous route: URL validation runs the SSRF guard (multi mode), and
    nothing is POSTed at connect time — the Test button sends the real test
    delivery. Reconnecting the same URL updates the stored row (name always
    refreshes) instead of duplicating it. Header values are treated like
    credentials and never echoed back into the form, so a blank
    ``headers_text`` on reconnect is the natural result of a display-name-only
    edit, not a request to clear headers — it leaves the previously stored
    headers untouched. Submitting non-empty headers still fully replaces the
    old set.

    ``body_template`` follows the same keep-on-blank convention; the explicit
    ``{}`` sentinel clears a stored template and returns the endpoint to the
    default flat payload. Unlike headers, the template is not a credential —
    it renders the same sandboxed template engine as echo templates and is
    validated at connect time by rendering a sample item and parsing the
    result as JSON.
    """
    url = url.strip()
    if not url:
        return _render_accounts_error(request, "Enter the webhook URL.")

    try:
        url = webhook_normalize_url(url)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    if len(headers_text) > MAX_HEADERS_TEXT:
        return _render_accounts_error(request, "Headers are too long.")
    try:
        headers = webhook_parse_headers(headers_text)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    try:
        body_for_sql = webhook_normalize_body_template(body_template)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    if webhook_is_clear_body(body_template):
        # Explicit clear, expressed in SQL as the sentinel value.
        body_for_sql = webhook_clear_body_sentinel

    display_name = name.strip()[:100] or "Webhook"

    with get_db() as db:
        uid = current_user_id(request)
        # Cap counts NEW rows only: reconnecting the same endpoint (to rotate
        # a token) updates in place and must not be blocked at the cap.
        existing = db.execute(
            "SELECT id FROM webhook_accounts"
            " WHERE user_id = ? AND url = ?",
            (uid, url),
        ).fetchone()
        if not existing:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))
        # The keep-on-blank / clear-on-sentinel logic lives in the SQL CASE
        # so two concurrent first-connects cannot clobber each other (the
        # headers column follows the same pattern). Blank keeps the stored
        # template; the sentinel clears it; anything else replaces it.
        # On INITIAL connect there is no conflict row, so the CASE never
        # fires — a fresh row must store '' for the clear sentinel, never
        # the literal '{}', or dispatch would send empty JSON objects.
        if webhook_is_clear_body(body_template) and not existing:
            body_for_sql = ""
        db.execute(
            """
            INSERT INTO webhook_accounts (name, url, headers, body_template, user_id)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id, url) DO UPDATE SET
                name = excluded.name,
                headers = CASE WHEN excluded.headers != '{}'
                               THEN excluded.headers
                               ELSE webhook_accounts.headers END,
                body_template = CASE
                    WHEN excluded.body_template = '{}'
                        THEN ''
                    WHEN excluded.body_template != ''
                        THEN excluded.body_template
                    ELSE webhook_accounts.body_template
                END
            """,
            (display_name, url, dump_headers(headers), body_for_sql, uid),
        )
    return RedirectResponse(url="/accounts?status=webhook_connected", status_code=303)


@app.post("/api/webhook-accounts/{account_id}/test")
def test_webhook_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM webhook_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="Webhook account not found")
    # Sends a real test delivery to the endpoint with the account's headers —
    # a generic webhook has no read-only check. With a custom body template
    # the test payload is rendered from it, so a broken template shows up on
    # the Test button instead of on the first real echo.
    success, message = test_webhook_connection(
        account["url"],
        load_headers(account["headers"]),
        body_template=(account["body_template"] if "body_template" in account.keys() else "") or "",
    )
    return {"success": success, "message": message}


@app.post("/api/webhook-accounts/{account_id}/delete")
def delete_webhook_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "webhook", account_id):
            return _render_accounts_error(
                request,
                "This webhook is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM webhook_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=webhook_deleted", status_code=303)


@app.post("/api/github-accounts")
def add_github_account(
    request: Request,
    repo: str = Form(...),
    branch: str = Form("main"),
    token: str = Form(""),
    name: str = Form(""),
    committer_name: str = Form(""),
    committer_email: str = Form(""),
):
    """Validate a GitHub token against a repository and store one account row.

    Synchronous route: connect runs two live read-only GitHub calls
    (``GET /repos/{repo}`` proving the token, visibility, and push
    permission, then ``GET /repos/{repo}/branches/{branch}`` proving the
    branch), because a credential that saves but cannot write fails at 6am
    inside the scheduler where nobody sees it. Reconnecting the same
    repo+branch updates the stored row instead of duplicating it.

    The token is a credential: it is encrypted before storage, never
    selected for display, and a blank token field on reconnect leaves the
    stored token untouched (the ``{}`` sentinel clears it). A fresh token
    is re-validated with the same live read before it is stored.
    """
    try:
        repo = github.parse_repo(repo)
    except ValueError as e:
        return _render_accounts_error(request, str(e))
    branch = (branch or "").strip() or "main"

    token = token or ""
    token_value = token.strip()
    is_clear = token_value == "{}"
    encrypted_new = ""
    if token_value and not is_clear:
        # Live validation runs BEFORE and OUTSIDE the DB block: connect can
        # spend up to 2x30s on HTTP and must not hold a connection or a write
        # lock. A token that cannot see the repo or lacks push permission is
        # refused here, not at 6am in the scheduler.
        try:
            github.connect(token_value, repo, branch)
        except ValueError as e:
            return _render_accounts_error(request, str(e))
        except github.GitHubError as e:
            return _render_accounts_error(request, str(e))
        encrypted_new = security.encrypt_secret(token_value)

    with get_db() as db:
        uid = current_user_id(request)
        # Cap counts NEW rows only: reconnecting the same repo+branch (to
        # rotate a token) updates in place and must not be blocked at the cap.
        existing = db.execute(
            "SELECT id FROM github_accounts"
            " WHERE user_id = ? AND repo = ? AND branch = ?",
            (uid, repo, branch),
        ).fetchone()
        if not existing:
            try:
                _check_destination_cap(db, uid)
            except PlanError as e:
                return _render_accounts_error(request, str(e))

        stored_token = ""
        if existing:
            row = db.execute(
                "SELECT token FROM github_accounts WHERE id = ? AND user_id = ?",
                (existing["id"], uid),
            ).fetchone()
            stored_token = row["token"] if row else ""

        if not token_value and not stored_token and not existing:
            return _render_accounts_error(
                request, "Enter a GitHub personal access token."
            )
        if is_clear and not existing:
            # First connect has no stored credential to clear; the v1.65.0
            # lesson applies in reverse — ON CONFLICT never fires on a fresh
            # row, so the literal sentinel would be stored as the token.
            return _render_accounts_error(
                request, "Enter a GitHub personal access token."
            )
        if encrypted_new:
            encrypted = encrypted_new
        elif is_clear:
            # Explicit clear, expressed in SQL as the sentinel value.
            encrypted = "{}"
        else:
            # Blank token keeps the stored credential (display-name-only edit).
            encrypted = stored_token

        display_name = name.strip()[:100] or repo
        db.execute(
            """
            INSERT INTO github_accounts
                (name, repo, branch, token, committer_name, committer_email, user_id)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(user_id, repo, branch) DO UPDATE SET
                name = excluded.name,
                token = CASE
                    WHEN excluded.token = '{}' THEN ''
                    WHEN excluded.token != '' THEN excluded.token
                    ELSE github_accounts.token
                END,
                committer_name = excluded.committer_name,
                committer_email = excluded.committer_email
            """,
            (
                display_name, repo, branch, encrypted,
                committer_name.strip()[:100], committer_email.strip()[:200], uid,
            ),
        )
    return RedirectResponse(url="/accounts?status=github_connected", status_code=303)


@app.post("/api/github-accounts/{account_id}/test")
def test_github_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        account = db.execute(
            "SELECT * FROM github_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        ).fetchone()
    if not account:
        raise HTTPException(status_code=404, detail="GitHub account not found")
    # Read-only: re-reads the branch head with the stored token. Nothing is
    # written to the repository.
    try:
        token = security.decrypt_secret(account["token"])
    except ValueError:
        return {"success": False, "message": "GitHub credential could not be decrypted"}
    success, message = github.test_connection(token, account["repo"], account["branch"])
    return {"success": success, "message": message}


@app.post("/api/github-accounts/{account_id}/delete")
def delete_github_account(request: Request, account_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        if _dependent_echo_count_tx(db, uid, "github", account_id):
            return _render_accounts_error(
                request,
                "This GitHub account is used by echoes. Delete or reassign those echoes first.",
            )
        db.execute(
            "DELETE FROM github_accounts WHERE id = ? AND user_id = ?",
            (account_id, uid),
        )
    return RedirectResponse(url="/accounts?status=github_deleted", status_code=303)


# ── API: Settings ───────────────────────────────────────────────────────────

# Hosted-mode SMTP restrictions: the relay is tenant-supplied, and because it
# is dialed with raw smtplib (not the pinned httpx transport) none of the SSRF
# machinery applies. A tenant could otherwise aim it at cloud metadata or an
# internal service and use the connection result as a port/banner oracle.
_SMTP_ALLOWED_PORTS = {25, 465, 587, 2525}

# Per-IP throttle on the SMTP test button (hosted mode): caps outbound SMTP
# connection attempts so it cannot be used as an egress/connection hammer.
_smtp_test_lock = threading.Lock()
_smtp_test_attempts: dict[str, list[float]] = {}
_SMTP_TEST_LIMIT = 10
_SMTP_TEST_WINDOW = 600  # seconds
_SMTP_TEST_MAX_IPS = 10_000  # hard cap on tracked buckets
_smtp_last_sweep = 0.0  # monotonic time of the last global sweep


def _check_and_record_smtp_test(ip: str) -> bool:
    """Atomically rate-limit one SMTP test attempt.

    Returns True when the attempt must be refused (limit reached); otherwise
    records the attempt and returns False. Check-and-append happen under one
    lock so concurrent requests cannot all slip past the limit. A global
    sweep (at most once per window) prunes stale buckets, and a brand-new IP
    is refused once the bucket map is at _SMTP_TEST_MAX_IPS — so a source-IP
    spray cannot grow the dict unboundedly even within a single window.
    """
    with _smtp_test_lock:
        global _smtp_last_sweep
        now = time.monotonic()
        if now - _smtp_last_sweep >= _SMTP_TEST_WINDOW:
            for key in list(_smtp_test_attempts):
                recent = [t for t in _smtp_test_attempts[key] if now - t < _SMTP_TEST_WINDOW]
                if recent:
                    _smtp_test_attempts[key] = recent
                else:
                    _smtp_test_attempts.pop(key, None)
            _smtp_last_sweep = now
        recent = [t for t in _smtp_test_attempts.get(ip, []) if now - t < _SMTP_TEST_WINDOW]
        if recent:
            _smtp_test_attempts[ip] = recent
        else:
            _smtp_test_attempts.pop(ip, None)
        if len(recent) >= _SMTP_TEST_LIMIT:
            return True
        if len(_smtp_test_attempts) >= _SMTP_TEST_MAX_IPS and ip not in _smtp_test_attempts:
            # At the tracked-IP ceiling with a brand-new source: refuse rather
            # than let the map grow past the cap during an active spray.
            return True
        _smtp_test_attempts.setdefault(ip, []).append(now)
        return False


@app.post("/api/settings/smtp")
async def save_smtp_settings(
    request: Request,
    smtp_host: str = Form(...),
    smtp_port: int = Form(587),
    smtp_username: str = Form(""),
    smtp_password: str = Form(""),
    smtp_from_email: str = Form(""),
    smtp_from_name: str = Form("FeedEcho"),
    smtp_use_tls: str = Form("1"),
):
    uid = current_user_id(request)
    # Same validation the admin system-mail route enforces: these become
    # message headers, so control characters must fail here rather than at
    # send time. Per-tenant SMTP only reaches that tenant's own relay, but a
    # stored out-of-range port or CRLF in a header field is still a defect.
    if not 1 <= smtp_port <= 65535:
        return render("error.html", request, status_code=400, code=400,
                      message="SMTP port must be a number between 1 and 65535")
    # Hosted mode only: the tenant is untrusted relative to the VPS's internal
    # network, so the relay address must be a public host on a standard SMTP
    # port. Single mode keeps 1-65535 + localhost/LAN relays (unchanged).
    if settings.MULTI:
        if smtp_port not in _SMTP_ALLOWED_PORTS:
            return render(
                "error.html", request, status_code=400, code=400,
                message="SMTP port must be 25, 465, 587, or 2525",
            )
        try:
            validate_outbound_url(f"http://{smtp_host.strip()}")
        except (SSRFError, ValueError):
            # ValueError: a malformed host (unbracketed IPv6, embedded port,
            # stray whitespace) that urlparse chokes on — reject the same way.
            return render(
                "error.html", request, status_code=400, code=400,
                message="SMTP host must be a public hostname or IP address",
            )
    # No dot required in the domain: a self-hosted relay legitimately uses
    # feedecho@localhost or a bare internal hostname. This rejects
    # whitespace, control characters and anything without a single @.
    if smtp_from_email and not is_valid_email(
        smtp_from_email.strip(), require_domain_dot=False
    ):
        return render("error.html", request, status_code=400, code=400,
                      message="From address is not a valid email address")
    for label, field in (
        ("smtp_host", smtp_host),
        ("smtp_username", smtp_username),
        ("smtp_from_name", smtp_from_name),
        ("smtp_from_email", smtp_from_email),
    ):
        if re.search(r"[\r\n]", field):
            return render("error.html", request, status_code=400, code=400,
                          message=f"Invalid characters in {label}")
    values = {
        "smtp_host": smtp_host.strip(),
        "smtp_port": str(smtp_port),
        "smtp_username": smtp_username.strip(),
        "smtp_from_email": smtp_from_email.strip(),
        "smtp_from_name": smtp_from_name.strip(),
        "smtp_use_tls": smtp_use_tls,
    }
    with get_db() as db:
        _upsert_settings(db, uid, values)
        # Only update password if it's not the mask placeholder
        if smtp_password and smtp_password != "********":
            db.execute(
                "INSERT INTO settings (user_id, key, value) VALUES (?, ?, ?)"
                " ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value",
                (uid, "smtp_password", security.encrypt_secret(smtp_password)),
            )
    return RedirectResponse(url="/settings?status=saved", status_code=303)


@app.post("/api/settings/smtp/test")
def test_smtp(
    request: Request,
    test_email: str = Form(""),
):
    uid = current_user_id(request)
    if settings.MULTI:
        # Rate-limit the outbound SMTP connection attempt, and scrub the raw
        # smtplib error text below (it is a port/banner oracle to a tenant).
        if _check_and_record_smtp_test(auth._client_ip(request)):
            return {
                "success": False,
                "message": "Too many test attempts. Please wait before trying again.",
            }
    success, message = test_smtp_connection(
        test_email, user_id=current_user_id(request)
    )
    if not success and settings.MULTI:
        # Hosted: the exception string reveals what answered the connection
        # (refused / timeout / non-SMTP banner), which turns the test button
        # into a scanner. Log the detail server-side; hand back a generic
        # message. Single mode keeps the detail for the operator's own
        # debugging.
        logger.info("SMTP test failed for user %s: %s", uid, message)
        message = "SMTP connection failed. Check the server settings and try again."
    return {"success": success, "message": message}


@app.post("/api/settings/retry-notify")
async def save_retry_notify_settings(
    request: Request,
    retry_max_attempts: int = Form(5),
    retry_backoff_minutes: int = Form(5),
    notify_failure_threshold: int = Form(3),
    notify_email: str = Form(""),
):
    uid = current_user_id(request)
    values = {
        "retry_max_attempts": str(max(0, min(retry_max_attempts, 100))),
        "retry_backoff_minutes": str(max(1, min(retry_backoff_minutes, 1440))),
        "notify_failure_threshold": str(max(0, min(notify_failure_threshold, 100))),
        "notify_email": notify_email.strip(),
    }
    with get_db() as db:
        _upsert_settings(db, uid, values)
    return RedirectResponse(url="/settings?status=saved", status_code=303)


@app.post("/api/settings/alt-text")
async def save_alt_text_settings(
    request: Request,
    alt_text_ai_enabled: str = Form(""),
    alt_text_ai_base_url: str = Form(""),
    alt_text_ai_model: str = Form(""),
    alt_text_ai_api_key: str = Form(""),
):
    uid = current_user_id(request)
    import alt_text

    values = {
        "alt_text_ai_enabled": "1" if alt_text_ai_enabled else "0",
        # normalize_base_url: tenants paste FULL endpoints (Mistral's docs
        # show the complete /chat/completions URL); the code appends the
        # suffix itself, and a doubled path 404s silently at post time.
        "alt_text_ai_base_url": alt_text.normalize_base_url(alt_text_ai_base_url),
        "alt_text_ai_model": alt_text_ai_model.strip(),
    }
    with get_db() as db:
        _upsert_settings(db, uid, values)
        # Only update API key if it's not the mask placeholder
        if alt_text_ai_api_key and alt_text_ai_api_key != "********":
            db.execute(
                "INSERT INTO settings (user_id, key, value) VALUES (?, ?, ?)"
                " ON CONFLICT(user_id, key) DO UPDATE SET value = excluded.value",
                (uid, "alt_text_ai_api_key", security.encrypt_secret(alt_text_ai_api_key.strip())),
            )
    return RedirectResponse(url="/settings?status=saved", status_code=303)


@app.post("/api/settings/alt-text/test")
def test_alt_text(request: Request):
    """Test the vision API connection with a minimal request."""
    import alt_text
    uid = current_user_id(request)
    if not alt_text.is_enabled(user_id=uid):
        return {"success": False, "message": "AI alt text is not configured"}
    # A blocked address would otherwise surface as the same empty result as a
    # working endpoint that had nothing to say, so the test button would report
    # success for a configuration that never produces alt text.
    blocked = alt_text.endpoint_rejection_reason(user_id=uid)
    if blocked:
        return {"success": False, "message": f"Address refused: {blocked}"}
    try:
        # The probe image is a 96x96 white canvas with a solid red circle, not
        # a 1x1 pixel: current models (OpenAI's gpt-6 family confirmed
        # 2026-11-06) answer a 1x1 with "No image is attached", so the button
        # went green while proving nothing about whether vision actually
        # works. A visible shape comes back as "A solid red circle centered on
        # a plain white background", which is the proof the tenant wants.
        probe_png = alt_text.TEST_IMAGE_BYTES
        # user_id is mandatory here: omitting it silently fell back to
        # tenant 1, so a tenant tested user 1's endpoint and spent user 1's
        # API key while their own config went unverified.
        # attempt_alt_text (not generate_alt_text): a permanent 4xx used to
        # map to "" and report "API reachable" — the 2026-09-09 Mistral
        # 404 (pasted full endpoint as base URL) tested green while every
        # real image silently posted without alt text.
        result, reason = alt_text.attempt_alt_text(
            probe_png, "image/png", user_id=uid
        )
        if result and alt_text.looks_like_no_image_answer(result):
            # HTTP 200 with "No image is attached" means the probe never
            # reached the model: report it as a failure and show what came
            # back, instead of a green light over a sentence about no image.
            return {
                "success": False,
                "message": (
                    "API answered but reported no image: "
                    f"{result[:160]}"
                ),
            }
        if result:
            return {"success": True, "message": f"API working. Response: {result[:100]}"}
        if reason:
            return {"success": False, "message": f"API test failed: {reason}"}
        return {"success": False, "message": "API test failed: vision API is not configured or disabled"}
    except Exception as e:
        return {"success": False, "message": f"API test failed: {e}"}


# ── Saved Searches API (Phase 5) ──────────────────────────────────────────

@app.post("/api/saved-searches")
def create_saved_search(
    request: Request,
    name: str = Form(...),
    query: str = Form(...),
):
    uid = current_user_id(request)
    name = " ".join(name.strip().split())
    query = query.strip()
    if not (1 <= len(name) <= 60):
        raise HTTPException(status_code=400, detail="Name must be between 1 and 60 characters")
    if not query:
        raise HTTPException(status_code=400, detail="Query cannot be empty")
    if len(query) > 500:
        raise HTTPException(status_code=400, detail="Query is too long (max 500 characters)")

    with get_db() as db:
        _require_reader(db, uid)
        if settings.MULTI:
            plan = _user_plan(db, uid)
            current_count = db.execute(
                "SELECT COUNT(*) AS c FROM saved_searches WHERE user_id = ?",
                (uid,),
            ).fetchone()["c"]
            try:
                plans.check_saved_search_allowance(current_count, plan)
            except PlanError as e:
                raise HTTPException(status_code=402, detail=str(e))

        existing = db.execute(
            "SELECT id FROM saved_searches WHERE user_id = ? AND LOWER(name) = ?",
            (uid, name.lower()),
        ).fetchone()
        if existing:
            raise HTTPException(status_code=409, detail="Saved search already exists")

        max_pos = db.execute(
            "SELECT COALESCE(MAX(position), 0) AS m FROM saved_searches WHERE user_id = ?",
            (uid,),
        ).fetchone()["m"]
        row = db.execute(
            """
            INSERT INTO saved_searches (user_id, name, query, position)
            VALUES (?, ?, ?, ?) RETURNING id
            """,
            (uid, name, query, max_pos + 1),
        ).fetchone()

    # Invalidate count cache for this user
    _saved_search_counts_cache.invalidate(uid)

    if request.headers.get("accept", "").startswith("application/json"):
        return {"success": True, "id": row["id"], "name": name, "query": query}
    return RedirectResponse(url=f"/reader?saved={row['id']}", status_code=303)


@app.post("/api/saved-searches/{search_id}/rename")
def rename_saved_search(request: Request, search_id: int, name: str = Form(...)):
    uid = current_user_id(request)
    name = " ".join(name.strip().split())
    if not (1 <= len(name) <= 60):
        raise HTTPException(status_code=400, detail="Name must be between 1 and 60 characters")

    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT id FROM saved_searches WHERE id = ? AND user_id = ?",
            (search_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Saved search not found")

        dup = db.execute(
            "SELECT id FROM saved_searches WHERE user_id = ? AND LOWER(name) = ? AND id != ?",
            (uid, name.lower(), search_id),
        ).fetchone()
        if dup:
            raise HTTPException(status_code=409, detail="Saved search with this name already exists")

        db.execute(
            "UPDATE saved_searches SET name = ? WHERE id = ? AND user_id = ?",
            (name, search_id, uid),
        )

    _saved_search_counts_cache.invalidate(uid)
    if request.headers.get("accept", "").startswith("application/json"):
        return {"success": True, "name": name}
    return RedirectResponse(url=f"/reader?saved={search_id}", status_code=303)


@app.post("/api/saved-searches/{search_id}/delete")
def delete_saved_search(request: Request, search_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT id FROM saved_searches WHERE id = ? AND user_id = ?",
            (search_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Saved search not found")

        db.execute(
            "DELETE FROM saved_searches WHERE id = ? AND user_id = ?",
            (search_id, uid),
        )

    _saved_search_counts_cache.invalidate(uid)
    if request.headers.get("accept", "").startswith("application/json"):
        return {"success": True}
    return RedirectResponse(url="/reader", status_code=303)


# ── Folders API ─────────────────────────────────────────────────────────────

@app.post("/api/folders")
def create_folder(request: Request, name: str = Form(...)):
    uid = current_user_id(request)
    name = " ".join(name.strip().split())
    if not (1 <= len(name) <= 60):
        raise HTTPException(status_code=400, detail="Folder name must be between 1 and 60 characters")
    with get_db() as db:
        existing = db.execute(
            "SELECT id FROM folders WHERE user_id = ? AND LOWER(name) = ?",
            (uid, name.lower()),
        ).fetchone()
        if existing:
            raise HTTPException(status_code=409, detail="Folder already exists")
        max_pos = db.execute(
            "SELECT COALESCE(MAX(position), 0) AS m FROM folders WHERE user_id = ?",
            (uid,),
        ).fetchone()["m"]
        db.execute(
            "INSERT INTO folders (user_id, name, position) VALUES (?, ?, ?)",
            (uid, name, max_pos + 1),
        )
    return RedirectResponse(url="/feeds", status_code=303)


@app.post("/api/folders/{folder_id}/rename")
def rename_folder(request: Request, folder_id: int, name: str = Form(...)):
    uid = current_user_id(request)
    name = " ".join(name.strip().split())
    if not (1 <= len(name) <= 60):
        raise HTTPException(status_code=400, detail="Folder name must be between 1 and 60 characters")
    with get_db() as db:
        folder = db.execute(
            "SELECT id FROM folders WHERE id = ? AND user_id = ?",
            (folder_id, uid),
        ).fetchone()
        if not folder:
            raise HTTPException(status_code=404, detail="Folder not found")
        dup = db.execute(
            "SELECT id FROM folders WHERE user_id = ? AND LOWER(name) = ? AND id != ?",
            (uid, name.lower(), folder_id),
        ).fetchone()
        if dup:
            raise HTTPException(status_code=409, detail="Folder already exists")
        db.execute(
            "UPDATE folders SET name = ? WHERE id = ? AND user_id = ?",
            (name, folder_id, uid),
        )
    # Saved searches can filter on folder:<name>, so a rename changes them
    _saved_search_counts_cache.invalidate(uid)
    return RedirectResponse(url="/feeds", status_code=303)


@app.post("/api/folders/{folder_id}/delete")
def delete_folder(request: Request, folder_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        folder = db.execute(
            "SELECT id FROM folders WHERE id = ? AND user_id = ?",
            (folder_id, uid),
        ).fetchone()
        if not folder:
            raise HTTPException(status_code=404, detail="Folder not found")
        # Null out folder_id on feeds belonging to this user first
        db.execute(
            "UPDATE feeds SET folder_id = NULL WHERE folder_id = ? AND user_id = ?",
            (folder_id, uid),
        )
        db.execute(
            "DELETE FROM folders WHERE id = ? AND user_id = ?",
            (folder_id, uid),
        )
    # Saved searches can filter on folder:<name>, so deleting one changes them
    _saved_search_counts_cache.invalidate(uid)
    return RedirectResponse(url="/feeds", status_code=303)


@app.post("/api/feeds/{feed_id}/folder")
def set_feed_folder(request: Request, feed_id: int, folder_id: str = Form("")):
    uid = current_user_id(request)
    target_folder_id = _filter_int(folder_id)
    with get_db() as db:
        feed = db.execute(
            "SELECT id FROM feeds WHERE id = ? AND user_id = ? AND deleted_at IS NULL",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        if target_folder_id is not None:
            folder = db.execute(
                "SELECT id FROM folders WHERE id = ? AND user_id = ?",
                (target_folder_id, uid),
            ).fetchone()
            if not folder:
                raise HTTPException(status_code=404, detail="Folder not found")
        db.execute(
            "UPDATE feeds SET folder_id = ? WHERE id = ? AND user_id = ?",
            (target_folder_id, feed_id, uid),
        )
    # Saved searches can filter on folder:<name>, so moving a feed changes them
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "folder_id": target_folder_id}


@app.post("/api/folders/reorder")
def reorder_folders(request: Request, ids: str = Form(...)):
    uid = current_user_id(request)
    id_list = [_filter_int(i.strip()) for i in ids.split(",") if i.strip()]
    id_list = [i for i in id_list if i is not None]
    with get_db() as db:
        for pos, fid in enumerate(id_list):
            db.execute(
                "UPDATE folders SET position = ? WHERE id = ? AND user_id = ?",
                (pos, fid, uid),
            )
    return {"success": True}


# ── Queue Slots Helper ──────────────────────────────────────────────────────
# next_free_slot lives in scheduler.py (single source of truth, shared with
# the dispatch loop); app.py imports it at module scope.


def _resolve_feed_url(url: str) -> str:
    """Resolve a pasted site URL to its feed URL via HTML link autodiscovery.

    Add/edit-time convenience only — never a gate. Any failure (network
    error, SSRF block, malformed page) falls back to the pasted URL
    unchanged, and the resolved URL is re-validated through app.validate_url
    for scheme/host consistency before it is stored.
    """
    try:
        info = discover_feed_url(url)
    except (httpx.HTTPError, SSRFError, ValueError):
        return url
    resolved = info.get("url") or url
    try:
        return validate_url(resolved)
    except HTTPException:
        return url


@app.post("/api/feeds")
def add_feed(
    request: Request,
    name: str = Form(...),
    url: str = Form(...),
    poll_interval: int = Form(15),
    folder_id: str = Form(""),
):
    uid = current_user_id(request)
    url = validate_url(url)
    poll_interval = max(1, min(poll_interval, 1440))
    target_folder_id = _filter_int(folder_id)

    # HTML link autodiscovery (F4) runs BEFORE any DB work: it does blocking
    # network I/O (httpx + DNS), and sync `def` routes run in the threadpool
    # instead of on the event loop, while a `with get_db()` block held open
    # across the fetch would tie up a connection on both dialects. Resolve
    # first so the plan-cap and upsert checks below see the final URL (a
    # re-add via a home-page paste upserts instead of 402ing at the cap);
    # any discovery failure keeps the raw input.
    url = _resolve_feed_url(url)
    with get_db() as db:
        if target_folder_id is not None:
            folder = db.execute(
                "SELECT id FROM folders WHERE id = ? AND user_id = ?",
                (target_folder_id, uid),
            ).fetchone()
            if not folder:
                target_folder_id = None
        if settings.MULTI:
            # Cap NEW rows only: re-adding an existing URL is an upsert that
            # changes nothing the cap measures. The re-add check runs on the
            # RESOLVED url so a home-page paste of an already-stored feed
            # isn't misread as a new feed (and 402ed) at the plan cap.
            existing_feed = db.execute(
                "SELECT id FROM feeds"
                " WHERE user_id = ? AND url = ? AND deleted_at IS NULL",
                (uid, url),
            ).fetchone()
            if existing_feed is None:
                plan = _user_plan(db, uid)
                count = db.execute(
                    "SELECT COUNT(*) AS c FROM feeds WHERE user_id = ? AND deleted_at IS NULL",
                    (uid,),
                ).fetchone()["c"]
                try:
                    plans.check_feed_allowance(count, plan)
                except PlanError as e:
                    raise HTTPException(status_code=402, detail=str(e))
                poll_interval = plans.clamp_poll_interval(poll_interval, plan)
            else:
                poll_interval = plans.clamp_poll_interval(
                    poll_interval, _user_plan(db, uid)
                )
        db.execute(
            """
            INSERT INTO feeds (name, url, poll_interval, user_id, folder_id)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id, url) WHERE deleted_at IS NULL DO UPDATE SET
                name = excluded.name,
                poll_interval = excluded.poll_interval,
                folder_id = excluded.folder_id
            """,
            (name, url, poll_interval, uid, target_folder_id),
        )
    return RedirectResponse(url="/feeds", status_code=303)


@app.get("/api/feeds/opml")
def export_opml(request: Request):
    """Export this user's feeds as an OPML 2.0 document (issue #11, Tier 2 & Phase 1)."""
    uid = current_user_id(request)
    with get_db() as db:
        folders = db.execute(
            "SELECT id, name FROM folders WHERE user_id = ? ORDER BY position, name",
            (uid,),
        ).fetchall()
        feeds = db.execute(
            "SELECT name, url, folder_id FROM feeds WHERE deleted_at IS NULL AND user_id = ? ORDER BY name",
            (uid,),
        ).fetchall()

    folder_map = {f["id"]: [] for f in folders}
    uncategorized = []
    for f in feeds:
        if f["folder_id"] and f["folder_id"] in folder_map:
            folder_map[f["folder_id"]].append(f)
        else:
            uncategorized.append(f)

    outline_lines = []
    for fol in folders:
        fol_feeds = folder_map[fol["id"]]
        if not fol_feeds:
            continue
        outline_lines.append(f'    <outline text={quoteattr(fol["name"])}>')
        for f in fol_feeds:
            outline_lines.append(
                f'      <outline text={quoteattr(f["name"])} type="rss" xmlUrl={quoteattr(f["url"])} />'
            )
        outline_lines.append("    </outline>")

    for f in uncategorized:
        outline_lines.append(
            f'    <outline text={quoteattr(f["name"])} type="rss" xmlUrl={quoteattr(f["url"])} />'
        )

    outlines = "\n".join(outline_lines)
    xml = (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        '<opml version="2.0">\n'
        "  <head><title>FeedEcho subscriptions</title></head>\n"
        "  <body>\n"
        f"{outlines}\n"
        "  </body>\n"
        "</opml>\n"
    )
    return Response(
        content=xml,
        media_type="text/x-opml",
        headers={"Content-Disposition": 'attachment; filename="feedecho-subscriptions.opml"'},
    )


@app.post("/api/feeds/opml")
def import_opml(
    request: Request,
    opml: str = Form(""),
    file: UploadFile = File(None),
):
    """Bulk-add feeds from an OPML document or file upload (issue #11, Tier 2 & Phase 1)."""
    uid = current_user_id(request)
    content_str = ""
    if file is not None and file.filename:
        raw_bytes = file.file.read(2_000_001)
        if len(raw_bytes) > 2_000_000:
            raise HTTPException(status_code=400, detail="OPML file exceeds 2 MB limit")
        try:
            content_str = raw_bytes.decode("utf-8")
        except UnicodeDecodeError:
            raise HTTPException(status_code=400, detail="OPML file is not valid UTF-8 text")
    elif opml:
        if len(opml.encode("utf-8")) > 2_000_000:
            raise HTTPException(status_code=400, detail="OPML content exceeds 2 MB limit")
        content_str = opml
    else:
        raise HTTPException(status_code=400, detail="No OPML data or file provided")

    # Reject DTD/entity declarations before parsing: ElementTree doesn't fetch
    # external entities, but a <!DOCTYPE> subset can still trigger entity
    # expansion (billion-laughs) DoS.
    if "<!DOCTYPE" in content_str.upper() or "<!ENTITY" in content_str.upper():
        raise HTTPException(status_code=400, detail="OPML must not contain a DOCTYPE or entity declarations")
    try:
        root = ElementTree.fromstring(content_str)
    except ElementTree.ParseError:
        raise HTTPException(status_code=400, detail="Invalid OPML XML")

    with get_db() as db:
        reader_allowed = not settings.MULTI or plans.reader_enabled(_user_plan(db, uid))
        existing_urls = {
            r["url"] for r in db.execute(
                "SELECT url FROM feeds WHERE deleted_at IS NULL AND user_id = ?", (uid,)
            ).fetchall()
        }
        # Pre-fetch user's folders
        folder_rows = db.execute(
            "SELECT id, name FROM folders WHERE user_id = ?", (uid,)
        ).fetchall()
        folders_by_lower = {r["name"].lower(): r["id"] for r in folder_rows}

        imported = duplicate = invalid = capped = 0
        total_outlines = 0

        def _local_tag(elem):
            tag = elem.tag
            return tag.split("}", 1)[1] if "}" in tag else tag

        def walk(node, current_folder_id: int | None, depth: int):
            nonlocal imported, duplicate, invalid, capped, total_outlines
            if depth > 10:
                return
            for child in node:
                if _local_tag(child).lower() != "outline":
                    continue
                total_outlines += 1
                if total_outlines > 1000:
                    break
                xml_url = (child.get("xmlUrl") or "").strip()
                if xml_url:
                    # Feed outline
                    try:
                        valid_url = validate_url(xml_url)
                    except HTTPException:
                        invalid += 1
                        continue
                    if valid_url in existing_urls:
                        duplicate += 1
                        continue
                    if settings.MULTI:
                        count = db.execute(
                            "SELECT COUNT(*) AS c FROM feeds WHERE user_id = ? AND deleted_at IS NULL",
                            (uid,),
                        ).fetchone()["c"]
                        try:
                            plans.check_feed_allowance(count, _user_plan(db, uid))
                        except PlanError:
                            capped += 1
                            continue
                    title = (child.get("title") or child.get("text") or "").strip() or valid_url
                    row = db.execute(
                        "INSERT INTO feeds (name, url, read_enabled, user_id, folder_id)"
                        " VALUES (?, ?, ?, ?, ?)"
                        " ON CONFLICT(user_id, url) WHERE deleted_at IS NULL DO NOTHING",
                        (title, valid_url, 1 if reader_allowed else 0, uid, current_folder_id),
                    )
                    if row.rowcount == 0:
                        # Lost a race with a concurrent add/import of the same
                        # user+url: count it as a duplicate, not a 500.
                        duplicate += 1
                    else:
                        existing_urls.add(valid_url)
                        imported += 1
                else:
                    # Folder / container outline
                    folder_name = (child.get("title") or child.get("text") or "").strip()
                    folder_name = " ".join(folder_name.split())[:60]
                    folder_id_to_pass = current_folder_id
                    if folder_name:
                        low_name = folder_name.lower()
                        if low_name in folders_by_lower:
                            folder_id_to_pass = folders_by_lower[low_name]
                        else:
                            max_pos = db.execute(
                                "SELECT COALESCE(MAX(position), 0) AS m FROM folders WHERE user_id = ?",
                                (uid,),
                            ).fetchone()["m"]
                            new_fid = db.execute(
                                "INSERT INTO folders (user_id, name, position) VALUES (?, ?, ?) RETURNING id",
                                (uid, folder_name, max_pos + 1),
                            ).fetchone()["id"]
                            folders_by_lower[low_name] = new_fid
                            folder_id_to_pass = new_fid
                    walk(child, folder_id_to_pass, depth + 1)

        body_elem = None
        for child in root:
            if _local_tag(child).lower() == "body":
                body_elem = child
                break
        walk(body_elem if body_elem is not None else root, None, 1)

    # Importing feeds and folders can alter saved-search matches
    _saved_search_counts_cache.invalidate(uid)
    return RedirectResponse(
        url=f"/feeds?imported={imported}&duplicate={duplicate}&invalid={invalid}&capped={capped}",
        status_code=303,
    )


@app.post("/api/feeds/{feed_id}/edit")
def edit_feed(
    request: Request,
    feed_id: int,
    name: str = Form(...),
    url: str = Form(...),
    poll_interval: int = Form(15),
    mute_keywords: str = Form(""),
    folder_id: str | None = Form(None),
    ai_alt_override: str = Form(""),
):
    """Update a feed's name, URL, poll interval, mute keywords, folder, or
    AI-alt override flag (issues #3, #11).

    Changing the URL invalidates the cursor: last_item_id belonged to the
    old feed, and comparing it against the new feed's item IDs could
    silently skip (or mis-dedupe) items. Resetting it means the next poll
    re-initializes to the newest item — the same no-backfill behaviour as
    adding a fresh feed. Renames and interval changes keep the cursor.
    """
    uid = current_user_id(request)
    name = name.strip()
    if not name:
        raise HTTPException(status_code=400, detail="Feed name is required")
    url = validate_url(url)
    poll_interval = max(1, min(poll_interval, 1440))
    mute_keywords = (mute_keywords or "").strip()
    target_folder_id = _filter_int(folder_id) if folder_id is not None else None
    # Checkbox convention, same as the echo edit handler: a checked box posts
    # its value, an unchecked box posts nothing. Only the known truthy tokens
    # count as on; anything else (including absent) is off.
    ai_alt_override_flag = 1 if ai_alt_override in ("1", "true", "on") else 0

    # Discovery (F4) runs BEFORE the DB transaction — blocking network I/O
    # in a sync `def` route stays in the threadpool, off the event loop.
    with get_db() as db:
        stored_url = db.execute(
            "SELECT url FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
        if not stored_url:
            raise HTTPException(status_code=404, detail="Feed not found")
    if stored_url["url"] != url:
        # URL change: resolve a pasted site page to its feed URL (F4).
        # Failure falls back to the raw input, same as add_feed.
        url = _resolve_feed_url(url)

    with get_db() as db:
        if target_folder_id is not None:
            fol = db.execute(
                "SELECT id FROM folders WHERE id = ? AND user_id = ?",
                (target_folder_id, uid),
            ).fetchone()
            if not fol:
                target_folder_id = None
        if settings.MULTI:
            # Clamp to the plan's floor, never reject: tightening an existing
            # feed's cadence is fine, lowering it below the plan is not.
            poll_interval = plans.clamp_poll_interval(
                poll_interval, _user_plan(db, uid)
            )
        feed = db.execute(
            "SELECT url, folder_id FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        url_changed = feed["url"] != url

        # If folder_id was omitted entirely from the form, preserve existing folder
        final_folder_id = target_folder_id if folder_id is not None else feed["folder_id"]

        if url_changed:
            db.execute(
                """
                UPDATE feeds
                   SET name = ?, url = ?, poll_interval = ?, mute_keywords = ?,
                       folder_id = ?, ai_alt_override = ?, last_item_id = NULL,
                       etag = NULL, last_modified = NULL
                 WHERE id = ? AND deleted_at IS NULL AND user_id = ?
                """,
                (name, url, poll_interval, mute_keywords, final_folder_id, ai_alt_override_flag, feed_id, uid),
            )
        else:
            db.execute(
                "UPDATE feeds SET name = ?, url = ?, poll_interval = ?, mute_keywords = ?, folder_id = ?, "
                "ai_alt_override = ? WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
                (name, url, poll_interval, mute_keywords, final_folder_id, ai_alt_override_flag, feed_id, uid),
            )
    # Saved-search counts apply mute keywords and folder scoping, so editing
    # either changes them.
    _saved_search_counts_cache.invalidate(uid)
    return RedirectResponse(url="/feeds", status_code=303)


@app.post("/api/feeds/{feed_id}/delete")
async def delete_feed(request: Request, feed_id: int):
    """Soft-delete a feed. Echoes and post history are preserved.

    A hard DELETE would cascade (echoes -> posted_items/digest_items) and wipe
    the cross-post audit trail, so feeds are only marked deleted_at. The feed
    disappears from listings and is skipped by the scheduler, but its echo
    config and history remain on the /echoes and /history pages.

    The feed's URL, poke token, stored reader items, and any queued curation
    posts do NOT survive, though: feed URLs often embed private tokens and the
    reader rows are unreachable once the feed is gone, so they are removed
    here. posted_items keeps its own copies of what was delivered, so the
    audit trail is unaffected.
    """
    uid = current_user_id(request)
    with get_db() as db:
        cur = db.execute(
            """
            UPDATE feeds
               SET deleted_at = CURRENT_TIMESTAMP,
                   url = '',
                   poke_token = NULL
             WHERE id = ?
               AND deleted_at IS NULL
               AND user_id = ?
            """,
            (feed_id, uid),
        )
        if cur.rowcount > 0:
            # Purge ONLY when this user actually owned a live feed row.
            # feed_items has no user_id column, so an unconditional DELETE
            # would let any tenant wipe another tenant's reader items (IDOR).
            db.execute("DELETE FROM feed_items WHERE feed_id = ?", (feed_id,))
            # Queued reader-curation posts must never send from a deleted
            # feed (their stored item rows are gone now, so a flush would
            # dispatch empty content). Finalize them with a reason.
            db.execute(
                "UPDATE queued_posts SET status = 'failed',"
                " error_message = 'Feed deleted'"
                " WHERE feed_id = ? AND status IN ('queued', 'draft')",
                (feed_id,),
            )
    # Soft-deleting a feed changes which items saved-search counts cover
    _saved_search_counts_cache.invalidate(uid)
    return RedirectResponse(url="/feeds", status_code=303)


@app.post("/api/feeds/{feed_id}/test")
def test_feed(request: Request, feed_id: int):
    """Fetch a feed and return preview of items."""
    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT * FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
    if not feed:
        raise HTTPException(status_code=404, detail="Feed not found")
    try:
        feed_data = fetch_feed(feed["url"])
        return {"success": True, "preview": _preview_payload(feed_data)}
    except SSRFError as e:
        return {"success": False, "error": str(e)}
    except Exception as e:
        return {"success": False, "error": str(e)}


def _preview_payload(feed_data: dict) -> dict:
    """Shared item-preview shape for /api/feeds/{id}/test and /api/feeds/preview."""
    return {
        "title": feed_data["title"],
        "type": feed_data["type"],
        "item_count": len(feed_data["items"]),
        "items": feed_data["items"][:5],
    }


@app.get("/api/feeds/discover")
def discover_feed_api(request: Request, url: str = Query(...)):
    """Resolve a pasted site URL to its feed URL via link autodiscovery (F4)."""
    uid = current_user_id(request)
    url = validate_url(url)
    try:
        info = discover_feed_url(url)
    except (httpx.HTTPError, SSRFError, ValueError) as e:
        raise HTTPException(status_code=400, detail=f"Could not discover feeds for {url}: {e}")
    return {
        "found": bool(info.get("url")),
        "url": info.get("url"),
        "title": info.get("title") or "",
        "alternates": info.get("alternates") or [],
    }


@app.post("/api/feeds/preview")
def test_feed_preview(request: Request, url: str = Form(...)):
    """Preview a feed's items from a raw URL, without saving a feed row (F4)."""
    url = validate_url(url)
    try:
        feed_data = fetch_feed(url)
        return {"success": True, "preview": _preview_payload(feed_data)}
    except SSRFError as e:
        return {"success": False, "error": str(e)}
    except Exception as e:
        return {"success": False, "error": str(e)}


@app.post("/api/feeds/{feed_id}/init")
def init_feed(request: Request, feed_id: int):
    """Initialize a feed's last_item_id so it only posts new items going forward."""
    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT * FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        try:
            feed_data = fetch_feed(feed["url"])
            if feed_data["items"]:
                last_id = feed_data["items"][0]["id"]
                db.execute(
                    "UPDATE feeds SET last_item_id = ? WHERE id = ? AND user_id = ?",
                    (last_id, feed_id, uid),
                )
                return {"success": True, "message": f"Initialized. Last item: {feed_data['items'][0]['title'][:60]}"}
            return {"success": True, "message": "Feed has no items"}
        except SSRFError as e:
            return {"success": False, "error": str(e)}
        except Exception as e:
            return {"success": False, "error": str(e)}


@app.post("/api/feeds/{feed_id}/pause")
async def pause_feed(request: Request, feed_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT paused FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        new_val = 0 if feed["paused"] else 1
        db.execute(
            "UPDATE feeds SET paused = ? WHERE id = ? AND user_id = ?",
            (new_val, feed_id, uid),
        )
    return {"success": True, "paused": bool(new_val)}


@app.post("/api/feeds/{feed_id}/fetch")
def fetch_now(request: Request, feed_id: int):
    """Trigger an immediate feed check."""
    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT id FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
    if not feed:
        raise HTTPException(status_code=404, detail="Feed not found")
    try:
        check_feed(feed_id)
        return {"success": True, "message": "Feed checked"}
    except SSRFError as e:
        return {"success": False, "error": str(e)}
    except Exception as e:
        return {"success": False, "error": str(e)}


POKE_COOLDOWN_SECONDS = 60


@app.post("/api/feeds/{feed_id}/poke-token")
def poke_token(request: Request, feed_id: int, regenerate: str = Form("0")):
    """Generate or reveal the secret poke URL for a feed.

    The token lets a publisher/automation trigger an immediate fetch without
    a FeedEcho session (see GET /poke/{token}). Revealing is idempotent;
    `regenerate=1` rotates the token, revoking any previously shared URL.
    """
    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT poke_token FROM feeds"
            " WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        token = feed["poke_token"]
        if not token or regenerate in ("1", "true", "on"):
            token = secrets.token_urlsafe(24)
            db.execute(
                "UPDATE feeds SET poke_token = ? WHERE id = ? AND user_id = ?",
                (token, feed_id, uid),
            )
    base = settings.BASE_URL.rstrip("/") if settings.BASE_URL else ""
    poke_url = f"{base}/poke/{token}" if base else f"/poke/{token}"
    return {"success": True, "poke_token": token, "poke_url": poke_url}


@app.api_route("/poke/{token}", methods=["GET", "POST"])
def poke(token: str, response: Response, background_tasks: BackgroundTasks):
    """Unauthenticated poke: fetch the feed that owns this secret token now.

    Deliberately session-free, and the feed id is carried only inside the
    token, so the URL is safe to hand to third parties. No URL is taken from
    the request (the feed URL was validated at add time), so there is no SSRF
    surface here. A poke inside the cooldown window is a no-op so a leaked
    token can't be turned into a fetch hammer against the feed's origin.
    """
    response.headers["X-Robots-Tag"] = "noindex, nofollow"
    now_dt = datetime.now(timezone.utc)
    now = now_dt.strftime("%Y-%m-%d %H:%M:%S")
    cutoff = (now_dt - timedelta(seconds=POKE_COOLDOWN_SECONDS)).strftime(
        "%Y-%m-%d %H:%M:%S"
    )
    with get_db() as db:
        feed = db.execute(
            "SELECT id, paused FROM feeds"
            " WHERE poke_token = ? AND deleted_at IS NULL",
            (token,),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Unknown poke token")
        if feed["paused"]:
            return {"ok": True, "skipped": "feed_paused"}

        # Atomic cooldown acquisition: exactly one concurrent request updates the row
        res = db.execute(
            "UPDATE feeds SET last_poked_at = ? WHERE id = ?"
            " AND (last_poked_at IS NULL OR last_poked_at <= ?)",
            (now, feed["id"], cutoff),
        )
        if res.rowcount == 0:
            return {"ok": True, "skipped": "recently_poked"}
        feed_id = feed["id"]
    background_tasks.add_task(check_feed, feed_id)
    return {"ok": True}


@app.post("/api/history/{posted_id}/retry")
async def retry_post(request: Request, posted_id: int):
    """Force a failed or gave_up row back to retryable: clears backoff and
    resets the attempt counter so the next feed check reprocesses it."""
    uid = current_user_id(request)
    with get_db() as db:
        result = db.execute(
            """UPDATE posted_items
                  SET status = 'failed',
                      attempt_count = 0,
                      next_retry_at = NULL,
                      error_message = NULL
                WHERE id = ?
                  AND status IN ('failed', 'gave_up')
                  AND echo_id IN (SELECT id FROM echoes WHERE user_id = ?)""",
            (posted_id, uid),
        )
        if result.rowcount != 1:
            raise HTTPException(status_code=404, detail="No failed post with that id")
    return {"success": True, "message": "Post queued for retry on next feed check"}


@app.post("/api/history/{posted_id}/give-up")
async def give_up_post(request: Request, posted_id: int):
    """Mark a failed row terminal so the feed cursor can advance past it."""
    uid = current_user_id(request)
    with get_db() as db:
        result = db.execute(
            """UPDATE posted_items
                  SET status = 'gave_up',
                      next_retry_at = NULL
                WHERE id = ?
                  AND status = 'failed'
                  AND echo_id IN (SELECT id FROM echoes WHERE user_id = ?)""",
            (posted_id, uid),
        )
        if result.rowcount != 1:
            raise HTTPException(status_code=404, detail="No failed post with that id")
    return {"success": True, "message": "Post marked as given up"}


# ── Reader ─────────────────────────────────────────────────────────────────

@app.post("/api/feeds/{feed_id}/reader-toggle")
def toggle_reader_feed(request: Request, feed_id: int):
    """Toggle whether a feed is ingested for reading (issue #11)."""
    uid = current_user_id(request)
    with get_db() as db:
        result = db.execute(
            "UPDATE feeds SET read_enabled = 1 - read_enabled"
            " WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        )
        if result.rowcount != 1:
            raise HTTPException(status_code=404, detail="Feed not found")
        row = db.execute("SELECT read_enabled FROM feeds WHERE id = ?", (feed_id,)).fetchone()
    # Toggling ingestion changes which items saved-search counts cover
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "read_enabled": bool(row["read_enabled"])}


@app.post("/api/reader/{item_id}/read")
def reader_toggle_read(request: Request, item_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        result = db.execute(
            "UPDATE feed_items SET is_read = 1 - is_read"
            " WHERE id = ? AND feed_id IN (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
            (item_id, uid),
        )
        if result.rowcount != 1:
            raise HTTPException(status_code=404, detail="Item not found")
        row = db.execute("SELECT is_read FROM feed_items WHERE id = ?", (item_id,)).fetchone()
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "is_read": bool(row["is_read"])}


@app.post("/api/reader/{item_id}/star")
def reader_toggle_star(request: Request, item_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        result = db.execute(
            "UPDATE feed_items SET starred = 1 - starred"
            " WHERE id = ? AND feed_id IN (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
            (item_id, uid),
        )
        if result.rowcount != 1:
            raise HTTPException(status_code=404, detail="Item not found")
        row = db.execute("SELECT starred FROM feed_items WHERE id = ?", (item_id,)).fetchone()
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "starred": bool(row["starred"])}


def _csv_formula_safe(value: str) -> str:
    """Neutralize spreadsheet formula injection in exported CSV cells.

    A cell whose text starts with =, +, - or @ is evaluated as a formula
    when the CSV is opened in Excel/LibreOffice. Prefixing with an apostrophe
    defuses it (OWASP CSV injection guidance).
    """
    if value.lstrip()[:1] in ("=", "+", "-", "@"):
        return "'" + value
    return value


@app.get("/api/reader/starred/export")
def reader_export_starred(request: Request, format: str = "csv"):
    """Download every starred reader item as CSV or JSON."""
    uid = current_user_id(request)
    fmt = (format or "").lower()
    if fmt not in ("csv", "json"):
        raise HTTPException(status_code=400, detail="format must be csv or json")
    with get_db() as db:
        _require_reader(db, uid)
        rows = db.execute(
            "SELECT fi.id, fi.item_id, fi.title, fi.author, fi.link, fi.summary,"
            " fi.is_read, fi.published_at, f.name AS feed_name"
            " FROM feed_items fi JOIN feeds f ON f.id = fi.feed_id"
            " WHERE fi.starred = 1 AND f.user_id = ? AND f.deleted_at IS NULL"
            " ORDER BY fi.published_at IS NULL, fi.published_at DESC, fi.id DESC",
            (uid,),
        ).fetchall()

    items = [
        {
            "id": r["id"],
            "item_id": r["item_id"],
            "title": r["title"] or "",
            "feed": r["feed_name"] or "",
            "author": r["author"] or "",
            "link": r["link"] or "",
            "published_at": timestamp_str(r["published_at"]),
            "is_read": bool(r["is_read"]),
            "summary": r["summary"] or "",
        }
        for r in rows
    ]

    if fmt == "csv":
        buf = io.StringIO()
        writer = csv.writer(buf, lineterminator="\n")
        writer.writerow(["title", "feed", "author", "link", "published_at"])
        for it in items:
            writer.writerow(
                [_csv_formula_safe(it[c]) for c in ("title", "feed", "author", "link", "published_at")]
            )
        return Response(
            content=buf.getvalue(),
            media_type="text/csv; charset=utf-8",
            headers={"Content-Disposition": 'attachment; filename="feedecho-starred.csv"'},
        )

    payload = {
        "format": "feedecho-starred-export",
        "version": 1,
        "count": len(items),
        "items": items,
    }
    return Response(
        content=json.dumps(payload, indent=2),
        media_type="application/json",
        headers={"Content-Disposition": 'attachment; filename="feedecho-starred.json"'},
    )


@app.post("/api/reader/mark-all-read")
def reader_mark_all_read(
    request: Request,
    feed_id: int = Form(None),
    folder_id: int = Form(None),
):
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        if feed_id is not None:
            owns = db.execute(
                "SELECT id FROM feeds WHERE id = ? AND user_id = ? AND deleted_at IS NULL",
                (feed_id, uid),
            ).fetchone()
            if not owns:
                raise HTTPException(status_code=404, detail="Feed not found")
            rows = db.execute(
                "SELECT id FROM feed_items WHERE is_read = 0 AND feed_id = ?",
                (feed_id,),
            ).fetchall()
            db.execute(
                "UPDATE feed_items SET is_read = 1 WHERE is_read = 0 AND feed_id = ?",
                (feed_id,),
            )
        elif folder_id is not None:
            owns_folder = db.execute(
                "SELECT id FROM folders WHERE id = ? AND user_id = ?",
                (folder_id, uid),
            ).fetchone()
            if not owns_folder:
                raise HTTPException(status_code=404, detail="Folder not found")
            rows = db.execute(
                "SELECT id FROM feed_items WHERE is_read = 0 AND feed_id IN"
                " (SELECT id FROM feeds WHERE user_id = ? AND folder_id = ? AND deleted_at IS NULL)",
                (uid, folder_id),
            ).fetchall()
            db.execute(
                "UPDATE feed_items SET is_read = 1 WHERE is_read = 0 AND feed_id IN"
                " (SELECT id FROM feeds WHERE user_id = ? AND folder_id = ? AND deleted_at IS NULL)",
                (uid, folder_id),
            )
        else:
            rows = db.execute(
                "SELECT id FROM feed_items WHERE is_read = 0 AND feed_id IN"
                " (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
                (uid,),
            ).fetchall()
            db.execute(
                "UPDATE feed_items SET is_read = 1 WHERE is_read = 0 AND feed_id IN"
                " (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
                (uid,),
            )
    _saved_search_counts_cache.invalidate(uid)
    ids = [r["id"] for r in rows]
    return {"success": True, "count": len(ids), "ids": ids}


@app.post("/api/reader/mark-unread")
def reader_mark_unread(request: Request, ids: str = Form("")):
    """Undo a mark-all-read: restore the given items to unread (issue #11).

    Accepts a comma-separated list of feed_items ids; each id is validated and
    the update is scoped to the caller's feeds.
    """
    uid = current_user_id(request)
    id_list = []
    for part in ids.split(","):
        val = _filter_int(part.strip())
        if val is not None:
            id_list.append(val)
    if not id_list:
        raise HTTPException(status_code=400, detail="No valid item ids")
    with get_db() as db:
        _require_reader(db, uid)
        placeholders = ", ".join("?" for _ in id_list)
        result = db.execute(
            f"UPDATE feed_items SET is_read = 0 WHERE id IN ({placeholders})"
            " AND feed_id IN (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
            (*id_list, uid),
        )
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "count": result.rowcount}


@app.post("/api/reader/mark-read")
def reader_mark_read(request: Request, ids: str = Form("")):
    """Bulk mark-as-read for a set of item ids (issue #11, auto-read-on-scroll).

    Idempotent: sets is_read = 1 (never toggles), scoped to the caller's feeds.
    Empty/invalid ids are a no-op rather than an error, since the auto-read
    batching can send an empty set.
    """
    uid = current_user_id(request)
    id_list = []
    for part in ids.split(","):
        val = _filter_int(part.strip())
        if val is not None:
            id_list.append(val)
    if not id_list:
        return {"success": True, "count": 0}
    with get_db() as db:
        _require_reader(db, uid)
        placeholders = ", ".join("?" for _ in id_list)
        result = db.execute(
            f"UPDATE feed_items SET is_read = 1 WHERE is_read = 0 AND id IN ({placeholders})"
            " AND feed_id IN (SELECT id FROM feeds WHERE user_id = ? AND deleted_at IS NULL)",
            (*id_list, uid),
        )
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "count": result.rowcount}


@app.get("/api/reader/{item_id}/body")
def reader_item_body(request: Request, item_id: int):
    """Return an item's display body for lazy loading (issue #11, Tier 3)."""
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT i.content_text, i.content, i.summary FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE i.id = ? AND f.user_id = ? AND f.deleted_at IS NULL",
            (item_id, uid),
        ).fetchone()
    if not row:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"content": row["content_text"] or row["content"] or row["summary"] or ""}


# ── Reader Mute Training (Phase 4) ──────────────────────────────────────────

@app.post("/api/reader/{item_id}/mute")
def reader_mute(
    request: Request,
    item_id: int,
    phrase: str = Form(...),
    scope: str = Form("feed"),
):
    """Append a phrase to mute_keywords for this feed or all read-enabled feeds."""
    uid = current_user_id(request)
    phrase = phrase.strip()
    if not phrase or len(phrase) > 60:
        raise HTTPException(status_code=400, detail="Phrase must be between 1 and 60 characters")
    if "," in phrase:
        raise HTTPException(status_code=400, detail="Phrase cannot contain commas")
    if scope not in ("feed", "all"):
        raise HTTPException(status_code=400, detail="Invalid scope (must be 'feed' or 'all')")

    with get_db() as db:
        _require_reader(db, uid)
        item = db.execute(
            "SELECT i.id, i.feed_id FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE i.id = ? AND f.user_id = ? AND f.deleted_at IS NULL",
            (item_id, uid),
        ).fetchone()
        if not item:
            raise HTTPException(status_code=404, detail="Item not found")

        if scope == "feed":
            target_feeds = db.execute(
                "SELECT id, mute_keywords FROM feeds WHERE id = ? AND user_id = ? AND deleted_at IS NULL",
                (item["feed_id"], uid),
            ).fetchall()
        else:
            target_feeds = db.execute(
                "SELECT id, mute_keywords FROM feeds WHERE user_id = ? AND read_enabled = 1 AND deleted_at IS NULL",
                (uid,),
            ).fetchall()

        affected_feed_ids = []
        for tf in target_feeds:
            existing = [k.strip() for k in (tf["mute_keywords"] or "").split(",") if k.strip()]
            if not any(k.casefold() == phrase.casefold() for k in existing):
                existing.append(phrase)
                new_kw_str = ", ".join(existing)
                db.execute(
                    "UPDATE feeds SET mute_keywords = ? WHERE id = ? AND user_id = ?",
                    (new_kw_str, tf["id"], uid),
                )
                affected_feed_ids.append(tf["id"])

    # Saved-search counts apply mute keywords, so a mute changes them
    _saved_search_counts_cache.invalidate(uid)
    return {
        "success": True,
        "phrase": phrase,
        "scope": scope,
        "affected_feed_ids": affected_feed_ids,
    }


@app.post("/api/reader/unmute")
def reader_unmute(
    request: Request,
    phrase: str = Form(...),
    feed_ids: str = Form(...),
):
    """Remove a previously muted phrase from specified feeds (undo support)."""
    uid = current_user_id(request)
    phrase = phrase.strip()
    if not phrase:
        raise HTTPException(status_code=400, detail="Phrase cannot be empty")

    fids = []
    for part in feed_ids.split(","):
        fid = _filter_int(part.strip())
        if fid is not None:
            fids.append(fid)

    if not fids:
        return {"success": True, "phrase": phrase, "count": 0}

    with get_db() as db:
        _require_reader(db, uid)
        placeholders = ", ".join("?" for _ in fids)
        target_feeds = db.execute(
            f"SELECT id, mute_keywords FROM feeds WHERE user_id = ? AND id IN ({placeholders}) AND deleted_at IS NULL",
            (uid, *fids),
        ).fetchall()

        count = 0
        for tf in target_feeds:
            existing = [k.strip() for k in (tf["mute_keywords"] or "").split(",") if k.strip()]
            filtered = [k for k in existing if k.casefold() != phrase.casefold()]
            if len(filtered) != len(existing):
                new_kw_str = ", ".join(filtered)
                db.execute(
                    "UPDATE feeds SET mute_keywords = ? WHERE id = ? AND user_id = ?",
                    (new_kw_str, tf["id"], uid),
                )
                count += 1

    # Saved-search counts apply mute keywords, so an unmute changes them
    _saved_search_counts_cache.invalidate(uid)
    return {"success": True, "phrase": phrase, "count": count}


@app.get("/api/reader/new-count")
def reader_new_count(request: Request, since_id: int = 0):
    """Count unread-capable items newer than since_id (poll-pill support)."""
    uid = current_user_id(request)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT COUNT(*) AS c FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE f.user_id = ? AND f.read_enabled = 1 AND f.deleted_at IS NULL AND i.id > ?",
            (uid, since_id),
        ).fetchone()
    return {"count": row["c"]}


@app.get("/api/reader/{item_id}/compose")
def reader_compose_preview(
    request: Request,
    item_id: int,
    template: str = "{{ title }} {{ link }}",
):
    """Return JSON preview state for an item in the Compose desk (Phase 2)."""
    uid = current_user_id(request)
    _validate_echo_template(template)
    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT i.*, f.name AS feed_name FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE i.id = ? AND f.user_id = ? AND f.deleted_at IS NULL",
            (item_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Item not found")

    item_dict = {
        "id": row["item_id"],
        "title": row["title"] or "",
        "link": row["link"] or "",
        "summary": row["summary"] or "",
        "content": row["content"] or "",
        "content_text": row["content_text"] or "",
        "content_link": row["content_link"] or "",
        "author": row["author"] or "",
        "date": timestamp_str(row["published_at"]) if row["published_at"] else "",
        "categories": categories_from_value(row["categories"]) if "categories" in row.keys() else {},
        "image_url": row["image_url"] or "",
        "image_alt": row["image_alt"] or "",
    }

    try:
        rendered_text = render_template(template, item_dict, feed_name=row["feed_name"] or "")
    except Exception as e:
        rendered_text = ""

    dests = _shout_destinations(uid)
    rendered_map = {}
    for d in dests:
        rendered_map[d["value"]] = rendered_text

    return {
        "item": {
            "id": row["id"],
            "title": row["title"] or "",
            "link": row["link"] or "",
            "summary": row["summary"] or "",
            "image_url": row["image_url"] or "",
            "image_alt": row["image_alt"] or "",
        },
        "destinations": dests,
        "default_template": "{{ title }} {{ link }}",
        "rendered": rendered_map,
    }


@app.post("/api/reader/{item_id}/compose")
def reader_compose(
    request: Request,
    item_id: int,
    destinations: str = Form(...),
    content: str | None = Form(None),
    template: str = Form("{{ title }} {{ link }}"),
    visibility: str = Form("public"),
    attach_image: str = Form("0"),
    image_alt: str | None = Form(None),
    enqueue: str = Form("0"),
    scheduled_at: str | None = Form(None),
):
    """Post an edited body or template to 1..5 destinations (Phase 2), or add to queue (Phase 3)."""
    uid = current_user_id(request)
    dest_keys = [d.strip() for d in destinations.split(",") if d.strip()]
    if not (1 <= len(dest_keys) <= 5):
        raise HTTPException(status_code=400, detail="Must select between 1 and 5 destinations")
    if visibility not in VALID_VISIBILITY:
        raise HTTPException(status_code=400, detail="Invalid visibility")

    has_content = content is not None and content.strip() != ""
    if content is not None and not has_content:
        # User provided an explicitly empty or whitespace-only content body
        raise HTTPException(status_code=400, detail="Post body cannot be empty")
    if not has_content:
        if not template or not template.strip():
            raise HTTPException(status_code=400, detail="Post body or template cannot be empty")
        _validate_echo_template(template)

    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    is_attach_image = 1 if attach_image in ("1", "true", "on") else 0
    is_enqueue = enqueue in ("1", "true", "on")

    with get_db() as db:
        _require_reader(db, uid)
        row = db.execute(
            "SELECT i.*, f.name AS feed_name FROM feed_items i"
            " JOIN feeds f ON i.feed_id = f.id"
            " WHERE i.id = ? AND f.user_id = ? AND f.deleted_at IS NULL",
            (item_id, uid),
        ).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Item not found")

        validated_dests = []
        for dk in dest_keys:
            dest_type, sep, raw_id = dk.partition(":")
            if not sep or dest_type not in VALID_DEST_TYPES:
                raise HTTPException(status_code=400, detail=f"Invalid destination: {dk}")
            destination_id = _filter_int(raw_id)
            if destination_id is None:
                raise HTTPException(status_code=400, detail=f"Invalid destination: {dk}")
            dest_table = DESTINATION_TABLES_BY_TYPE[dest_type]
            dest = db.execute(
                f"SELECT id FROM {dest_table} WHERE id = ? AND user_id = ?",
                (destination_id, uid),
            ).fetchone()
            if not dest:
                raise HTTPException(status_code=404, detail=f"Destination not found: {dk}")
            validated_dests.append((dest_type, destination_id, dk))

        if is_enqueue and settings.MULTI:
            plan = _user_plan(db, uid)
            current_q_count = db.execute(
                "SELECT COUNT(*) AS c FROM queued_posts WHERE user_id = ? AND status IN ('queued', 'sending')",
                (uid,),
            ).fetchone()["c"]
            needed = current_q_count + len(validated_dests) - 1
            try:
                plans.check_queue_allowance(needed, plan)
            except PlanError as e:
                raise HTTPException(status_code=402, detail=str(e))

    # Determine image_alt override
    if image_alt is not None:
        chosen_alt = image_alt.strip()
    else:
        chosen_alt = (row["image_alt"] or "").strip()

    item = {
        "id": row["item_id"],
        "title": row["title"] or "",
        "link": row["link"] or "",
        "summary": row["summary"] or "",
        "content": row["content"] or "",
        "content_text": row["content_text"] or "",
        "content_html": (row["content_html"] if "content_html" in row.keys() else "") or "",
        "content_link": row["content_link"] or "",
        "author": row["author"] or "",
        "date": timestamp_str(row["published_at"]) if row["published_at"] else "",
        "categories": categories_from_value(row["categories"]) if "categories" in row.keys() else {},
        "image_url": row["image_url"] or "",
        "image_alt": chosen_alt,
    }

    # Final materialized content string
    if has_content and content is not None:
        final_content = content
    else:
        try:
            final_content = render_template(template, item, feed_name=row["feed_name"] or "")
        except Exception:
            raise HTTPException(status_code=400, detail="Template rendering failed. Check your template syntax.")

    results = []

    if is_enqueue:
        # Enqueue for scheduled dispatch
        for dest_type, destination_id, dk in validated_dests:
            with get_db() as db:
                if scheduled_at and scheduled_at.strip():
                    norm_dt = as_utc_naive(scheduled_at.strip())
                    if not norm_dt:
                        raise HTTPException(status_code=400, detail="Invalid scheduled_at datetime")
                    # Cap scheduling horizon to 365 days out
                    max_dt = datetime.now(timezone.utc).replace(tzinfo=None) + timedelta(days=365)
                    if norm_dt > max_dt:
                        raise HTTPException(status_code=400, detail="Cannot schedule more than 365 days ahead")
                    target_time = norm_dt.strftime("%Y-%m-%d %H:%M:%S")
                else:
                    target_time = next_free_slot(db, uid, dest_type, destination_id)

                # Create the one-shot echo once at enqueue time so crash-recovery
                # retries reuse the same echo_id (idempotency key for _claim_post).
                now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
                echo_id = db.execute(
                    """
                    INSERT INTO echoes (feed_id, destination_type, destination_id, template,
                                       visibility, attach_image, enabled, one_shot, deleted_at, user_id)
                    VALUES (?, ?, ?, '{{ title }}', ?, ?, 0, 1, ?, ?) RETURNING id
                    """,
                    (row["feed_id"], dest_type, destination_id, visibility, is_attach_image, now_utc, uid),
                ).fetchone()["id"]

                qp_id = db.execute(
                    """
                    INSERT INTO queued_posts (
                        user_id, feed_item_id, item_id, feed_id, destination_type, destination_id,
                        content, visibility, attach_image, image_alt, scheduled_at, status, echo_id
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'queued', ?) RETURNING id
                    """,
                    (
                        uid,
                        row["id"],
                        row["item_id"],
                        row["feed_id"],
                        dest_type,
                        destination_id,
                        final_content,
                        visibility,
                        is_attach_image,
                        chosen_alt,
                        target_time,
                        echo_id,
                    ),
                ).fetchone()["id"]

            results.append({
                "destination": dk,
                "success": True,
                "status": "queued",
                "queue_id": qp_id,
                "scheduled_at": target_time,
            })
        return {
            "success": True,
            "queued": True,
            "results": results,
        }

    # Immediate dispatch (Post now)
    for dest_type, destination_id, dk in validated_dests:
        with get_db() as db:
            echo_id = db.execute(
                """INSERT INTO echoes (feed_id, destination_type, destination_id, template,
                                       visibility, attach_image, enabled, one_shot, deleted_at, user_id)
                   VALUES (?, ?, ?, ?, ?, ?, 0, 1, ?, ?) RETURNING id""",
                (row["feed_id"], dest_type, destination_id, template, visibility, is_attach_image, now, uid),
            ).fetchone()["id"]
            echo = db.execute("SELECT * FROM echoes WHERE id = ?", (echo_id,)).fetchone()

        ok = process_echo(
            echo,
            item,
            feed_name=row["feed_name"] or "",
            override_content=final_content,
        )

        with get_db() as db:
            pi = db.execute(
                "SELECT status, post_url, error_message FROM posted_items"
                " WHERE echo_id = ? AND item_id = ?",
                (echo_id, row["item_id"]),
            ).fetchone()

        is_success = bool(pi and pi["status"] == "success")
        results.append({
            "destination": dk,
            "success": is_success,
            "status": pi["status"] if pi else "unknown",
            "post_url": pi["post_url"] if pi else None,
            "error_message": pi["error_message"] if pi else None,
        })

    all_success = all(r["success"] for r in results)
    return {
        "success": all_success,
        "results": results,
    }


@app.post("/api/reader/{item_id}/shout")
def reader_shout(
    request: Request,
    item_id: int,
    destination: str = Form(...),
    template: str = Form("{{ title }} {{ link }}"),
    visibility: str = Form("public"),
):
    """Shout alias for Compose (Phase 2 backwards compatibility)."""
    res = reader_compose(
        request=request,
        item_id=item_id,
        destinations=destination,
        content=None,
        template=template,
        visibility=visibility,
        attach_image="0",
        image_alt=None,
    )
    first_res = res["results"][0] if res.get("results") else None
    return {
        "success": res["success"],
        "status": first_res["status"] if first_res else "unknown",
        "post_url": first_res["post_url"] if first_res else None,
        "error_message": first_res["error_message"] if first_res else None,
    }


# ── API: Echoes ─────────────────────────────────────────────────────────────

VALID_VISIBILITY = {"public", "unlisted", "private", "direct"}
VALID_DEST_TYPES = set(DESTINATION_TABLES_BY_TYPE)
VALID_FILTER_MODES = {"exclude", "include"}
# Delivery modes accepted at the echo routes. queue/review bridge scheduler
# echoes into the curation queue (scheduler enforces the plan cap at enqueue,
# not here); digest stays email-only.
VALID_DELIVERY_MODES = ("instant", "digest", "queue", "review")


def _revalidate_stored_templates() -> None:
    """Log stored templates that no longer parse under the Jinja2 engine.

    Templates saved by the old regex engine could contain literal ``{%``
    or ``{#`` text that is now real Jinja2 syntax. Save-time validation
    only protects new edits, so revalidate everything at startup and
    warn instead of letting those echoes silently gave_up at render time.
    """
    with get_db() as db:
        rows = db.execute(
            "SELECT id, feed_id, template FROM echoes WHERE deleted_at IS NULL"
        ).fetchall()
    for row in rows:
        try:
            validate_template(row["template"])
        except Exception as e:
            logger.warning(
                "Echo %s: stored template no longer parses and will give up at "
                "render time (edit the echo to fix): %s",
                row["id"],
                e,
            )


def _validate_echo_template(template: str) -> None:
    """Reject templates the Jinja2 engine cannot parse at save time.

    Without this, a syntax error only surfaces later as a gave_up post.
    """
    try:
        validate_template(template)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Template error: {e}")


def _github_echo_fields(destination_type: str, github_path: str, github_commit: str) -> tuple[str, str]:
    """Resolve the per-echo github_path / github_commit columns for a save.

    Returns the two column values. A path is REQUIRED for github echoes and
    is validated by rendering the deliberately-nasty SAMPLE_ITEM through it,
    so a template that would permanently fail on the first real item (bad
    syntax, or a path traversal / over-limit segment) is refused AT SAVE
    TIME with a pointer to the ``| slug`` filter. The commit message is
    optional (blank = the default ``Add {{ title }}``).

    Non-github echoes always store '' for both: a stale path must never
    ride along when the destination type changes (the hidden-input leak).
    """
    if destination_type != "github":
        return "", ""
    path = (github_path or "").strip()
    if not path:
        raise HTTPException(
            status_code=400,
            detail="GitHub echoes need a file path template, e.g. content/blog/{{ title | slug }}.md",
        )
    try:
        github.render_path(path, github.SAMPLE_ITEM)
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail=f"GitHub path template is invalid: {e}"
            " (use {{ title | slug }} for filename-safe titles)",
        )
    commit = (github_commit or "").strip()
    if commit:
        try:
            # build_commit_message deliberately swallows render errors (a
            # drifted stored template must never fail a delivery), so it
            # cannot validate a new template. Render directly instead.
            render_template(commit, github.SAMPLE_ITEM)
        except Exception as e:
            raise HTTPException(
                status_code=400,
                detail=f"GitHub commit message template is invalid: {e}",
            )
    return path, commit


@app.post("/api/echoes")
async def add_echo(
    request: Request,
    feed_id: int = Form(...),
    destination_type: str = Form("mastodon"),
    account_id: int = Form(None),
    email_account_id: int = Form(None),
    bluesky_account_id: int = Form(None),
    microblog_account_id: int = Form(None),
    matrix_account_id: int = Form(None),
    discord_account_id: int = Form(None),
    telegram_account_id: int = Form(None),
    webhook_account_id: int = Form(None),
    github_account_id: int = Form(None),
    github_path: str = Form(""),
    github_commit: str = Form(""),
    template: str = Form("{{ title }} {{ link }}"),
    visibility: str = Form("public"),
    filter_keywords: str = Form(""),
    filter_mode: str = Form("exclude"),
    content_warning: str = Form(""),
    attach_image: str = Form(""),
    render_html: str = Form(""),
    use_markdown: str = Form(""),
    delivery_mode: str = Form("instant"),
    drip_limit: int = Form(0),
    enabled: str = Form(""),
    return_to: str = Form("/echoes"),
):
    uid = current_user_id(request)
    if destination_type not in VALID_DEST_TYPES:
        raise HTTPException(status_code=400, detail=f"Invalid destination type")
    if visibility not in VALID_VISIBILITY:
        raise HTTPException(status_code=400, detail=f"Invalid visibility")
    if filter_mode not in VALID_FILTER_MODES:
        raise HTTPException(status_code=400, detail=f"Invalid filter mode")
    if delivery_mode not in VALID_DELIVERY_MODES:
        raise HTTPException(status_code=400, detail="Invalid delivery mode")
    # GitHub echoes clamp silently instead of erroring: digest is email-only
    # (github delivery is per-item file writes, so digest collapses to
    # instant) and file bodies never carry feed images.
    if destination_type == "github":
        if delivery_mode == "digest":
            delivery_mode = "instant"
        attach_image = ""
    # Digest mode is email-only
    if delivery_mode == "digest" and destination_type != "email":
        raise HTTPException(status_code=400, detail="Digest mode is only available for email destinations")
    if drip_limit < 0 or drip_limit > 1000:
        raise HTTPException(status_code=400, detail="Drip limit must be between 0 and 1000")
    if settings.MULTI:
        with get_db() as db:
            drip_limit = plans.clamp_drip_limit(drip_limit, _user_plan(db, current_user_id(request)))

    # render_html applies only to instant email/Matrix echoes: the rendered
    # template (which may embed ingest-sanitized {{ content_html }}) becomes
    # the email HTML part or the Matrix formatted_body. Digests stay
    # text-only and other destinations speak plain text, so the flag clamps
    # off there rather than erroring.
    is_render_html = (
        1 if render_html in ("1", "true", "on")
        and destination_type in ("email", "matrix")
        and delivery_mode == "instant"
        else 0
    )
    # use_markdown applies only to Mastodon destinations (GoToSocial parses
    # content_type=text/markdown; stock Mastodon ignores the field).
    is_use_markdown = (
        1 if use_markdown in ("1", "true", "on")
        and destination_type == "mastodon"
        else 0
    )

    _validate_echo_template(template)

    # Resolve destination_id based on type
    if destination_type == "mastodon":
        destination_id = account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="account_id required for mastodon destination")
    elif destination_type == "email":
        destination_id = email_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="email_account_id required for email destination")
    elif destination_type == "bluesky":
        destination_id = bluesky_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="bluesky_account_id required for bluesky destination")
    elif destination_type == "microblog":
        destination_id = microblog_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="microblog_account_id required for microblog destination")
    elif destination_type == "matrix":
        destination_id = matrix_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="matrix_account_id required for matrix destination")
    elif destination_type == "discord":
        destination_id = discord_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="discord_account_id required for discord destination")
    elif destination_type == "telegram":
        destination_id = telegram_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="telegram_account_id required for telegram destination")
    elif destination_type == "webhook":
        destination_id = webhook_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="webhook_account_id required for webhook destination")
    elif destination_type == "github":
        destination_id = github_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="github_account_id required for github destination")
    else:
        # Unreachable (VALID_DEST_TYPES is checked above), kept explicit so a
        # future destination type cannot silently fall through.
        raise HTTPException(status_code=400, detail="Invalid destination type")

    github_path_value, github_commit_value = _github_echo_fields(
        destination_type, github_path, github_commit
    )

    is_enabled = 1 if enabled else 0
    is_attach_image = 1 if attach_image else 0
    with get_db() as db:
        # Ownership: the feed and the destination must belong to this user.
        feed = db.execute(
            "SELECT id FROM feeds WHERE id = ? AND user_id = ? AND deleted_at IS NULL",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        # No destination cap here: connecting accounts is capped at
        # connect time (_check_destination_cap); editing an existing echo
        # only ever references destinations that already exist.
        dest_table = DESTINATION_TABLES_BY_TYPE[destination_type]
        dest = db.execute(
            f"SELECT id FROM {dest_table} WHERE id = ? AND user_id = ?",
            (destination_id, uid),
        ).fetchone()
        if not dest:
            raise HTTPException(status_code=404, detail="Destination not found")

        db.execute(
            """INSERT INTO echoes (feed_id, destination_type, destination_id, template, visibility,
                                   filter_keywords, filter_mode, content_warning, attach_image,
                                   render_html, use_markdown, delivery_mode, drip_limit, enabled,
                                   github_path, github_commit, user_id)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (feed_id, destination_type, destination_id, template, visibility,
             filter_keywords.strip(), filter_mode, content_warning.strip(), is_attach_image,
             is_render_html, is_use_markdown, delivery_mode, drip_limit, is_enabled,
             github_path_value, github_commit_value, uid),
        )
    # Return to the originating surface. Allowlist rather than prefix checks:
    # the WHATWG URL parser treats backslashes as slashes, so a prefix check
    # that accepts "/\evil.com" would be an open redirect.
    if return_to not in ("/echoes", "/reader"):
        return_to = "/echoes"
    return RedirectResponse(url=return_to, status_code=303)


@app.post("/api/echoes/{echo_id}/toggle")
async def toggle_echo(request: Request, echo_id: int):
    uid = current_user_id(request)
    with get_db() as db:
        echo = db.execute(
            "SELECT enabled FROM echoes WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (echo_id, uid),
        ).fetchone()
        if not echo:
            raise HTTPException(status_code=404, detail="Echo not found")
        new_val = 0 if echo["enabled"] else 1
        db.execute(
            "UPDATE echoes SET enabled = ? WHERE id = ? AND user_id = ?",
            (new_val, echo_id, uid),
        )
    return {"success": True, "enabled": bool(new_val)}


# The echo row the booster route reads, shared by the initial check and the
# post-race re-read so both paths explain refusals with the SAME predicates
# (mirrors the _HISTORY_*_SQL constants pattern).
_ECHO_BOOSTER_SELECT = """
    SELECT e.destination_type, a.booster_enabled AS account_boost_enabled
    FROM echoes e
    LEFT JOIN accounts a
      ON e.destination_type = 'mastodon' AND e.destination_id = a.id AND a.user_id = e.user_id
    WHERE e.id = ? AND e.deleted_at IS NULL AND e.user_id = ?
"""


@app.post("/api/echoes/{echo_id}/booster")
async def toggle_echo_booster(request: Request, echo_id: int):
    """Toggle FeedBooster for one echo: this echo's posts to its Mastodon
    destination get announced by @feedbooster@feedecho.net. Refused when the
    destination account itself has Boost on — the account setting already
    boosts every echo to it, so the two cannot both be newly enabled."""
    uid = current_user_id(request)
    if not (settings.BOOSTER_URL and settings.BOOSTER_TOKEN):
        return _render_echoes_error(request, "FeedBooster is not configured on this server.")

    def _refusal_for(row):
        """Refusal message for an echo row, or None when a flip may proceed."""
        if row["destination_type"] != "mastodon":
            return "Boost is only available for echoes to Mastodon destinations."
        if row["account_boost_enabled"]:
            return (
                "Boost is already enabled for this Mastodon account, so every echo to it is"
                " boosted. Turn off the account setting on the Accounts page to control"
                " individual echoes."
            )
        return None

    refusal = None
    with get_db() as db:
        row = db.execute(_ECHO_BOOSTER_SELECT, (echo_id, uid)).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="Echo not found")
        refusal = _refusal_for(row)
        if refusal is None:
            # Atomic flip, conditional on the account flag staying off: a
            # concurrent account-level enable cannot interleave between the
            # check above and this write, and an echo soft-deleted in that
            # window is never touched.
            flipped = db.execute(
                """
                UPDATE echoes SET booster_enabled = 1 - booster_enabled
                WHERE id = ? AND user_id = ? AND destination_type = 'mastodon'
                  AND deleted_at IS NULL
                  AND COALESCE((
                        SELECT a.booster_enabled FROM accounts a
                        WHERE a.id = echoes.destination_id AND a.user_id = echoes.user_id
                      ), 0) = 0
                """,
                (echo_id, uid),
            )
            if flipped.rowcount == 0:
                # Race window: re-read and explain with the CURRENT state.
                row2 = db.execute(_ECHO_BOOSTER_SELECT, (echo_id, uid)).fetchone()
                if not row2:
                    raise HTTPException(status_code=404, detail="Echo not found")
                refusal = _refusal_for(row2) or (
                    "Could not update the Boost setting for this echo. Try again."
                )
    # Rendered OUTSIDE the connection block: _render_echoes_error opens its
    # own connection, and re-rendering the page while still holding this
    # request's connection would tie up two per refusal.
    if refusal:
        return _render_echoes_error(request, refusal)
    return RedirectResponse(url="/echoes", status_code=303)


@app.post("/api/echoes/{echo_id}/edit")
async def edit_echo(
    request: Request,
    echo_id: int,
    feed_id: int = Form(...),
    destination_type: str = Form("mastodon"),
    account_id: int = Form(None),
    email_account_id: int = Form(None),
    bluesky_account_id: int = Form(None),
    microblog_account_id: int = Form(None),
    matrix_account_id: int = Form(None),
    discord_account_id: int = Form(None),
    telegram_account_id: int = Form(None),
    webhook_account_id: int = Form(None),
    github_account_id: int = Form(None),
    github_path: str = Form(""),
    github_commit: str = Form(""),
    template: str = Form("{{ title }} {{ link }}"),
    visibility: str = Form("public"),
    filter_keywords: str = Form(""),
    filter_mode: str = Form("exclude"),
    content_warning: str = Form(""),
    attach_image: str = Form(""),
    render_html: str = Form(""),
    use_markdown: str = Form(""),
    delivery_mode: str = Form("instant"),
    drip_limit: int = Form(0),
    enabled: str = Form(""),
):
    if destination_type not in VALID_DEST_TYPES:
        raise HTTPException(status_code=400, detail=f"Invalid destination type")
    if visibility not in VALID_VISIBILITY:
        raise HTTPException(status_code=400, detail=f"Invalid visibility")
    if filter_mode not in VALID_FILTER_MODES:
        raise HTTPException(status_code=400, detail=f"Invalid filter mode")
    if delivery_mode not in VALID_DELIVERY_MODES:
        raise HTTPException(status_code=400, detail="Invalid delivery mode")
    # GitHub echoes clamp silently instead of erroring: digest is email-only
    # (github delivery is per-item file writes, so digest collapses to
    # instant) and file bodies never carry feed images.
    if destination_type == "github":
        if delivery_mode == "digest":
            delivery_mode = "instant"
        attach_image = ""
    if delivery_mode == "digest" and destination_type != "email":
        raise HTTPException(status_code=400, detail="Digest mode is only available for email destinations")
    if drip_limit < 0 or drip_limit > 1000:
        raise HTTPException(status_code=400, detail="Drip limit must be between 0 and 1000")
    if settings.MULTI:
        with get_db() as db:
            drip_limit = plans.clamp_drip_limit(drip_limit, _user_plan(db, current_user_id(request)))

    # render_html applies only to instant email/Matrix echoes: the rendered
    # template (which may embed ingest-sanitized {{ content_html }}) becomes
    # the email HTML part or the Matrix formatted_body. Digests stay
    # text-only and other destinations speak plain text, so the flag clamps
    # off there rather than erroring.
    is_render_html = (
        1 if render_html in ("1", "true", "on")
        and destination_type in ("email", "matrix")
        and delivery_mode == "instant"
        else 0
    )
    # use_markdown applies only to Mastodon destinations (GoToSocial parses
    # content_type=text/markdown; stock Mastodon ignores the field).
    is_use_markdown = (
        1 if use_markdown in ("1", "true", "on")
        and destination_type == "mastodon"
        else 0
    )

    _validate_echo_template(template)

    if destination_type == "mastodon":
        destination_id = account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="account_id required for mastodon destination")
    elif destination_type == "email":
        destination_id = email_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="email_account_id required for email destination")
    elif destination_type == "bluesky":
        destination_id = bluesky_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="bluesky_account_id required for bluesky destination")
    elif destination_type == "microblog":
        destination_id = microblog_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="microblog_account_id required for microblog destination")
    elif destination_type == "matrix":
        destination_id = matrix_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="matrix_account_id required for matrix destination")
    elif destination_type == "discord":
        destination_id = discord_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="discord_account_id required for discord destination")
    elif destination_type == "telegram":
        destination_id = telegram_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="telegram_account_id required for telegram destination")
    elif destination_type == "webhook":
        destination_id = webhook_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="webhook_account_id required for webhook destination")
    elif destination_type == "github":
        destination_id = github_account_id
        if not destination_id:
            raise HTTPException(status_code=400, detail="github_account_id required for github destination")
    else:
        # Unreachable (VALID_DEST_TYPES is checked above), kept explicit so a
        # future destination type cannot silently fall through.
        raise HTTPException(status_code=400, detail="Invalid destination type")

    github_path_value, github_commit_value = _github_echo_fields(
        destination_type, github_path, github_commit
    )

    is_enabled = 1 if enabled else 0
    is_attach_image = 1 if attach_image else 0
    uid = current_user_id(request)
    with get_db() as db:
        echo = db.execute(
            "SELECT * FROM echoes WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (echo_id, uid),
        ).fetchone()
        if not echo:
            raise HTTPException(status_code=404, detail="Echo not found")
        # Ownership: the new feed and destination must belong to this user.
        feed = db.execute(
            "SELECT id FROM feeds WHERE id = ? AND user_id = ? AND deleted_at IS NULL",
            (feed_id, uid),
        ).fetchone()
        if not feed:
            raise HTTPException(status_code=404, detail="Feed not found")
        # No destination cap here: connecting accounts is capped at
        # connect time (_check_destination_cap); editing an existing echo
        # only ever references destinations that already exist.
        dest_table = DESTINATION_TABLES_BY_TYPE[destination_type]
        dest = db.execute(
            f"SELECT id FROM {dest_table} WHERE id = ? AND user_id = ?",
            (destination_id, uid),
        ).fetchone()
        if not dest:
            raise HTTPException(status_code=404, detail="Destination not found")
        db.execute(
            """UPDATE echoes SET feed_id = ?, destination_type = ?, destination_id = ?,
               template = ?, visibility = ?, filter_keywords = ?, filter_mode = ?,
               content_warning = ?, attach_image = ?, render_html = ?, use_markdown = ?,
               delivery_mode = ?, drip_limit = ?,
               github_path = ?, github_commit = ?,
               enabled = ?
               WHERE id = ? AND user_id = ?""",
            (feed_id, destination_type, destination_id, template, visibility,
             filter_keywords.strip(), filter_mode, content_warning.strip(), is_attach_image,
             is_render_html, is_use_markdown, delivery_mode, drip_limit,
             github_path_value, github_commit_value,
             is_enabled, echo_id, uid),
        )
        if (
            echo["booster_enabled"]
            and (echo["destination_type"], echo["destination_id"]) != (destination_type, destination_id)
        ):
            # Per-echo boost does not survive a destination change: a stale
            # flag must not silently attach to a different destination (and
            # on a non-Mastodon or account-boosted target it could never be
            # turned off through the UI, which hides the toggle there).
            db.execute(
                "UPDATE echoes SET booster_enabled = 0 WHERE id = ? AND user_id = ?",
                (echo_id, uid),
            )
    return RedirectResponse(url="/echoes", status_code=303)


@app.post("/api/echoes/{echo_id}/delete")
async def delete_echo(request: Request, echo_id: int):
    """Soft-delete an echo so its posted-item history survives.

    A hard DELETE cascades to posted_items/digest_items and erases the
    cross-post audit trail. Marking deleted_at and disabling the echo removes
    it from listings and stops delivery while keeping its history intact.
    """
    uid = current_user_id(request)
    with get_db() as db:
        db.execute(
            """
            UPDATE echoes
               SET deleted_at = CURRENT_TIMESTAMP,
                   enabled = 0
             WHERE id = ?
               AND deleted_at IS NULL
               AND user_id = ?
            """,
            (echo_id, uid),
        )
    return RedirectResponse(url="/echoes", status_code=303)


# ── API: Preview ────────────────────────────────────────────────────────────

# Per-IP throttle on the template preview endpoint (hosted mode): the preview
# fetches a live feed on every call, so it is both an outbound-egress hammer
# and a way to hammer the target feed. Gated on MULTI — single-mode operators
# preview their own feeds.
_preview_throttle_lock = threading.Lock()
_preview_throttle_attempts: dict[str, list[float]] = {}
_PREVIEW_THROTTLE_LIMIT = 20
_PREVIEW_THROTTLE_WINDOW = 600  # seconds
_PREVIEW_THROTTLE_MAX_IPS = 10_000
_preview_last_sweep = 0.0


def _preview_throttled(ip: str) -> bool:
    """Atomically rate-limit one preview attempt (check-and-record)."""
    global _preview_last_sweep
    now = time.monotonic()
    with _preview_throttle_lock:
        # Global sweep (at most once per window) prunes stale buckets, so a
        # rotating-IP spray cannot permanently exhaust the tracked-IP ceiling
        # with expired entries and 429 every new legitimate user.
        if now - _preview_last_sweep >= _PREVIEW_THROTTLE_WINDOW:
            for key in list(_preview_throttle_attempts):
                recent = [
                    t for t in _preview_throttle_attempts[key]
                    if now - t < _PREVIEW_THROTTLE_WINDOW
                ]
                if recent:
                    _preview_throttle_attempts[key] = recent
                else:
                    _preview_throttle_attempts.pop(key, None)
            _preview_last_sweep = now
        recent = [
            t for t in _preview_throttle_attempts.get(ip, [])
            if now - t < _PREVIEW_THROTTLE_WINDOW
        ]
        if recent:
            _preview_throttle_attempts[ip] = recent
        else:
            _preview_throttle_attempts.pop(ip, None)
        if len(recent) >= _PREVIEW_THROTTLE_LIMIT:
            return True
        if (
            len(_preview_throttle_attempts) >= _PREVIEW_THROTTLE_MAX_IPS
            and ip not in _preview_throttle_attempts
        ):
            return True
        _preview_throttle_attempts.setdefault(ip, []).append(now)
        return False


@app.post("/api/preview")
def preview_template(
    request: Request,
    template: str = Form(...),
    feed_id: int = Form(...),
):
    """Preview a template rendered against the feed's most recent items.

    Renders the given template against up to 3 recent feed items so users
    can check output before saving an echo. The template is validated
    first; syntax errors return 400 with the Jinja2 error message.
    """
    if settings.MULTI and _preview_throttled(auth._client_ip(request)):
        return JSONResponse(
            {"success": False, "error": "Too many previews. Please wait before trying again."},
            status_code=429,
        )
    try:
        validate_template(template)
    except Exception as e:
        return JSONResponse(
            {"success": False, "error": f"Template syntax error: {e}"},
            status_code=400,
        )

    uid = current_user_id(request)
    with get_db() as db:
        feed = db.execute(
            "SELECT * FROM feeds WHERE id = ? AND deleted_at IS NULL AND user_id = ?",
            (feed_id, uid),
        ).fetchone()
    if not feed:
        raise HTTPException(status_code=404, detail="Feed not found")

    try:
        feed_data = fetch_feed(feed["url"])
    except SSRFError as e:
        return JSONResponse({"success": False, "error": str(e)}, status_code=400)
    except Exception:
        logger.warning("Template preview: feed %s fetch failed", feed_id, exc_info=True)
        return JSONResponse(
            {"success": False, "error": "Could not fetch the feed"}, status_code=502
        )

    items = (feed_data.get("items") or [])[:3]
    previews = []
    for item in items:
        try:
            rendered = render_template(template, item, feed_name=feed["name"])
        except Exception as e:
            return JSONResponse(
                {"success": False, "error": f"Render error: {e}"}, status_code=400
            )
        previews.append(
            {"title": item.get("title") or "(untitled)", "rendered": rendered}
        )

    return {"success": True, "items": previews}


# ── API: OAuth ───────────────────────────────────────────────────────────────

def _display_instance(instance: str) -> str:
    """Sanitize an instance string for display in a user-facing banner.

    The raw input can carry embedded credentials (user:password@host);
    echoing it verbatim would print the password back into the page.
    Show the URL without userinfo when it parses, else a fixed label.
    Review-gate finding (Gemini 3.8 Flash, 2026-09-10).
    """
    try:
        parsed = urlsplit(instance)
        if parsed.username or parsed.password:
            host = parsed.hostname or ""
            port = f":{parsed.port}" if parsed.port else ""
            return f"{parsed.scheme}://{host}{port}"
    except ValueError:
        pass
    return instance


@app.get("/oauth/connect")
def oauth_connect(request: Request, instance: str = ""):
    """Start a session-bound Mastodon OAuth authorization flow."""
    if not instance:
        raise HTTPException(status_code=400, detail="Instance URL is required")

    # Form input ("social.lol") is common; default the scheme to https
    # instead of bouncing the user to a raw 400 JSON page. Only prefix when
    # no scheme is present — a URL that already carries one must pass
    # through validation unchanged.
    instance = instance.strip().rstrip("/")
    if "://" not in instance:
        instance = f"https://{instance}"

    try:
        instance = validate_url(instance)
    except HTTPException:
        return _render_accounts_error(
            request,
            f"{_display_instance(instance)} isn't a usable Mastodon instance. "
            "Check the URL (for example https://dmv.community) and try again.",
        )

    # OAuth authorization and token endpoints must be TLS (RFC 6749 3.1.3
    # requires it for the user's credentials to survive the browser trip).
    # http:// here would also be killed by the page's own CSP form-action
    # ('self' https:) a hop later — reject it up front with a readable
    # message instead.
    if instance.startswith("http://"):
        return _render_accounts_error(
            request,
            f"{_display_instance(instance)} uses http://, but Mastodon "
            "instances must be reachable over https. Check the address.",
        )

    # Reduce to scheme + host so the authorize URL is always
    # {origin}/oauth/authorize and oauth_apps caches on a stable origin —
    # stray paths (/web, /@user) or query strings would otherwise corrupt
    # both (review-gate finding, 2026-09-10).
    parsed = urlparse(instance)
    instance = f"{parsed.scheme}://{parsed.netloc}"

    # This cookie is independent from shared-secret auth. It ties the OAuth
    # callback to the browser session that initiated the flow.
    oauth_session = request.cookies.get(OAUTH_SESSION_COOKIE)
    if not oauth_session:
        oauth_session = _secrets.token_urlsafe(32)

    try:
        auth_url = get_authorize_url(
            instance, oauth_session, user_id=current_user_id(request)
        )
    except Exception:
        logger.exception("OAuth connect failed for %s", instance)
        return _render_accounts_error(
            request,
            "Failed to start OAuth authorization. Verify the instance URL and try again.",
        )

    response = RedirectResponse(url=auth_url, status_code=302)
    response.set_cookie(
        key=OAUTH_SESSION_COOKIE,
        value=oauth_session,
        max_age=OAUTH_SESSION_MAX_AGE,
        httponly=True,
        # FORCE_SECURE_COOKIE exists because the scheme reads http behind
        # Caddy; the session cookie honours it and so must this one.
        secure=request.url.scheme == "https" or settings.FORCE_SECURE_COOKIE,
        samesite="lax",
        path="/oauth",
    )
    return response


@app.get("/oauth/callback")
def oauth_callback(
    request: Request,
    code: str = "",
    state: str = "",
    error: str = "",
):
    """Handle a one-time, session-bound Mastodon OAuth callback."""
    if error:
        # Not _render_accounts_error: /oauth/callback is auth-exempt, so
        # request.state.user_id is unset and current_user_id() would raise
        # 401 in multi mode — on exactly the paths that exist to report a
        # failure to the user.
        return _render_oauth_error(
            request, "Authorization was denied by the OAuth provider."
        )

    if not code or not state:
        raise HTTPException(status_code=400, detail="Missing code or state parameter")

    oauth_session = request.cookies.get(OAUTH_SESSION_COOKIE)
    if not oauth_session:
        raise HTTPException(
            status_code=400,
            detail="OAuth session is missing or expired. Start the connection again.",
        )

    try:
        instance, state_user_id = verify_state(state, oauth_session)
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail="Invalid, expired, already-used, or session-mismatched OAuth state.",
        )

    # In multi mode the state must be bound to a user. A NULL user_id
    # (legacy row) would otherwise attribute the account to tenant 1.
    if settings.MULTI and state_user_id is None:
        raise HTTPException(
            status_code=400,
            detail="OAuth session is not bound to a user. Start the connection again.",
        )

    try:
        token_data = exchange_code(instance, code)
    except Exception:
        logger.exception("OAuth token exchange failed for %s", instance)
        response = _render_oauth_error(
            request, "OAuth token exchange failed. Please try connecting again."
        )
        response.delete_cookie(OAUTH_SESSION_COOKIE, path="/oauth")
        return response

    access_token = token_data.get("access_token")
    if not isinstance(access_token, str) or not access_token:
        raise HTTPException(status_code=502, detail="OAuth provider returned no access token")

    try:
        creds = verify_credentials(instance, access_token)
        display_name = creds.get("display_name") or creds.get("username", "Unknown")
        username = creds.get("username", "unknown")
    except Exception:
        logger.exception("Could not verify OAuth credentials for %s", instance)
        display_name = "Unknown"
        username = "unknown"

    with get_db() as db:
        # Cap NEW rows only: reconnecting an existing account is an upsert
        # that changes nothing the cap measures.
        existing = db.execute(
            "SELECT id FROM accounts"
            " WHERE user_id = ? AND instance = ? AND username = ?",
            (state_user_id or 1, instance, username),
        ).fetchone()
        if existing is None:
            try:
                _check_destination_cap(db, state_user_id or 1)
            except PlanError as e:
                return _render_oauth_error(request, str(e))
        db.execute(
            """
            INSERT INTO accounts (name, username, instance, access_token, user_id)
            VALUES (?, ?, ?, ?, ?)
            ON CONFLICT(user_id, instance, username) DO UPDATE SET
                name = excluded.name,
                access_token = excluded.access_token
            """,
            (display_name, username, instance, security.encrypt_secret(access_token), state_user_id or 1),
        )

    response = RedirectResponse(url="/accounts?status=connected", status_code=303)
    response.delete_cookie(OAUTH_SESSION_COOKIE, path="/oauth")
    return response


# ── Misc ─────────────────────────────────────────────────────────────────────

@app.exception_handler(404)
async def not_found_handler(request: Request, exc: HTTPException):
    if request.headers.get("accept", "").startswith("application/json"):
        return JSONResponse({"detail": "Not found"}, status_code=404)
    return render("404.html", request, status_code=404)


@app.exception_handler(403)
async def forbidden_handler(request: Request, exc: HTTPException):
    message = exc.detail if isinstance(exc.detail, str) else "Forbidden"
    if request.headers.get("accept", "").startswith("application/json"):
        return JSONResponse({"detail": message}, status_code=403)
    return render("error.html", request, status_code=403, code=403, message=message)


@app.exception_handler(Exception)
async def unhandled_exception_handler(request: Request, exc: Exception):
    # RequestIdMiddleware (above) already logs this exception with full
    # request context (method, path, duration, peer, user id) before
    # re-raising, and its own traceback capture is what actually records
    # this for debugging — this handler only shapes the response so an
    # unexpected bug gets the same branded page as every other error
    # category instead of Starlette's bare default. Registering a handler
    # for the bare Exception type does not change how HTTPException (400/
    # 402/403/404/409/502/etc.) is handled: Starlette resolves handlers by
    # walking the raised exception's MRO, and HTTPException's own handler
    # (built-in, or the 403/404 overrides above) is more specific and is
    # matched first — this handler only ever receives exceptions that
    # aren't an HTTPException of any kind. Error-handling audit finding 1.1
    # (docs/reviews/2026-09-08-error-handling-audit.md).
    if request.headers.get("accept", "").startswith("application/json"):
        return JSONResponse({"detail": "Internal server error"}, status_code=500)
    return render("error.html", request, status_code=500,
                  code=500, message="Something went wrong. Please try again.")


@app.get("/favicon.svg", include_in_schema=False)
async def favicon():
    # Served from static/img/logo.svg via FileResponse so the file stays the
    # single source of the brand mark (also mounted at /static/img/logo.svg).
    from fastapi.responses import FileResponse

    return FileResponse(
        BASE_DIR / "static" / "img" / "logo.svg",
        media_type="image/svg+xml",
        headers={"Cache-Control": "public, max-age=86400"},
    )


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8453, access_log=False)