"""Background scheduler and concurrency-safe feed delivery pipeline."""

from __future__ import annotations

import base64
import json
import logging
import secrets
import threading
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse

import httpx
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger

import settings
import plans
from database import get_db, timestamp_str, prune_feed_items, as_utc_naive
from email_sender import send_email
from security import decrypt_secret, encrypt_secret
from feed_parser import (
    fetch_feed,
    fetch_image,
    fetch_page_metadata,
    get_new_items,
    get_backdated_items,
    _parse_item_date,
    html_to_text,
    sanitize_html,
    truncate,
)
from filters import is_filtered, match_reason
from mastodon import (
    MastodonAuthError,
    MastodonError,
    idempotency_key as mastodon_idempotency_key,
    post_status,
    upload_media,
)
from bluesky import (
    BLUESKY_IMAGE_TYPES,
    MAX_BLOB_BYTES,
    MAX_IMAGES as BLUESKY_MAX_IMAGES,
    MAX_POST_BYTES,
    MAX_POST_GRAPHEMES,
    BlueskyAccountGoneError,
    BlueskyAuthError,
    BlueskyError,
    build_external_embed,
    build_facets,
    build_image_embed,
    create_post,
    create_session,
    EXTERNAL_THUMB_MAX_BYTES,
    first_link_uri,
    refresh_session,
    resolve_pds,
    session_expiry,
    truncate_graphemes,
    upload_blob,
)
from microblog import (
    MicroblogAuthError,
    MicroblogError,
    create_post as microblog_create_post,
)
from matrix import (
    MATRIX_IMAGE_TYPES,
    MAX_UPLOAD_BYTES as MATRIX_MAX_UPLOAD_BYTES,
    MatrixAuthError,
    MatrixError,
    MatrixPermissionError,
    permalink as matrix_permalink,
    send_image as matrix_send_image,
    send_message as matrix_send_message,
    transaction_id as matrix_transaction_id,
    upload_media as matrix_upload_media,
)
from discord import (
    DiscordAuthError,
    DiscordBadRequestError,
    DiscordError,
    DiscordNotFoundError,
    build_embed as discord_build_embed,
    send_webhook as discord_send_webhook,
)
from webhook import (
    WebhookAuthError,
    WebhookError,
    WebhookNotFoundError,
    WebhookRejectedError,
    build_body_payload as webhook_build_body_payload,
    build_payload as webhook_build_payload,
    load_headers,
    send_webhook as webhook_send_webhook,
)
from github import (
    MAX_FILE_BYTES as GITHUB_MAX_FILE_BYTES,
    GitHubAuthError,
    GitHubError,
    GitHubNotFoundError,
    GitHubRateLimitError,
    GitHubRejectedError,
    build_commit_message as github_build_commit_message,
    build_committer as github_build_committer,
    render_path as github_render_path,
    send_file as github_send_file,
)
from notify import (
    max_attempts,
    next_retry_delay,
    record_failure,
    record_success,
)
import telegram
from telegram import (
    MAX_CAPTION_CHARS,
    MAX_MESSAGE_CHARS,
    TelegramAuthError,
    TelegramBadRequestError,
    TelegramError,
    TelegramNotFoundError,
    build_message as telegram_build_message,
    message_url as telegram_message_url,
    send_message as telegram_send_message,
    send_photo as telegram_send_photo,
)
from template_engine import render_template, render_template_rich
import alt_text
import images
from utils import FeedItem
from utils import utc_now_str as _now
from utils import categories_from_value

logger = logging.getLogger("feedecho.scheduler")

scheduler: BackgroundScheduler | None = None

MASTODON_MAX_CHARS = 500
PENDING_RECLAIM_SECONDS = 10 * 60
FEED_LEASE_SECONDS = 15 * 60
DRIP_QUEUE_CAP = 30
# Releases per flush on the unlimited (limit removed) path, to avoid
# dumping a stale backlog as one burst.
DRIP_DOWNGRADE_BATCH = 10

# Digest email size cap. Bodies are assembled item-by-item; overflow is
# held for the next flush instead of silently truncated (which reported
# dropped content as delivered). A single item larger than the whole cap
# is sent truncated so one giant item cannot wedge the queue forever.
DIGEST_MAX_CHARS = 10000
# Room reserved for the held-items notice appended when overflow occurs.
DIGEST_NOTICE_RESERVE = 80

# Email image embedding caps. Instant-mode email echoes with attach_image
# embed up to EMAIL_MAX_IMAGES images inline (mirrors Mastodon's default
# media limit); EMAIL_MAX_IMAGE_BYTES bounds the total raw bytes so the
# MIME message stays well under common relay size limits after base64
# (roughly 1.37x) inflation. Images beyond the budget are skipped, not
# silently dropped from the log.
EMAIL_MAX_IMAGES = 4
EMAIL_MAX_IMAGE_BYTES = 8 * 1024 * 1024

# Per-echo locks serializing drip rate checks across the feed-check and
# flush threads, so concurrent workers cannot exceed the hourly cap.
_drip_locks: dict[int, threading.Lock] = {}
_drip_locks_guard = threading.Lock()


def _drip_lock(echo_id: int) -> threading.Lock:
    with _drip_locks_guard:
        lock = _drip_locks.get(echo_id)
        if lock is None:
            lock = threading.Lock()
            _drip_locks[echo_id] = lock
        return lock


def _timestamp_after(seconds: int) -> str:
    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).strftime(
        "%Y-%m-%d %H:%M:%S"
    )


def _acquire_feed_lease(feed_id: int, lease_token: str) -> bool:
    """Atomically acquire a per-feed lease, reclaiming expired leases only."""
    now = _now()
    expires_at = _timestamp_after(FEED_LEASE_SECONDS)

    with get_db() as db:
        result = db.execute(
            """
            UPDATE feeds
               SET lease_token = ?,
                   lease_expires_at = ?
             WHERE id = ?
               AND (
                    lease_token IS NULL
                    OR lease_expires_at IS NULL
                    OR lease_expires_at <= ?
               )
            """,
            (lease_token, expires_at, feed_id, now),
        )
        return result.rowcount == 1


def _renew_feed_lease(feed_id: int, lease_token: str) -> bool:
    """Extend a lease owned by this worker without taking another worker's lease."""
    with get_db() as db:
        result = db.execute(
            """
            UPDATE feeds
               SET lease_expires_at = ?
             WHERE id = ?
               AND lease_token = ?
            """,
            (_timestamp_after(FEED_LEASE_SECONDS), feed_id, lease_token),
        )
        return result.rowcount == 1


def _release_feed_lease(feed_id: int, lease_token: str) -> None:
    """Release only the lease held by this worker."""
    with get_db() as db:
        db.execute(
            """
            UPDATE feeds
               SET lease_token = NULL,
                   lease_expires_at = NULL
             WHERE id = ?
               AND lease_token = ?
            """,
            (feed_id, lease_token),
        )


# ── Job-level leases (D2) ──────────────────────────────────────────────────

# Per-process identifier so every instance generates a unique lease owner.
# Reused across all job-level leases for the lifetime of this process.
_instance_id = secrets.token_urlsafe(16)

# TTLs are generous: each job runs minutes, not seconds, and the lease is
# released in a finally block. The TTL is the crash-safety net.
FLUSH_DIGEST_LEASE_SECONDS = 15 * 60
FLUSH_DRIP_LEASE_SECONDS = 5 * 60
FLUSH_QUEUE_LEASE_SECONDS = 10 * 60


def _acquire_job_lease(job_name: str, ttl_seconds: int) -> bool:
    """Atomically claim a scheduler lease for `job_name`.

    Returns True if the lease was taken (either created fresh, stolen from
    an expired owner, or renewed by the current owner). Returns False if
    another live instance holds the lease.
    """
    now = _now()
    expires_at = _timestamp_after(ttl_seconds)
    with get_db() as db:
        result = db.execute(
            """
            INSERT INTO scheduler_leases (job_name, instance_id, expires_at)
            VALUES (?, ?, ?)
            ON CONFLICT(job_name) DO UPDATE
            SET instance_id = excluded.instance_id,
                expires_at = excluded.expires_at
            WHERE scheduler_leases.instance_id = excluded.instance_id
               OR scheduler_leases.expires_at <= ?
            """,
            (job_name, _instance_id, expires_at, now),
        )
        return result.rowcount == 1


def _release_job_lease(job_name: str) -> None:
    """Release only the lease held by this instance."""
    with get_db() as db:
        db.execute(
            "DELETE FROM scheduler_leases WHERE job_name = ? AND instance_id = ?",
            (job_name, _instance_id),
        )


def _update_last_fetched(feed_id: int, lease_token: str) -> None:
    with get_db() as db:
        db.execute(
            """
            UPDATE feeds
               SET last_fetched = ?
             WHERE id = ?
               AND lease_token = ?
            """,
            (_now(), feed_id, lease_token),
        )


def _set_feed_error(feed_id: int, lease_token: str, message: str) -> None:
    with get_db() as db:
        db.execute(
            "UPDATE feeds SET last_error = ? WHERE id = ? AND lease_token = ?",
            (message[:300], feed_id, lease_token),
        )


def _clear_feed_error(feed_id: int, lease_token: str) -> None:
    with get_db() as db:
        db.execute(
            "UPDATE feeds SET last_error = NULL WHERE id = ? AND lease_token = ?",
            (feed_id, lease_token),
        )


def _save_feed_validators(
    feed_id: int,
    lease_token: str,
    etag: str | None,
    last_modified: str | None,
) -> None:
    """Persist the conditional-GET validators observed on the last fetch.

    A 200 overwrites unconditionally (a response without ETag clears the
    stored one so the next poll fetches unconditionally again); a 304 echoes
    the previously-sent value back through fetch_feed, so this write is a
    no-op on the not-modified path.
    """
    with get_db() as db:
        db.execute(
            "UPDATE feeds SET etag = ?, last_modified = ?"
            " WHERE id = ? AND lease_token = ?",
            (etag, last_modified, feed_id, lease_token),
        )


def _update_cursor(feed_id: int, lease_token: str, cursor_id: str) -> bool:
    """Advance a cursor only while the current worker still owns the lease."""
    with get_db() as db:
        result = db.execute(
            """
            UPDATE feeds
               SET last_item_id = ?,
                   last_fetched = ?
             WHERE id = ?
               AND lease_token = ?
            """,
            (cursor_id, _now(), feed_id, lease_token),
        )
        return result.rowcount == 1


def _store_feed_items(feed_id: int, items: list[dict]) -> None:
    """Persist parsed feed items for the reader, idempotent per (feed_id, item_id).

    Runs on every poll for read-enabled feeds, decoupled from echo delivery.
    ``ON CONFLICT DO NOTHING`` (backed by the UNIQUE(feed_id, item_id) index)
    skips already-stored rows; ``prune_feed_items`` then trims the feed to the
    retention cap. Item dates are normalized to the canonical UTC string so
    publish ordering is stable on both dialects.
    """
    rows = []
    for item in items:
        item_id = item.get("id")
        if not item_id:
            continue
        parsed = _parse_item_date(item.get("date")) if item.get("date") else None
        published_at = (
            parsed.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
            if parsed is not None
            else None
        )
        rows.append((
            feed_id,
            item_id,
            item.get("title") or "",
            item.get("link") or "",
            item.get("summary") or "",
            item.get("content") or "",
            item.get("content_text") or "",
            item.get("content_link") or "",
            item.get("content_html") or "",
            json.dumps(item.get("categories") or {}, ensure_ascii=False),
            item.get("author") or "",
            item.get("image_url") or "",
            item.get("image_alt") or "",
            json.dumps(item.get("image_urls") or [], ensure_ascii=False),
            item.get("enclosure_url") or "",
            published_at,
        ))

    if not rows:
        return

    with get_db() as db:
        for row in rows:
            db.execute(
                """
                INSERT INTO feed_items (
                    feed_id, item_id, title, link, summary, content,
                    content_text, content_link, content_html, categories, author,
                    image_url, image_alt, image_urls, enclosure_url, published_at
                )
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(feed_id, item_id) DO UPDATE SET
                    title = excluded.title,
                    link = excluded.link,
                    summary = excluded.summary,
                    content = excluded.content,
                    content_text = excluded.content_text,
                    content_link = excluded.content_link,
                    content_html = excluded.content_html,
                    categories = excluded.categories,
                    author = excluded.author,
                    image_url = excluded.image_url,
                    image_alt = excluded.image_alt,
                    image_urls = excluded.image_urls,
                    enclosure_url = excluded.enclosure_url,
                    published_at = excluded.published_at
                """,
                row,
            )
        prune_feed_items(db, feed_id)


def check_feed(feed_id: int) -> None:
    """Fetch and deliver new feed items under an exclusive per-feed lease."""
    lease_token = secrets.token_urlsafe(32)

    if not _acquire_feed_lease(feed_id, lease_token):
        logger.info("Feed %s is already being checked; skipping", feed_id)
        return

    try:
        _check_feed_with_lease(feed_id, lease_token)
    finally:
        _release_feed_lease(feed_id, lease_token)


def _check_feed_with_lease(feed_id: int, lease_token: str) -> None:
    with get_db() as db:
        feed = db.execute(
            "SELECT * FROM feeds WHERE id = ? AND deleted_at IS NULL", (feed_id,)
        ).fetchone()
        if not feed:
            logger.warning("Feed %s not found", feed_id)
            return

        if feed["paused"]:
            logger.info("Feed %s (%s): paused; skipping", feed_id, feed["name"])
            _update_last_fetched(feed_id, lease_token)
            return

        # Hosted mode gate: unverified owners' echoes are skipped until the
        # account email is verified, and EXPIRED-TRIAL owners' feeds are
        # paused entirely (posting_paused: plan='trial' past trial_ends_at).
        # Both gates live in the SQL so no user table is materialized per
        # feed check. Single mode keeps the original query (user 1 has no
        # trial semantics there).
        if settings.MULTI:
            echoes = db.execute(
                "SELECT e.*, u.plan FROM echoes e JOIN users u ON u.id = e.user_id"
                " WHERE e.feed_id = ? AND e.enabled = 1 AND u.email_verified = 1"
                " AND u.suspended = 0"
                " AND NOT (u.plan = 'trial' AND u.trial_ends_at IS NOT NULL"
                "          AND u.trial_ends_at <= ?)",
                (feed_id, _now()),
            ).fetchall()
        else:
            echoes = db.execute(
                "SELECT * FROM echoes WHERE feed_id = ? AND enabled = 1",
                (feed_id,),
            ).fetchall()

    feed_url = feed["url"]
    feed_name = feed["name"]
    last_seen_id = feed["last_item_id"]
    read_enabled = bool(feed["read_enabled"] or 0)

    if not echoes and not read_enabled:
        logger.info(
            "Feed %s (%s): no enabled echoes and reading disabled",
            feed_id,
            feed_name,
        )
        _update_last_fetched(feed_id, lease_token)
        return

    # Conditional GET: advertise validators only when we actually hold them, so
    # fresh feeds fetch unconditionally and existing single-arg fetch_feed
    # fakes keep their seam.
    fetch_kwargs: dict = {}
    if feed["etag"]:
        fetch_kwargs["etag"] = feed["etag"]
    if feed["last_modified"]:
        fetch_kwargs["last_modified"] = feed["last_modified"]

    try:
        feed_data = fetch_feed(feed_url, **fetch_kwargs)
        _clear_feed_error(feed_id, lease_token)
        _save_feed_validators(
            feed_id,
            lease_token,
            feed_data.get("etag"),
            feed_data.get("last_modified"),
        )
    except Exception as exc:
        logger.exception("Feed %s (%s): fetch failed", feed_id, feed_name)
        _set_feed_error(feed_id, lease_token, str(exc))
        _update_last_fetched(feed_id, lease_token)
        return

    if feed_data.get("not_modified"):
        # The feed is unchanged since the stored validators: nothing parsed,
        # nothing to deliver, cursor untouched.
        _update_last_fetched(feed_id, lease_token)
        return

    items = feed_data.get("items") or []
    if not items:
        logger.info("Feed %s (%s): no items", feed_id, feed_name)
        _update_last_fetched(feed_id, lease_token)
        return

    # Reader ingestion is decoupled from echo delivery: store every parsed
    # item regardless of echo config when reading is on, so a read-enabled
    # feed keeps its reading list even before (or without) any attached
    # echo. Queue/review echoes ALSO need the ingestion: the curation queue
    # reconstructs item metadata (images, link cards, categories) from
    # feed_items at dispatch, so a feed destined for the queue must have
    # its items stored even without the reader. The cursor/ordering logic
    # below runs only when echoes exist and is otherwise unchanged.
    try:
        ingest_items = read_enabled or any(
            (e["delivery_mode"] or "instant") in ("queue", "review") for e in echoes
        )
    except (KeyError, IndexError):
        # Malformed echo rows: fall back to the read-enabled gate only.
        ingest_items = read_enabled
    if ingest_items:
        _store_feed_items(feed_id, items)

    if not echoes:
        _update_last_fetched(feed_id, lease_token)
        return

    if last_seen_id is None:
        _update_cursor(feed_id, lease_token, items[0]["id"])
        logger.info(
            "Feed %s (%s): initialized cursor to %s",
            feed_id,
            feed_name,
            items[0]["id"],
        )
        return

    new_items = get_new_items(items, last_seen_id)
    if not new_items and not settings.ALLOW_BACKDATED_ENTRIES:
        _update_last_fetched(feed_id, lease_token)
        return

    cursor_id = last_seen_id

    # Positionally-new items: delivered oldest-first, advancing the cursor.
    for item in new_items:
        if not _renew_feed_lease(feed_id, lease_token):
            logger.warning(
                "Feed %s (%s): lease was lost before processing item %s",
                feed_id,
                feed_name,
                item["id"],
            )
            return

        all_succeeded = True
        for echo in echoes:
            if not process_echo(echo, item, feed_name=feed_name):
                all_succeeded = False

        if not all_succeeded:
            # H-2: Never process or advance past a failed earlier item.
            logger.warning(
                "Feed %s (%s): delivery failed for item %s; stopping cursor advancement",
                feed_id,
                feed_name,
                item["id"],
            )
            break

        cursor_id = item["id"]

    # Backdated items: delivered without moving the cursor. Each echo's
    # _claim_post upsert deduplicates against already-posted rows, so items
    # that were already delivered (positionally or in a prior backdated pass)
    # are silently skipped. A delivery failure here does NOT stop the cursor
    # — the cursor was never at risk of advancing past these items.
    if settings.ALLOW_BACKDATED_ENTRIES:
        backdated = get_backdated_items(
            items,
            last_seen_id,
            max_days=settings.MAX_BACKDATED_ENTRY_DAYS,
            limit=20,
        )
        for item in backdated:
            if not _renew_feed_lease(feed_id, lease_token):
                logger.warning(
                    "Feed %s (%s): lease was lost during backdated delivery",
                    feed_id,
                    feed_name,
                )
                return

            for echo in echoes:
                process_echo(echo, item, feed_name=feed_name)

    if cursor_id != last_seen_id:
        if not _update_cursor(feed_id, lease_token, cursor_id):
            logger.warning("Feed %s: cursor was not updated because lease was lost", feed_id)
    else:
        _update_last_fetched(feed_id, lease_token)

    _retry_due_failures(feed_id, echoes, feed_name=feed_name)


def _retry_due_failures(feed_id: int, echoes, feed_name: str = "") -> None:
    """Reprocess failed rows whose backoff has elapsed, regardless of cursor.

    Normal cursor replay only covers items at-or-after the cursor. Failed rows
    can sit behind it (e.g. an item that failed, was manually reset, or was
    blocked while another echo's item gated advancement). This sweep gives
    them their scheduled retries without disturbing feed ordering guarantees.
    """
    if not echoes:
        return
    echo_ids = [e["id"] for e in echoes]
    placeholders = ",".join("?" for _ in echo_ids)

    with get_db() as db:
        due = db.execute(
            f"""
            SELECT id, echo_id, item_id FROM posted_items
             WHERE echo_id IN ({placeholders})
               AND status = 'failed'
               AND next_retry_at IS NOT NULL
               AND next_retry_at <= ?
             ORDER BY id
             LIMIT 25
            """,
            (*echo_ids, _now()),
        ).fetchall()

    if not due:
        return

    # We only have item_id stored, not the full item payload. Fetch the feed
    # once and match; items that have aged out of the feed are marked gave_up.
    # This fetch is deliberately UNCONDITIONAL (no etag/last_modified): the
    # sweep needs the full current body to match item_ids, and a 304 here
    # would leave every pending item unmatched and marked "gave_up".
    try:
        feed_data = fetch_feed(db_feed_url(feed_id))
    except Exception:
        logger.exception("Feed %s: fetch failed during retry sweep", feed_id)
        return

    items_by_id = {it["id"]: it for it in (feed_data.get("items") or [])}
    echoes_by_id = {e["id"]: e for e in echoes}

    for row in due:
        item = items_by_id.get(row["item_id"])
        if item is None:
            with get_db() as db:
                db.execute(
                    """UPDATE posted_items SET status = 'gave_up',
                          error_message = 'Item no longer in feed; cannot retry',
                          next_retry_at = NULL
                        WHERE id = ? AND status = 'failed'""",
                    (row["id"],),
                )
            continue
        echo = echoes_by_id.get(row["echo_id"])
        if echo is not None:
            process_echo(echo, item, feed_name=feed_name)


def db_feed_url(feed_id: int) -> str:
    with get_db() as db:
        row = db.execute(
            "SELECT url FROM feeds WHERE id = ? AND deleted_at IS NULL", (feed_id,)
        ).fetchone()
    if not row:
        raise ValueError(f"Feed {feed_id} not found")
    return row["url"]


def _claim_post(echo_id: int, item: FeedItem) -> tuple[int, str] | None:
    """Atomically claim an echo/item row.

    Returns ``(posted_item_id, claim_token)`` only for the worker that owns the
    pending attempt. Fresh pending rows cannot be claimed. Failed rows can be
    reclaimed once their backoff (next_retry_at) has elapsed, and pending rows
    abandoned for over ten minutes can be reclaimed.
    """
    item_id = item["id"]
    claim_token = secrets.token_urlsafe(32)
    now = _now()
    stale_before = _timestamp_after(-PENDING_RECLAIM_SECONDS)

    with get_db() as db:
        result = db.execute(
            """
            INSERT INTO posted_items (
                echo_id, item_id, item_title, item_url, status,
                claimed_at, claim_token, attempt_count, error_message
            )
            VALUES (?, ?, ?, ?, 'pending', ?, ?, 1, NULL)
            ON CONFLICT(echo_id, item_id) DO UPDATE SET
                item_title = excluded.item_title,
                item_url = excluded.item_url,
                status = 'pending',
                claimed_at = excluded.claimed_at,
                claim_token = excluded.claim_token,
                attempt_count = posted_items.attempt_count + 1,
                error_message = NULL
            WHERE posted_items.status = 'failed'
               AND (
                    posted_items.next_retry_at IS NULL
                    OR posted_items.next_retry_at <= ?
               )
               OR (
                    posted_items.status = 'pending'
                    AND (
                        posted_items.claimed_at IS NULL
                        OR posted_items.claimed_at <= ?
                    )
               )
            """,
            (
                echo_id,
                item_id,
                item.get("title", ""),
                item.get("link", ""),
                now,
                claim_token,
                now,
                stale_before,
            ),
        )

        if result.rowcount != 1:
            return None

        row = db.execute(
            """
            SELECT id, claim_token
              FROM posted_items
             WHERE echo_id = ?
               AND item_id = ?
            """,
            (echo_id, item_id),
        ).fetchone()

    if not row or row["claim_token"] != claim_token:
        return None

    return row["id"], claim_token


def _row_state(echo_id: int, item_id: str) -> str | None:
    with get_db() as db:
        row = db.execute(
            "SELECT status FROM posted_items WHERE echo_id = ? AND item_id = ?",
            (echo_id, item_id),
        ).fetchone()
    return row["status"] if row else None


def _post_succeeded(echo_id: int, item_id: str) -> bool:
    """True when the item is in a terminal state for this echo.

    'success', 'filtered', 'gave_up', and 'queued' all count as handled:
    they must not be retried and must not block cursor advancement.
    'queued' means the item is waiting for digest flush.
    A 'failed' row waiting out its retry backoff returns False here —
    it still gates the cursor — but process_echo distinguishes it from
    a fresh failure via _row_state so the run stops quietly instead of
    logging new failures.
    """
    return _row_state(echo_id, item_id) in ("success", "filtered", "gave_up", "queued")


def _record_filtered(echo_id: int, item: FeedItem, reason: str | None = None) -> None:
    """Record an item suppressed by the echo's keyword filter.

    Uses status 'filtered' so history shows what was dropped and the claim
    logic never retries it (only 'failed' and stale 'pending' rows are
    reclaimable). Storing the filter reason explains why it was dropped.
    """
    with get_db() as db:
        db.execute(
            """
            INSERT INTO posted_items (
                echo_id, item_id, item_title, item_url, status,
                attempt_count, error_message
            )
            VALUES (?, ?, ?, ?, 'filtered', 0, ?)
            ON CONFLICT(echo_id, item_id) DO UPDATE SET
                error_message = excluded.error_message
            WHERE posted_items.status = 'filtered'
            """,
            (echo_id, item["id"], item.get("title", ""), item.get("link", ""), reason),
        )


def process_echo(
    echo,
    item: FeedItem,
    feed_name: str = "",
    override_content: str | None = None,
) -> bool:
    """Deliver one item to one echo using an atomic pending-row claim.

    Items past a drip rate limit are held in the drip queue (status
    'queued') instead of dispatching immediately; flush_drips() releases
    them as the rate window allows.

    feed_name is optional context for template rendering ({{ feed_name }}).
    When override_content is provided, template rendering is skipped and
    that content is dispatched directly (e.g. from Reader Compose).
    """
    echo_id = echo["id"]
    item_id = item["id"]

    # Keyword filter: suppressed items count as handled so the cursor
    # advances and they are never delivered or retried.
    # .get-style access keeps plain-dict fixtures in tests working.
    try:
        filter_kw = echo["filter_keywords"]
        filter_mode = echo["filter_mode"]
    except (KeyError, IndexError):
        filter_kw, filter_mode = None, None
    reason = match_reason(item, filter_kw, filter_mode)
    if reason is not None:
        _record_filtered(echo_id, item, reason=reason)
        logger.info("Echo %s: item %s suppressed by keyword filter: %s", echo_id, item_id, reason)
        return True

    claimed = _claim_post(echo_id, item)
    if claimed is None:
        # Terminal states (success/filtered/gave_up) count as handled. A fresh
        # pending row belongs to another worker, and a failed row waiting out
        # its retry backoff is deferred — both are "not done", so the cursor
        # stays put and the run stops at this item (H-2), quietly.
        return _post_succeeded(echo_id, item_id)

    posted_id, claim_token = claimed

    try:
        delivery_mode = echo["delivery_mode"] or "instant"
    except (KeyError, IndexError):
        delivery_mode = "instant"
    if delivery_mode in ("queue", "review"):
        return _enqueue_for_queue_mode(
            echo, item, posted_id, claim_token, delivery_mode, feed_name=feed_name
        )

    if _drip_applies(echo):
        # Serialize the rate check + dispatch per echo so the flush job
        # and feed checks cannot both see an open window and exceed the cap.
        with _drip_lock(echo_id):
            if _drip_rate(echo_id) >= _drip_limit(echo):
                return _queue_for_drip(echo, item, posted_id, claim_token)
            return _render_and_dispatch(
                echo, item, feed_name, posted_id, claim_token, override_content=override_content
            )

    return _render_and_dispatch(
        echo, item, feed_name, posted_id, claim_token, override_content=override_content
    )


def _drip_limit(echo) -> int:
    """The echo's drip rate limit (posts/hour); 0 disables drip.

    Hosted mode clamps the configured limit DOWN to the plan's ceiling.
    The plan comes from the 'plan' key on the echo row (joined at selection
    time — set-based, never a per-echo lookup). A missing/unreadable plan
    fails CLOSED at the trial ceiling: drip enforcement never silently
    vanishes; the worst case is over-throttling until the next tick.
    """
    try:
        limit = int(echo["drip_limit"] or 0)
    except (KeyError, IndexError, TypeError, ValueError):
        return 0
    if settings.MULTI and limit > 0:
        try:
            plan = echo["plan"] or "trial"
        except (KeyError, IndexError):
            plan = "trial"  # fail closed
        limit = plans.clamp_drip_limit(limit, plan)
    return limit


def _drip_applies(echo) -> bool:
    """Drip applies to rate-limited echoes with instant delivery only.

    Digest-mode echoes already batch on their own schedule, and
    queue/review modes buffer into the curation queue at a different
    point, so a drip limit would double-throttle them.
    """
    if _drip_limit(echo) <= 0:
        return False
    try:
        delivery_mode = echo["delivery_mode"] or "instant"
    except (KeyError, IndexError):
        delivery_mode = "instant"
    return delivery_mode == "instant"


def _drip_rate(echo_id: int) -> int:
    """Successful posts by this echo within the sliding 60-minute window."""
    cutoff = _timestamp_after(-60 * 60)
    with get_db() as db:
        row = db.execute(
            """
            SELECT COUNT(*) AS n FROM posted_items
             WHERE echo_id = ?
               AND status = 'success'
               AND posted_at >= ?
            """,
            (echo_id, cutoff),
        ).fetchone()
    return row["n"] if row else 0


def _queue_for_drip(echo, item: FeedItem, posted_id: int, claim_token: str) -> bool:
    """Hold an item until the drip window has room.

    Marks the posted row 'queued' (terminal for the cursor, same as
    digest mode) and stores the item payload for flush_drips(). When the
    queue is full the item is dropped as gave_up so an oversubscribed
    feed cannot pile up an unbounded backlog.
    """
    echo_id = echo["id"]
    item_id = item["id"]

    with get_db() as db:
        row = db.execute(
            "SELECT COUNT(*) AS n FROM drip_items WHERE echo_id = ?", (echo_id,)
        ).fetchone()
    if row["n"] >= DRIP_QUEUE_CAP:
        logger.warning(
            "Echo %s: drip queue full (%d); dropping item %s",
            echo_id,
            DRIP_QUEUE_CAP,
            item_id,
        )
        return _fail_post(
            posted_id,
            claim_token,
            echo_id,
            f"Drip queue full ({DRIP_QUEUE_CAP} pending); item dropped",
            permanent=True,
        )

    # 'raw' holds feedparser internals (struct_time etc.) that are not
    # JSON-serializable and never used for rendering.
    payload = {k: v for k, v in item.items() if k != "raw"}
    with get_db() as db:
        db.execute(
            """
            INSERT INTO drip_items (echo_id, item_id, item_json)
            VALUES (?, ?, ?)
            ON CONFLICT(echo_id, item_id) DO NOTHING
            """,
            (echo_id, item_id, json.dumps(payload)),
        )

    ok = _update_post(posted_id, claim_token, "queued")
    if ok:
        # Deliberately no record_success() here: queueing is not a
        # delivery, and clearing the failure-notify state on a hold would
        # suppress alerts for echoes that are not actually delivering.
        logger.info("Echo %s: item %s held for drip release", echo_id, item_id)
    return ok


def _enqueue_for_queue_mode(
    echo,
    item: FeedItem,
    posted_id: int,
    claim_token: str,
    mode: str,
    feed_name: str = "",
) -> bool:
    """Bridge an auto-queue or review-mode echo item into the curation queue.

    Never dispatches via the destination sender: renders the stored template,
    inserts one queued_posts row (status 'queued' for auto-queue, 'draft' for
    review), marks the posted_items row 'queued' so the feed cursor advances
    (same terminal-hold semantics as drip/digest), and lets the existing
    _flush_queue / post-now machinery handle delivery. feed_name is template
    context only ({{ feed_name }}), mirroring the instant path.
    """
    echo_id = echo["id"]
    item_id = item["id"]

    # Defensive plan access mirroring _drip_applies: keys may be missing on
    # test fixtures and plain-dict rows; fail closed at 'trial'.
    try:
        user_id = echo["user_id"]
    except (KeyError, IndexError):
        user_id = 1
    try:
        plan = echo["plan"] or "trial"
    except (KeyError, IndexError):
        plan = "trial"

    with get_db() as db:
        if settings.MULTI:
            # Plan cap enforced like the Compose desk: fail LOUDLY, never
            # silently drop the item. _check_allowance adds its own +1
            # ("adding one more would exceed the cap"), so pass the raw
            # count — passing n+1 double-counted and rejected at cap-1.
            row = db.execute(
                """
                SELECT COUNT(*) AS n FROM queued_posts
                 WHERE user_id = ?
                   AND status IN ('queued', 'sending', 'draft')
                """,
                (user_id,),
            ).fetchone()
            try:
                plans.check_queue_allowance(row["n"], plan)
            except plans.PlanError:
                return _fail_post(
                    posted_id,
                    claim_token,
                    echo_id,
                    "Echo auto-queue is at your plan's queue limit;"
                    " free space or switch this echo back to Instant",
                )

    # Render the stored template into materialized text. The queued row
    # carries the content verbatim from here on (same as Reader Compose).
    try:
        content = render_template(echo["template"], item, feed_name=feed_name)
    except Exception as e:
        logger.exception("Echo %s: template render failed for item %s", echo_id, item_id)
        return _fail_post(posted_id, claim_token, echo_id, f"Template rendering failed: {e}")
    if not content.strip():
        return _fail_post(posted_id, claim_token, echo_id, "Rendered content was empty")

    try:
        visibility = echo["visibility"] or "public"
    except (KeyError, IndexError):
        visibility = "public"
    try:
        feed_id = echo["feed_id"]
    except (KeyError, IndexError):
        feed_id = 0
    attach_image = _echo_attach_image(echo)
    image_alt = item.get("image_alt") or ""
    try:
        content_warning = echo["content_warning"] or ""
    except (KeyError, IndexError):
        content_warning = ""
    try:
        use_markdown = 1 if echo["use_markdown"] else 0
    except (KeyError, IndexError):
        use_markdown = 0

    # Resolve the feed_items row id so _flush_queue dispatches with the
    # full item metadata (images, content_html, categories) the way the
    # Reader Compose enqueue does. _check_feed_with_lease ingests items for
    # ANY feed with a queue/review echo (the same condition that routes us
    # here), but the row can still be absent (pruned by retention before a
    # late slot, or the item fell out of the feed) — a miss is NOT fatal:
    # the queued row carries its rendered content and _flush_queue
    # tolerates NULL there (http metadata is then blank, matching the
    # manual-compose fallback).
    now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
    with get_db() as db:
        fi_row = db.execute(
            "SELECT id FROM feed_items WHERE feed_id = ? AND item_id = ?",
            (feed_id, item_id),
        ).fetchone()
        metadata_feed_item_id = fi_row["id"] if fi_row else None

        # One-shot echo at enqueue time so crash-recovery retries reuse the
        # same echo_id (idempotency key for _claim_post), mirroring the
        # Reader Compose desk's insert. The stored content carries the real
        # text; this echoed row is inert for re-rendering. The parent's
        # content warning / markdown flag are carried over so dispatch
        # keeps the echo's configured voice.
        new_echo_id = db.execute(
            """
            INSERT INTO echoes (feed_id, destination_type, destination_id, template,
                                visibility, attach_image, content_warning, use_markdown,
                                enabled, one_shot, deleted_at, user_id)
            VALUES (?, ?, ?, '{{ title }}', ?, ?, ?, ?, 0, 1, ?, ?) RETURNING id
            """,
            (feed_id, echo["destination_type"], echo["destination_id"], visibility,
             attach_image, content_warning, use_markdown, now_utc, user_id),
        ).fetchone()["id"]

        # Prefer the freshly-read feed_items id when we got one; fall back
        # to whatever the caller supplied (tests pass items without rows).
        supplied = None
        try:
            supplied = item.get("feed_item_id")
        except AttributeError:
            pass
        queued_feed_item_id = metadata_feed_item_id or supplied

        if mode == "review":
            target_time = _now()
        else:
            target_time = next_free_slot(
                db, user_id, echo["destination_type"], echo["destination_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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                user_id,
                queued_feed_item_id,
                item_id,
                feed_id,
                echo["destination_type"],
                echo["destination_id"],
                content,
                visibility,
                attach_image,
                image_alt,
                target_time,
                "draft" if mode == "review" else "queued",
                new_echo_id,
            ),
        )

    ok = _update_post(posted_id, claim_token, "queued")
    if ok:
        # Deliberately no record_success() here: queueing is not a delivery,
        # matching drip semantics.
        logger.info(
            "Echo %s: item %s enqueued for %s at %s",
            echo_id,
            item_id,
            mode,
            target_time,
        )
    return ok


def next_free_slot(db, uid: int, dest_type: str, dest_id: int) -> str:
    """Compute the next free scheduled_at timestamp for (uid, destination).

    max(now, last scheduled_at) + interval_minutes.
    Snaps into queue_active_hours ("HH:MM-HH:MM") in UTC.
    """
    dest_key = f"{dest_type}:{dest_id}"
    setting = db.execute(
        "SELECT queue_interval_minutes, queue_active_hours FROM queue_settings WHERE user_id = ? AND destination_key = ?",
        (uid, dest_key),
    ).fetchone()

    interval = setting["queue_interval_minutes"] if setting else 240
    active_hours = setting["queue_active_hours"] if setting else "09:00-21:00"

    last_row = db.execute(
        """
        SELECT scheduled_at FROM queued_posts
         WHERE user_id = ? AND destination_type = ? AND destination_id = ?
           AND status IN ('queued', 'sending')
         ORDER BY scheduled_at DESC LIMIT 1
        """,
        (uid, dest_type, dest_id),
    ).fetchone()

    now_dt = datetime.now(timezone.utc).replace(tzinfo=None, second=0, microsecond=0)
    if last_row and last_row["scheduled_at"]:
        last_dt = as_utc_naive(last_row["scheduled_at"])
        base_dt = max(now_dt, last_dt) if last_dt else now_dt
    else:
        base_dt = now_dt

    candidate = base_dt + timedelta(minutes=interval)

    # Snap to active hours if configured
    if active_hours and "-" in active_hours:
        try:
            start_str, end_str = active_hours.split("-", 1)
            sh, sm = [int(p) for p in start_str.split(":", 1)]
            eh, em = [int(p) for p in end_str.split(":", 1)]
            start_mins = sh * 60 + sm
            end_mins = eh * 60 + em
            cand_mins = candidate.hour * 60 + candidate.minute

            if start_mins <= end_mins:
                # Same day window, e.g. 09:00 - 21:00
                if cand_mins < start_mins:
                    candidate = candidate.replace(hour=sh, minute=sm)
                elif cand_mins > end_mins:
                    # Move to next day start
                    candidate = (candidate + timedelta(days=1)).replace(hour=sh, minute=sm)
            else:
                # Overnight window, e.g. 22:00 - 06:00
                if end_mins < cand_mins < start_mins:
                    candidate = candidate.replace(hour=sh, minute=sm)
        except Exception:
            pass

    return candidate.strftime("%Y-%m-%d %H:%M:%S")


def _reclaim_queued(echo_id: int, item_id: str) -> tuple[int, str] | None:
    """Transition a queued posted row back to pending with a fresh claim.

    Returns (posted_id, claim_token), or None if the row is no longer
    queued (already released or finalized by another worker).

    Deliberately does not bump attempt_count: release attempts are
    tracked on the drip_items row so failed releases can retry from the
    stored payload without burning the normal retry budget.
    """
    claim_token = secrets.token_urlsafe(32)
    with get_db() as db:
        row = db.execute(
            """
            SELECT id FROM posted_items
             WHERE echo_id = ? AND item_id = ? AND status = 'queued'
            """,
            (echo_id, item_id),
        ).fetchone()
        if not row:
            return None
        result = db.execute(
            """
            UPDATE posted_items
               SET status = 'pending',
                   claimed_at = ?,
                   claim_token = ?
             WHERE id = ? AND status = 'queued'
            """,
            (_now(), claim_token, row["id"]),
        )
        if result.rowcount != 1:
            return None
    return row["id"], claim_token


def _render_and_dispatch(
    echo,
    item: FeedItem,
    feed_name: str,
    posted_id: int,
    claim_token: str,
    override_content: str | None = None,
) -> bool:
    """Render the echo template and dispatch to the destination sender.

    Shared by process_echo (fresh feed items) and flush_drips (released
    queue items) so both paths get identical rendering, failure handling,
    and image/alt-text/CW behavior. When override_content is not None,
    template rendering is skipped.
    """
    echo_id = echo["id"]
    item_id = item["id"]
    destination_type = echo["destination_type"]

    if override_content is not None:
        content = override_content
        from template_engine import _MAX_OUTPUT
        if len(content) > _MAX_OUTPUT:
            gave_up = _fail_post(
                posted_id,
                claim_token,
                echo_id,
                f"Content exceeds {_MAX_OUTPUT}-char cap",
            )
            return gave_up
    else:
        try:
            if destination_type == "bluesky":
                # Rich render: content_html anchors become visible text plus
                # link spans the Bluesky sender merges into facets. Stored
                # as _rich_links on a copied item; the drip path (override
                # content) has no spans and falls back to bare-URL facets.
                content, rich_links = render_template_rich(
                    echo["template"], dict(item), feed_name=feed_name
                )
                item = dict(item)  # type: ignore[assignment]
                item["_rich_links"] = rich_links  # type: ignore[typeddict-unknown-key]
            else:
                content = render_template(echo["template"], item, feed_name=feed_name)
        except Exception as e:
            logger.exception("Echo %s: template render failed for item %s", echo_id, item_id)
            gave_up = _fail_post(
                posted_id,
                claim_token,
                echo_id,
                f"Template rendering failed: {e}",
            )
            return gave_up

    if not content.strip():
        gave_up = _fail_post(posted_id, claim_token, echo_id, "Rendered content was empty")
        return gave_up

    if destination_type == "webhook":
        return _send_webhook(echo, item, content, feed_name, echo["destination_id"], posted_id, claim_token)

    handler = _DESTINATION_HANDLERS.get(destination_type)
    if handler is not None:
        return handler(echo, item, content, echo["destination_id"], posted_id, claim_token)

    gave_up = _fail_post(
        posted_id,
        claim_token,
        echo_id,
        f"Unknown destination type: {echo['destination_type']}",
    )
    return gave_up


def _still_owns_claim(posted_id: int, claim_token: str) -> bool:
    """Whether this worker still holds the claim on a pending row.

    Dispatch paths do slow network I/O between _claim_post and the actual
    send. If the lease lapses and another worker reclaims the row, both would
    post. Check immediately before the irreversible step.
    """
    with get_db() as db:
        return (
            db.execute(
                """
                SELECT 1 FROM posted_items
                 WHERE id = ? AND status = 'pending' AND claim_token = ?
                """,
                (posted_id, claim_token),
            ).fetchone()
            is not None
        )


def _update_post(
    posted_id: int,
    claim_token: str,
    status: str,
    error: str | None = None,
    post_url: str | None = None,
) -> bool:
    """Finalize only the pending row still owned by this claim token."""
    with get_db() as db:
        result = db.execute(
            """
            UPDATE posted_items
               SET status = ?,
                   error_message = ?,
                   post_url = COALESCE(?, post_url),
                   claimed_at = NULL,
                   claim_token = NULL,
                   posted_at = ?
             WHERE id = ?
               AND status = 'pending'
               AND claim_token = ?
            """,
            (status, error, post_url, _now(), posted_id, claim_token),
        )
        return result.rowcount == 1


def _fail_post(
    posted_id: int,
    claim_token: str,
    echo_id: int,
    error: str,
    permanent: bool = False,
) -> bool:
    """Mark a claimed row failed, scheduling the next automatic retry.

    If the row has exhausted retry_max_attempts, it is marked 'gave_up'
    (terminal) instead, which unblocks the feed cursor. Passing
    ``permanent=True`` skips straight to 'gave_up' for failures that no
    number of retries can fix (missing account, rejected credentials).
    Returns the final status written.
    """
    cap = max_attempts(echo_id)

    with get_db() as db:
        row = db.execute(
            "SELECT attempt_count FROM posted_items WHERE id = ?", (posted_id,)
        ).fetchone()
    attempts = row["attempt_count"] if row else 1

    if permanent or (cap > 0 and attempts >= cap):
        final = "gave_up"
        if permanent:
            error_out = error
        else:
            error_out = f"Gave up after {attempts} attempts. Last error: {error}"
        retry_at = None
    else:
        final = "failed"
        error_out = error
        retry_at = next_retry_delay(attempts, echo_id=echo_id)

    with get_db() as db:
        db.execute(
            """
            UPDATE posted_items
               SET status = ?,
                   error_message = ?,
                   next_retry_at = ?,
                   claimed_at = NULL,
                   claim_token = NULL,
                   posted_at = ?
             WHERE id = ?
               AND status = 'pending'
               AND claim_token = ?
            """,
            (final, error_out, retry_at, _now(), posted_id, claim_token),
        )

    record_failure(echo_id)
    if final == "gave_up":
        logger.error(
            "Echo %s: item gave up after %s attempts: %s", echo_id, attempts, error
        )
    return final == "gave_up"


def _item_image_entries(item: FeedItem, limit: int = 4) -> list[dict]:
    """Image URL/alt entries from an item, shared by destination senders.

    Returns up to `limit` entries of {"url", "alt"} taken from the item's
    image_urls list, falling back to the single image_url for legacy rows.
    A caller-supplied image_alt (reader compose / queue override) wins over
    the feed alt for the primary slot — the user wrote it.
    """
    image_entries: list[dict] = []
    raw_urls = item.get("image_urls")
    if raw_urls:
        if isinstance(raw_urls, str):
            try:
                raw_urls = json.loads(raw_urls)
            except ValueError:
                raw_urls = []
        if isinstance(raw_urls, list):
            for img in raw_urls[:limit]:
                if isinstance(img, dict) and isinstance(img.get("url"), str) and img["url"].strip():
                    image_entries.append(
                        {"url": img["url"].strip(), "alt": (img.get("alt") or "").strip()}
                    )
                elif isinstance(img, str) and img.strip():
                    image_entries.append({"url": img.strip(), "alt": ""})
    if not image_entries and item.get("image_url"):
        image_entries.append(
            {"url": item.get("image_url"), "alt": (item.get("image_alt") or "").strip()}
        )
    if image_entries and item.get("image_alt") and not image_entries[0]["alt"]:
        image_entries[0]["alt"] = item["image_alt"].strip()
    return image_entries


# --- Shared destination-dispatch skeleton ----------------------------------
#
# The seven _send_* dispatchers below all run the same shape: fetch the
# account row (missing = permanent failure), resolve the attach_image flag,
# re-validate the claim right before the irreversible send, then finalize.
# These helpers hold that shared skeleton so a fix lands in one place — the
# mastodon missing-account path drifting off permanent=True is what happens
# when it is copy-pasted instead.


def _destination_account(
    table: str,
    account_id: int,
    echo,
    posted_id: int,
    claim_token: str,
    label: str,
    user_scoped: bool = False,
):
    """Fetch the destination account row; fail permanently when gone.

    A missing account can never heal on retry (the user deleted it), so the
    row finalizes straight to 'gave_up'. Returns ``(account, None)`` on
    success, or ``(None, fail_result)`` after finalizing the row — callers
    return ``fail_result`` directly, preserving _fail_post's gave_up=True
    contract that unblocks the feed cursor.
    """
    if user_scoped:
        sql = f"SELECT * FROM {table} WHERE id = ? AND user_id = ?"
        params = (account_id, echo["user_id"])
    else:
        sql = f"SELECT * FROM {table} WHERE id = ?"
        params = (account_id,)
    with get_db() as db:
        account = db.execute(sql, params).fetchone()

    if not account:
        fail_result = _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"{label} {account_id} not found",
            permanent=True,
        )
        return None, fail_result
    return account, None


def _echo_attach_image(echo) -> bool:
    """The echo's attach_image flag, defaulting off for legacy/partial rows."""
    try:
        return bool(echo["attach_image"])
    except (KeyError, IndexError):
        return False


def _echo_render_html(echo) -> bool:
    """The echo's render_html flag (email/Matrix destinations), defaulting off."""
    try:
        return bool(echo["render_html"])
    except (KeyError, IndexError):
        return False


def _echo_use_markdown(echo) -> bool:
    """The echo's use_markdown flag (Mastodon/GoToSocial), defaulting off."""
    try:
        return bool(echo["use_markdown"])
    except (KeyError, IndexError):
        return False


def _feed_ai_alt_override(feed_id) -> bool:
    """The feed's ai_alt_override flag, defaulting off for legacy/partial rows.

    Without the flag (missing column value, missing feed row, lookup
    failure) the historical feed-alt-wins precedence applies, so every
    failure mode here must read as False.
    """
    if feed_id is None:
        return False
    try:
        with get_db() as db:
            row = db.execute(
                "SELECT ai_alt_override FROM feeds WHERE id = ?", (feed_id,)
            ).fetchone()
    except Exception:
        logger.warning(
            "Feed %s: ai_alt_override lookup failed; defaulting to off",
            feed_id, exc_info=True,
        )
        return False
    if not row:
        return False
    try:
        return bool(row["ai_alt_override"])
    except (KeyError, IndexError):
        return False


def _echo_feed_id(echo):
    try:
        return echo["feed_id"]
    except (KeyError, IndexError):
        return None


def _resolve_alt_text(echo, item, feed_alt: str, img_bytes, img_type) -> str:
    """Resolve the alt text to attach to an uploaded image ("" when neither).

    Default precedence: feed-provided alt text wins; AI generation is the
    fallback, degrading to an empty string on AI failure. When the feed's
    feeds.ai_alt_override flag is set the precedence flips: AI generation
    runs even with non-empty feed_alt and wins on success, with feed_alt
    kept as the fallback when AI fails or is disabled for the user.
    """
    ai_override = _feed_ai_alt_override(_echo_feed_id(echo))
    description = feed_alt
    if ai_override:
        try:
            if alt_text.is_enabled(user_id=echo["user_id"]):
                generated = alt_text.generate_alt_text(
                    img_bytes, img_type, user_id=echo["user_id"]
                )
                if generated:
                    logger.info(
                        "Echo %s: AI alt text override wins for item %s (%d chars)",
                        echo["id"], item["id"], len(generated),
                    )
                    description = generated
        except Exception:
            # A failed override keeps the feed-provided alt rather than
            # posting the image with nothing. Documented in the v1.78.0 notes
            # ("Feed descriptions remain the fallback").
            logger.warning(
                "Echo %s: alt text override pipeline failed for item %s; "
                "falling back to feed-provided alt",
                echo["id"], item["id"],
                exc_info=True,
            )
    elif description:
        logger.info(
            "Echo %s: using feed-provided alt text for item %s (%d chars)",
            echo["id"], item["id"], len(description),
        )
    elif alt_text.is_enabled(user_id=echo["user_id"]):
        try:
            description = alt_text.generate_alt_text(
                img_bytes, img_type, user_id=echo["user_id"]
            )
            if description:
                logger.info(
                    "Echo %s: generated alt text for item %s (%d chars)",
                    echo["id"], item["id"], len(description),
                )
        except Exception:
            logger.warning(
                "Echo %s: alt text generation failed for item %s",
                echo["id"], item["id"],
                exc_info=True,
            )
    return description


def _guard_claim(posted_id: int, claim_token: str, echo_id, item_id, dest_name: str) -> bool:
    """Re-validate claim ownership immediately before the irreversible send.

    Dispatch paths do slow network I/O (image fetch, alt text, upload)
    between _claim_post and the send; if the lease lapsed and another worker
    reclaimed the row, sending now would duplicate the public post. Returns
    True when this worker still owns the claim.
    """
    if not _still_owns_claim(posted_id, claim_token):
        logger.warning(
            "Echo %s: claim lost before %s dispatch; skipping item %s",
            echo_id,
            dest_name,
            item_id,
        )
        return False
    return True


def _finalize_success(
    posted_id: int, claim_token: str, echo_id, post_url: str | None = None
) -> bool:
    """Mark the row delivered and clear the echo's failure alert state."""
    ok = _update_post(posted_id, claim_token, "success", post_url=post_url)
    if ok:
        record_success(echo_id)
    return ok


def _maybe_boost(account, echo, post_url, visibility: str = "public") -> None:
    """Fire-and-forget boost request to the FeedBooster service, when a
    booster is configured and either the destination account or this
    individual echo has boosting enabled (the account setting covers every
    echo to it, so either flag is sufficient).

    Caller-side visibility gate: only public/unlisted echoes are boosted.
    The booster re-checks the object's Public addressing on its side, but
    the leak must be prevented here, not caught downstream.

    Best-effort by design: the post is already published, so a booster outage
    or refusal must never fail the delivery — log-only, no retries from this
    side (the booster keeps its own retry queue).
    """
    if visibility not in ("public", "unlisted"):
        return
    try:
        account_on = bool(account["booster_enabled"])
    except (KeyError, IndexError, TypeError):
        account_on = False
    try:
        echo_on = bool(echo["booster_enabled"])
    except (KeyError, IndexError, TypeError):
        echo_on = False
    if not (account_on or echo_on):
        return
    if not settings.BOOSTER_URL or not settings.BOOSTER_TOKEN:
        return
    try:
        echo_id = echo["id"]
    except (KeyError, IndexError, TypeError):
        echo_id = "?"
    try:
        response = httpx.post(
            f"{settings.BOOSTER_URL}/internal/boost",
            json={"url": post_url},
            headers={"authorization": f"Bearer {settings.BOOSTER_TOKEN}"},
            timeout=5,
        )
    except Exception:
        logger.warning("Echo %s: boost request for %s failed", echo_id, post_url, exc_info=True)
        return
    if response.status_code >= 400:
        logger.warning(
            "Echo %s: booster refused %s: HTTP %s", echo_id, post_url, response.status_code
        )
    else:
        logger.info("Echo %s: boost requested for %s", echo_id, post_url)


def _send_mastodon(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    account, fail_result = _destination_account(
        "accounts", account_id, echo, posted_id, claim_token, "Account"
    )
    if account is None:
        return fail_result

    # Content warning: per-echo CW text, applied via Mastodon's spoiler_text
    try:
        cw_text = (echo["content_warning"] or "").strip()
    except (KeyError, IndexError):
        cw_text = ""
    sensitive = bool(cw_text)

    # Image attachment: if enabled, upload up to 4 images (Mastodon's default
    # limit) from the item's image_urls list, falling back to the single
    # image_url for legacy rows.
    media_ids: list[str] = []
    attach_image = _echo_attach_image(echo)

    if attach_image:
        image_entries = _item_image_entries(item)

        for entry in image_entries:
            image_result = fetch_image(entry["url"])
            if not image_result:
                logger.info(
                    "Echo %s: image fetch failed for %s, skipping that image",
                    echo["id"], entry["url"],
                )
                continue
            img_bytes, img_type = image_result
            # Feed-provided alt text wins (the author wrote it); AI
            # generation is the fallback when the feed has none.
            description = _resolve_alt_text(echo, item, entry["alt"], img_bytes, img_type)
            try:
                uploaded = upload_media(
                    instance=account["instance"],
                    access_token=decrypt_secret(account["access_token"]),
                    image_bytes=img_bytes,
                    content_type=img_type,
                    description=description,
                )
            except MastodonAuthError as e:
                # Permanent (revoked/expired token, or a token that can post
                # but lacks the media scope): finalize the post as
                # permanently failed right here. Falling through to
                # post_status would silently publish text-only whenever the
                # token can post but not upload media, and would route a
                # pure 401 into the transient-retry path whenever
                # post_status hit an unrelated transient failure.
                # _fail_post fences on the claim token, so a lease lost
                # during the image I/O no-ops here exactly as it does on
                # every other failure path.
                logger.error(
                    "Echo %s: Mastodon token rejected during image upload for item %s: %s",
                    echo["id"],
                    item["id"],
                    e,
                )
                return _fail_post(
                    posted_id,
                    claim_token,
                    echo["id"],
                    f"Mastodon token rejected: {e}",
                    permanent=True,
                )
            if uploaded and uploaded.get("id"):
                media_ids.append(str(uploaded["id"]))
                logger.info(
                    "Echo %s: uploaded image %s for item %s",
                    echo["id"],
                    uploaded["id"],
                    item["id"],
                )
            else:
                logger.warning(
                    "Echo %s: image upload failed for item %s, continuing without that image",
                    echo["id"],
                    item["id"],
                )
        if not media_ids and image_entries:
            logger.warning(
                "Echo %s: no images uploaded for item %s, posting text-only",
                echo["id"],
                item["id"],
            )

    # Re-validate claim ownership immediately before the post: the image
    # fetch, media upload and alt-text generation above are slow network I/O,
    # so the lease can lapse and another worker reclaim the row. Without this
    # both workers post and the loser's _update_post silently no-ops. The
    # Bluesky path has always done this.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "Mastodon"):
        return False

    try:
        result = post_status(
            instance=account["instance"],
            access_token=decrypt_secret(account["access_token"]),
            content=truncate(content, MASTODON_MAX_CHARS),
            visibility=echo["visibility"],
            sensitive=sensitive,
            spoiler_text=cw_text,
            media_ids=media_ids or None,
            # Deterministic key from (echo_id, item_id): a crash-then-reclaim
            # retry of this same logical post reuses the key, so Mastodon
            # returns the original status instead of creating a duplicate.
            idempotency_key=mastodon_idempotency_key(echo["id"], item["id"]),
            content_type="text/markdown" if _echo_use_markdown(echo) else None,
        )
    except MastodonAuthError as e:
        # Token rejected: retries cannot help until the user reconnects.
        logger.error("Echo %s: Mastodon token rejected: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Mastodon token rejected: {e}",
            permanent=True,
        )
    except (MastodonError, httpx.HTTPStatusError) as e:
        # _raise_for_status delegates every non-auth HTTP failure to
        # httpx's HTTPStatusError, so catch it here rather than in the
        # generic handler — the API's own error text belongs in the row's
        # error_message, not a flat "delivery failed".
        logger.exception("Echo %s: Mastodon post failed", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], f"Mastodon delivery failed: {e}")
    except Exception:
        logger.exception("Echo %s: Mastodon post failed unexpectedly", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], "Mastodon delivery failed")

    # The API returns the canonical permalink; persisting it gives the
    # history page a "view post" link exactly like the Bluesky and
    # micro.blog paths. Some instances omit `url`, and delivery still
    # succeeded, so a missing URL is not an error.
    post_url = ""
    raw_url = result.get("url") if isinstance(result, dict) else None
    if isinstance(raw_url, str) and raw_url:
        post_url = raw_url

    if post_url:
        # echo is a sqlite Row: bracket access can raise on a missing column.
        try:
            echo_visibility = echo["visibility"]
        except (KeyError, IndexError):
            echo_visibility = "public"
        _maybe_boost(account, echo, post_url, echo_visibility)

    return _finalize_success(posted_id, claim_token, echo["id"], post_url=post_url or None)


def _send_email_echo(
    echo,
    item: FeedItem,
    content: str,
    email_account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    # Check if this echo is in digest mode — if so, queue for later sending
    try:
        delivery_mode = echo["delivery_mode"]
    except (KeyError, IndexError):
        delivery_mode = "instant"

    if delivery_mode == "digest":
        return _queue_for_digest(echo, item, content, posted_id, claim_token)

    # Instant mode: send immediately
    account, fail_result = _destination_account(
        "email_accounts", email_account_id, echo, posted_id, claim_token, "Email account"
    )
    if account is None:
        return fail_result

    # Image embedding: mirroring the Mastodon path, attach_image fetches the
    # item's images (SSRF-validated, fallback-proxy aware) and embeds them
    # inline in the message. A failed fetch skips that image — the email
    # still goes out with the rendered text.
    images: list[dict] = []
    attach_image = _echo_attach_image(echo)
    if attach_image:
        total_bytes = 0
        for entry in _item_image_entries(item, limit=EMAIL_MAX_IMAGES):
            image_result = fetch_image(entry["url"])
            if not image_result:
                logger.info(
                    "Echo %s: image fetch failed for %s, skipping that image",
                    echo["id"], entry["url"],
                )
                continue
            img_bytes, img_type = image_result
            if total_bytes + len(img_bytes) > EMAIL_MAX_IMAGE_BYTES:
                logger.info(
                    "Echo %s: email image budget (%d bytes) reached, "
                    "skipping remaining images",
                    echo["id"], EMAIL_MAX_IMAGE_BYTES,
                )
                break
            total_bytes += len(img_bytes)
            images.append(
                {"data": img_bytes, "content_type": img_type, "alt": entry["alt"]}
            )

    # Re-validate claim ownership immediately before the send: email is a
    # one-way send too, and the image-fetch loop above is slow network I/O
    # that can outlast the lease, so the lease can lapse and another worker
    # reclaim the row. Checking here (right before send, not before the
    # fetch loop) matches the other senders and closes the window where a
    # slow fetch let both workers send.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "email"):
        return False

    try:
        send_email(
            to_email=account["email"],
            subject=truncate(
                item.get("title") or item.get("link") or "FeedEcho: New Post",
                200,
            ),
            body=content,
            user_id=echo["user_id"],
            images=images,
            render_html=_echo_render_html(echo),
        )
    except Exception:
        logger.exception("Echo %s: email delivery failed", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], "Email delivery failed")
    return _finalize_success(posted_id, claim_token, echo["id"])


def _bsky_session(account) -> dict:
    """Return a usable Bluesky session, caching JWTs in the account row.

    Resolves the account's PDS when unknown, reuses a cached access JWT while
    unexpired, refreshes via the refresh JWT when possible, and falls back to
    a fresh app-password login otherwise. All network I/O happens with no DB
    connection held; only the short cache write opens one.
    """
    handle = account["handle"]
    app_password = decrypt_secret(account["app_password"])
    pds = (account["pds"] or "").strip()
    did = (account["did"] or "").strip()
    access_jwt = decrypt_secret(account["access_jwt"] or "").strip()
    refresh_jwt = decrypt_secret(account["refresh_jwt"] or "").strip()
    now = _now()

    resolved_did = None
    if not pds:
        resolved_did, pds = resolve_pds(handle)

    if access_jwt:
        # Normalize before comparing: psycopg returns a datetime for the
        # TIMESTAMP column, sqlite returns the stored string. Comparing a
        # datetime against the _now() string raises TypeError, which the
        # caller's broad handler would report as "Bluesky session failed"
        # on every post after the first.
        expires_at = timestamp_str(account["session_expires_at"])
        if expires_at and expires_at > now:
            # Persist a late PDS/DID resolution without disturbing the session.
            if resolved_did:
                with get_db() as db:
                    db.execute(
                        "UPDATE bluesky_accounts SET did = ?, pds = ? WHERE id = ?",
                        (resolved_did, pds, account["id"]),
                    )
            return {
                "pds": pds,
                "did": did or resolved_did or "",
                "access_jwt": access_jwt,
            }

    session = None
    if refresh_jwt:
        try:
            session = refresh_session(pds, refresh_jwt)
        except BlueskyError:
            # Auth or network failure — fall through to a fresh login.
            session = None
    if session is None:
        session = create_session(pds, handle, app_password)

    with get_db() as db:
        db.execute(
            """
            UPDATE bluesky_accounts
               SET did = ?, pds = ?, access_jwt = ?, refresh_jwt = ?,
                   session_expires_at = ?
             WHERE id = ?
            """,
            (
                session["did"],
                pds,
                encrypt_secret(session["access_jwt"]),
                encrypt_secret(session["refresh_jwt"]),
                session_expiry(session["access_jwt"]),
                account["id"],
            ),
        )
    return {
        "pds": pds,
        "did": session["did"],
        "access_jwt": session["access_jwt"],
    }


def _bsky_reauth(account, session: dict) -> dict:
    """Re-login with the stored app password after a rejected session.

    Shared by the image-upload loop and the create_post retry so both heal
    the same way. Returns the fresh session dict with the new tokens
    persisted; raises BlueskyAccountGoneError when the account row is gone
    and BlueskyAuthError when the credentials themselves are dead (callers
    map both to a permanent failure — a plain BlueskyError from create_session,
    e.g. a PDS outage, stays a transient failure).
    """
    with get_db() as db:
        fresh = db.execute(
            "SELECT handle, app_password FROM bluesky_accounts WHERE id = ?",
            (account["id"],),
        ).fetchone()
    if not fresh:
        raise BlueskyAccountGoneError("Bluesky account was deleted during dispatch")
    refreshed = create_session(
        session["pds"], fresh["handle"], decrypt_secret(fresh["app_password"])
    )
    with get_db() as db:
        db.execute(
            """
            UPDATE bluesky_accounts
               SET did = ?, access_jwt = ?, refresh_jwt = ?,
                   session_expires_at = ?
             WHERE id = ?
            """,
            (
                refreshed["did"],
                encrypt_secret(refreshed["access_jwt"]),
                encrypt_secret(refreshed["refresh_jwt"]),
                session_expiry(refreshed["access_jwt"]),
                account["id"],
            ),
        )
    return {
        "pds": session["pds"],
        "did": refreshed["did"],
        "access_jwt": refreshed["access_jwt"],
    }


def _bsky_call_with_reauth(
    account,
    session: dict,
    echo,
    posted_id: int,
    claim_token: str,
    context: str,
    op,
    item=None,
) -> tuple[bool, dict, object]:
    """Run `op(session)`; on BlueskyAuthError, re-auth once and retry once.

    Shared by the image-upload loop and the create_post call so a fix to
    the recovery semantics (how many attempts, what counts as permanent)
    lands once instead of drifting between two hand-rolled copies. `op`
    performs the actual PDS call against the session it's given and
    returns its result, or raises BlueskyAuthError if that session was
    rejected; `context` is a short label ("post", "image upload") used in
    log messages.

    Returns (True, session, result) on success — result is whatever `op`
    returned (or mutated into shared state; the image-upload loop uses
    `op` for its side effects and ignores this slot). Returns (False,
    session, fail_result) when recovery could not complete: fail_result is
    already-persisted (via _fail_post) and the caller must `return
    fail_result` immediately. `item` (optional) adds the item id to the
    permanent-failure log lines, matching the neighboring image-loop logs.
    """

    def _give_up(reason: str) -> tuple[bool, dict, object]:
        return False, session, _fail_post(
            posted_id, claim_token, echo["id"], reason, permanent=True
        )

    try:
        return True, session, op(session)
    except BlueskyAuthError:
        logger.warning(
            "Echo %s: Bluesky session rejected during %s, re-authenticating",
            echo["id"],
            context,
        )
    except Exception:
        logger.exception("Echo %s: Bluesky %s failed", echo["id"], context)
        return False, session, _fail_post(
            posted_id, claim_token, echo["id"], "Bluesky delivery failed"
        )

    try:
        session = _bsky_reauth(account, session)
    except BlueskyAccountGoneError:
        if item is not None:
            logger.error(
                "Echo %s: Bluesky account deleted during %s for item %s; giving up",
                echo["id"],
                context,
                item["id"],
            )
        else:
            logger.error(
                "Echo %s: Bluesky account deleted during %s; giving up",
                echo["id"],
                context,
            )
        return _give_up("Bluesky account deleted")
    except BlueskyAuthError:
        if item is not None:
            logger.error(
                "Echo %s: Bluesky credentials rejected after re-auth during %s for item %s; giving up",
                echo["id"],
                context,
                item["id"],
            )
        else:
            logger.error(
                "Echo %s: Bluesky credentials rejected after re-auth during %s; giving up",
                echo["id"],
                context,
            )
        return _give_up("Bluesky credentials rejected")
    except Exception:
        logger.exception(
            "Echo %s: Bluesky re-auth failed during %s", echo["id"], context
        )
        return False, session, _fail_post(
            posted_id, claim_token, echo["id"], "Bluesky delivery failed"
        )

    try:
        return True, session, op(session)
    except BlueskyAuthError:
        if item is not None:
            logger.error(
                "Echo %s: Bluesky credentials rejected after re-auth during %s for item %s; giving up",
                echo["id"],
                context,
                item["id"],
            )
        else:
            logger.error(
                "Echo %s: Bluesky credentials rejected after re-auth during %s; giving up",
                echo["id"],
                context,
            )
        return _give_up("Bluesky credentials rejected")
    except Exception:
        logger.exception(
            "Echo %s: Bluesky %s failed after re-auth", echo["id"], context
        )
        return False, session, _fail_post(
            posted_id, claim_token, echo["id"], "Bluesky delivery failed"
        )


def _prepare_bluesky_image(
    echo, item: FeedItem, entry: dict
) -> tuple[bytes, str, str] | None:
    """Fetch, downscale, and resolve alt text for one image — no upload.

    Returns (img_bytes, img_type, alt_description) ready for upload_blob, or
    None to skip this image (dead link, unsupported type, still over the
    blob cap after a downscale attempt). This step touches no Bluesky
    session and can't raise BlueskyAuthError, so callers run it for every
    candidate image concurrently, ahead of the sequential, session-bound
    upload/retry loop. Oversized images are downscaled/re-encoded first
    (the 2026-09-09 glass.photo report: ~2.9 MB source JPEGs degraded every
    glass post to text-only under the old 1 MB cap).
    """
    try:
        image_result = fetch_image(entry["url"])
        if not image_result:
            logger.info(
                "Echo %s: image fetch failed for %s, skipping that image",
                echo["id"],
                entry["url"],
            )
            return None
        img_bytes, img_type = image_result
        # Oversized for Bluesky? Downscale/re-encode to fit. None = Pillow
        # unavailable or still too big — skip just this image.
        if len(img_bytes) > MAX_BLOB_BYTES:
            downscaled = images.downscale_image(img_bytes, img_type, MAX_BLOB_BYTES)
            if downscaled:
                img_bytes, img_type = downscaled
                logger.info(
                    "Echo %s: downscaled Bluesky image for item %s to %d bytes (%s)",
                    echo["id"],
                    item["id"],
                    len(img_bytes),
                    img_type,
                )
            else:
                logger.warning(
                    "Echo %s: Bluesky image still exceeds the %d byte blob limit after downscale attempt for item %s, skipping that image",
                    echo["id"],
                    MAX_BLOB_BYTES,
                    item["id"],
                )
                return None
        if img_type not in BLUESKY_IMAGE_TYPES or len(img_bytes) > MAX_BLOB_BYTES:
            logger.info(
                "Echo %s: image unsupported for Bluesky (type=%s, size=%d), skipping that image",
                echo["id"],
                img_type,
                len(img_bytes),
            )
            return None
        # Feed-provided alt text wins; AI is the fallback.
        alt_description = _resolve_alt_text(
            echo, item, entry.get("alt", ""), img_bytes, img_type
        ) or ""
        return img_bytes, img_type, alt_description
    except Exception:
        logger.warning(
            "Echo %s: Bluesky image pipeline failed for one image of item %s, skipping that image",
            echo["id"],
            item["id"],
            exc_info=True,
        )
        return None


def _prepare_bluesky_images(
    echo, item: FeedItem, entries: list[dict]
) -> list[tuple[bytes, str, str] | None]:
    """Run _prepare_bluesky_image for every entry concurrently.

    Fetch, downscale, and alt-text resolution are each independent,
    session-free network/CPU work per image — running them one at a time
    made a 4-image post pay for 4 sequential round-trips. Returns results
    aligned to `entries` by index (None where that entry was skipped), so
    the caller's upload loop keeps its existing sequential, index-based
    retry logic unchanged and just skips the None slots.
    """
    if not entries:
        return []
    if len(entries) == 1:
        return [_prepare_bluesky_image(echo, item, entries[0])]
    with ThreadPoolExecutor(max_workers=min(len(entries), BLUESKY_MAX_IMAGES)) as pool:
        futures = [
            pool.submit(_prepare_bluesky_image, echo, item, entry) for entry in entries
        ]
        return [future.result() for future in futures]


def _upload_bluesky_image(
    session: dict, echo, item: FeedItem, prepared: tuple[bytes, str, str]
) -> dict | None:
    """Upload one already-fetched, already-validated image blob.

    Returns {"blob", "alt"} on success, or None on an upload rejection
    (logs and lets the caller skip just this image). BlueskyAuthError
    propagates instead: a dead session is a post-level problem, not an
    image one.
    """
    img_bytes, img_type, alt_description = prepared
    try:
        blob = upload_blob(
            pds=session["pds"],
            access_jwt=session["access_jwt"],
            image_bytes=img_bytes,
            content_type=img_type,
        )
    except BlueskyAuthError:
        # A dead session is post-level, not image-level: propagate so the
        # caller fails the dispatch (fresh session on retry) instead of
        # silently publishing a degraded text-only post.
        raise
    except Exception:
        logger.warning(
            "Echo %s: Bluesky image upload failed for item %s, skipping that image",
            echo["id"],
            item["id"],
            exc_info=True,
        )
        return None
    if not blob:
        logger.warning(
            "Echo %s: Bluesky image upload failed for item %s, skipping that image",
            echo["id"],
            item["id"],
        )
        return None
    logger.info(
        "Echo %s: uploaded Bluesky image blob for item %s",
        echo["id"],
        item["id"],
    )
    return {"blob": blob, "alt": alt_description}


def _upload_bluesky_thumb(
    echo, item: FeedItem, image_url: str, session: dict
) -> dict | None:
    """Fetch, validate, and upload a link card thumbnail blob.

    Same pipeline as the image embeds (SSRF-validated fetch, downscale
    attempt, type allowlist) but under EXTERNAL_THUMB_MAX_BYTES: the
    app.bsky.embed.external lexicon caps the thumb at 1 MB (it did NOT get
    the 2 MB raise embed.images got), and an oversized blob would get the
    whole record rejected with a 400 — failing the post, not just the
    thumb. Returns the blob reference or None to render a text-only card.
    """
    prepared = fetch_image(image_url)
    if not prepared:
        return None
    img_bytes, img_type = prepared
    if len(img_bytes) > EXTERNAL_THUMB_MAX_BYTES:
        downscaled = images.downscale_image(img_bytes, img_type, EXTERNAL_THUMB_MAX_BYTES)
        if not downscaled:
            return None
        img_bytes, img_type = downscaled
    if img_type not in BLUESKY_IMAGE_TYPES or len(img_bytes) > EXTERNAL_THUMB_MAX_BYTES:
        return None
    return upload_blob(
        pds=session["pds"],
        access_jwt=session["access_jwt"],
        image_bytes=img_bytes,
        content_type=img_type,
    )


def _fetch_bluesky_link_meta(echo, item: FeedItem, uri: str) -> dict | None:
    """Fetch a link target's og: metadata. Session-free and best-effort.

    Runs once per dispatch, before the claim guard — the same placement as
    the image pipeline's prefetch — so a session re-auth retry reuses the
    cached metadata instead of re-fetching the page from origin.
    """
    try:
        meta = fetch_page_metadata(uri)
    except Exception:
        logger.warning(
            "Echo %s: Bluesky link card metadata fetch failed for item %s, "
            "posting without a card",
            echo["id"],
            item["id"],
            exc_info=True,
        )
        return None
    if meta is None:
        # fetch_page_metadata never raises and reports WHY it gave up at
        # DEBUG only (dead links are routine), so without this line a skipped
        # card is invisible at the default INFO level and an operator cannot
        # tell "card skipped" from "this build has no cards".
        logger.info(
            "Echo %s: no link card for item %s: could not read page metadata "
            "from %s (fetch failed, blocked, or not HTML; set "
            "FEEDECHO_LOG_LEVEL=DEBUG for the cause), posting without a card",
            echo["id"],
            item["id"],
            uri,
        )
    return meta


def _build_bluesky_link_card(
    echo, item: FeedItem, uri: str, meta: dict | None, session: dict
) -> dict | None:
    """Build an app.bsky.embed.external link card, or None.

    Bluesky renders link cards only when the posting client attaches an
    external embed (the AppView does not generate them for third-party
    records), so text-only echoes with a link attach one — the Echofeed
    behavior members expect. `meta` is the prefetched page metadata (see
    _fetch_bluesky_link_meta); only the thumb upload here touches the
    Bluesky session, which is why it lives inside the retried post unit.

    Best-effort decoration for everything the LINK can do to break the
    post (no metadata, a thumb that will not fetch or upload) — those
    degrade to a cardless or thumbless post. The one deliberate exception
    is BlueskyAuthError from the thumb upload: a dead session is a
    post-level problem, so it propagates and the caller's re-auth path
    retries the whole post (card included) with a fresh session, instead
    of shipping a thumbless card from a token that is about to fail
    create_post anyway. Pages without a title fall back to the hostname —
    an empty-title card renders as a blank container in clients, and the
    official app shows the host in the same situation.
    """
    if meta is None:
        return None
    try:
        thumb_blob = None
        image_url = meta.get("image")
        if image_url:
            try:
                thumb_blob = _upload_bluesky_thumb(echo, item, image_url, session)
            except BlueskyAuthError:
                raise
            except Exception:
                logger.warning(
                    "Echo %s: Bluesky link card thumb failed for item %s, "
                    "attaching a text-only card",
                    echo["id"],
                    item["id"],
                    exc_info=True,
                )
        title = (meta.get("title") or "").strip()
        if not title:
            title = urlparse(uri).netloc
        card = build_external_embed(
            uri=uri,
            title=title,
            description=meta.get("description") or "",
            thumb_blob=thumb_blob,
        )
        logger.info(
            "Echo %s: attached Bluesky link card for item %s (%s, %s)",
            echo["id"],
            item["id"],
            uri,
            "with thumbnail" if thumb_blob else "no thumbnail",
        )
        return card
    except BlueskyAuthError:
        # Session-level: let the caller's re-auth path retry the whole
        # post (card included) with a fresh session.
        raise
    except Exception:
        logger.warning(
            "Echo %s: Bluesky link card build failed for item %s, posting without one",
            echo["id"],
            item["id"],
            exc_info=True,
        )
        return None


def _send_bluesky(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    account, fail_result = _destination_account(
        "bluesky_accounts", account_id, echo, posted_id, claim_token, "Bluesky account"
    )
    if account is None:
        return fail_result

    # Content preparation is pure string work, but a bug here must not strand
    # the claimed row — finalize it as failed so the bounded retry owns it.
    try:
        text = truncate_graphemes(
            content or "", MAX_POST_GRAPHEMES, max_bytes=MAX_POST_BYTES
        )
        # Link spans from the rich render (content_html anchors) survive only
        # if fully inside the truncated text; partial spans are dropped. The
        # drip path (override content) has no spans and relies on bare-URL
        # detection alone.
        # Truncation always appends "…" as the final character — a span
        # reaching the last character covers the ellipsis and maps onto
        # text that was never part of the anchor, so it's dropped. Anchors
        # containing a literal "…" elsewhere keep their facet.
        limit = len(text) - 1 if len(text) < len(content or "") else len(text)
        # When middle truncation preserves a trailing URL (e.g. head + "… " + tail),
        # rich links beyond the head cut point no longer align with characters in text.
        # Restrict surviving rich links to those fully within the kept head prefix.
        if "… " in text and len(text) < len(content or ""):
            limit = min(limit, text.find("… "))
        rich_links = [
            (start, end, uri)
            for start, end, uri in (item.get("_rich_links") or [])
            if end <= limit
        ]
        facets = build_facets(text, extra_links=rich_links)
    except Exception:
        logger.exception("Echo %s: Bluesky content preparation failed", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], "Content preparation failed"
        )

    try:
        session = _bsky_session(account)
    except Exception:
        logger.exception("Echo %s: Bluesky session failed", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], "Bluesky session failed")

    # Image attachment: optional, up to BLUESKY_MAX_IMAGES images (Bluesky's
    # per-post embed cap), each with its own alt text. A failure on any
    # single image (fetch, downscale, type, upload) skips that image; only
    # when none survive does the post degrade to text-only.
    attach_image = _echo_attach_image(echo)

    image_entries_out: list[dict] = []
    if attach_image:
        image_entries = _item_image_entries(item, limit=BLUESKY_MAX_IMAGES)
        # Fetch/downscale/alt-text is session-free and independent per
        # image, so it runs concurrently up front; only the upload_blob
        # calls below stay sequential, since those are the ones that can
        # hit a rejected session and need the index-based retry.
        prepared = _prepare_bluesky_images(echo, item, image_entries)
        # Index-based resume: a per-image skip does not advance
        # image_entries_out, so the retry window must track the INPUT index —
        # slicing by output count would re-upload an already-uploaded image
        # as a duplicate after a re-auth.
        start_idx = 0

        def _upload_remaining(s: dict) -> None:
            nonlocal start_idx
            for idx in range(start_idx, len(image_entries)):
                if prepared[idx] is not None:
                    uploaded = _upload_bluesky_image(s, echo, item, prepared[idx])
                    if uploaded:
                        image_entries_out.append(uploaded)
                # Advance only after the entry has been handled (uploaded or
                # deliberately skipped); a BlueskyAuthError leaves the index
                # on the failed entry so the retry resumes there.
                start_idx = idx + 1

        ok, session, fail_result = _bsky_call_with_reauth(
            account,
            session,
            echo,
            posted_id,
            claim_token,
            "image upload",
            _upload_remaining,
            item=item,
        )
        if not ok:
            return fail_result
        if image_entries and not image_entries_out:
            logger.warning(
                "Echo %s: no images uploaded for item %s, posting text-only",
                echo["id"],
                item["id"],
            )

    embed = build_image_embed(image_entries_out) if image_entries_out else None

    # Link card prefetch: session-free page metadata, fetched ONCE per
    # dispatch before the claim guard (same placement as the image
    # pipeline's fetch phase) and cached, so a re-auth retry below rebuilds
    # the card without re-fetching the page from origin.
    link_uri = first_link_uri(facets) if embed is None else None
    link_meta = _fetch_bluesky_link_meta(echo, item, link_uri) if link_uri else None

    # Re-validate claim ownership immediately before the post: if the lease
    # lapsed and another worker reclaimed this row, posting would duplicate.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "Bluesky"):
        return False

    def _do_post(s: dict) -> dict:
        # Rebinding session also updates the DID post_url builds from below
        # (a re-login may have resolved it for the first time).
        post_embed = embed
        if post_embed is None and link_uri:
            # No image embed: attach a link card for the first link in the
            # post (rich anchor wins over a bare permalink, matching the
            # facet order). The thumb upload is here so a session that only
            # refreshes after create_post rejects it gets the card on the
            # retry too. Best-effort: _build_bluesky_link_card returns None
            # instead of failing the post (BlueskyAuthError excepted).
            post_embed = _build_bluesky_link_card(echo, item, link_uri, link_meta, s)
        return create_post(
            pds=s["pds"],
            access_jwt=s["access_jwt"],
            repo=s["did"],
            text=text,
            facets=facets or None,
            embed=post_embed,
        )

    ok, session, result = _bsky_call_with_reauth(
        account, session, echo, posted_id, claim_token, "post", _do_post
    )
    if not ok:
        return result

    # Persist the post URL for auditing/duplicate detection.
    post_url = ""
    uri = result.get("uri", "")
    if uri:
        rkey = uri.rsplit("/", 1)[-1]
        post_url = f"https://bsky.app/profile/{session['did']}/post/{rkey}"

    return _finalize_success(posted_id, claim_token, echo["id"], post_url=post_url)


def _send_microblog(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item to a micro.blog blog via Micropub.

    The stored token posts to the row's blog uid (mp-destination). When the
    echo attaches images, the feed item's first image URL is passed through
    as photo= (Micro.blog fetches and hosts it), with AI alt text in
    mp-photo-alt when the alt-text feature is enabled — matching the
    Mastodon/Bluesky degrade-to-text-only behavior on any image problem.
    """
    account, fail_result = _destination_account(
        "microblog_accounts", account_id, echo, posted_id, claim_token,
        "Micro.blog account", user_scoped=True,
    )
    if account is None:
        return fail_result

    # Image passthrough: Micro.blog fetches photo= itself, so unlike the
    # Mastodon/Bluesky paths there is no blob upload. Alt text still needs
    # the image bytes, so fetch here and degrade to photo-without-alt on
    # any failure. Content preparation is pure string work but must not
    # strand the claimed row either.
    photo_url = ""
    photo_alt = ""
    attach_image = _echo_attach_image(echo)

    if attach_image:
        image_url = item.get("image_url", "")
        if image_url:
            photo_url = image_url
            # Default precedence: feed-provided alt text wins; AI is the
            # fallback. With the feed's ai_alt_override flag set, AI
            # generation runs even with non-empty feed alt and wins on
            # success (feed alt remains the fallback when AI fails or is
            # disabled for the user).
            photo_alt = (item.get("image_alt") or "").strip()
            try:
                ai_override = _feed_ai_alt_override(_echo_feed_id(echo))
                if (ai_override or not photo_alt) and alt_text.is_enabled(
                    user_id=echo["user_id"]
                ):
                    image_result = fetch_image(image_url)
                    if image_result:
                        img_bytes, img_type = image_result
                        try:
                            generated = (
                                alt_text.generate_alt_text(
                                    img_bytes, img_type, user_id=echo["user_id"]
                                )
                                or ""
                            )
                            if generated:
                                if ai_override:
                                    logger.info(
                                        "Echo %s: AI alt text override wins for item %s (%d chars)",
                                        echo["id"], item["id"], len(generated),
                                    )
                                photo_alt = generated
                        except Exception:
                            # A failed override keeps the feed-provided
                            # alt rather than posting with nothing.
                            if ai_override:
                                logger.warning(
                                    "Echo %s: alt text generation failed for item %s; "
                                    "falling back to feed-provided alt",
                                    echo["id"],
                                    item["id"],
                                    exc_info=True,
                                )
                            else:
                                logger.warning(
                                    "Echo %s: alt text generation failed for item %s",
                                    echo["id"],
                                    item["id"],
                                    exc_info=True,
                                )
            except Exception:
                logger.warning(
                    "Echo %s: micro.blog alt-text pipeline failed for item %s",
                    echo["id"],
                    item["id"],
                    exc_info=True,
                )

    # Re-validate claim ownership immediately before the post: alt-text
    # generation above is slow network I/O, so the lease can lapse and
    # another worker reclaim the row. Same pattern as Mastodon/Bluesky.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "micro.blog"):
        return False

    try:
        result = microblog_create_post(
            token=decrypt_secret(account["token"]),
            content=content,
            destination=account["uid"],
            photo_url=photo_url,
            photo_alt=photo_alt,
        )
    except MicroblogAuthError as e:
        # Token rejected: retries cannot help until the user reconnects.
        logger.error("Echo %s: micro.blog token rejected: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Micro.blog token rejected: {e}",
            permanent=True,
        )
    except MicroblogError as e:
        logger.exception("Echo %s: micro.blog post failed", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], f"Micro.blog delivery failed: {e}")
    except Exception:
        logger.exception("Echo %s: micro.blog post failed unexpectedly", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], "Micro.blog delivery failed")

    # Persist the post URL for auditing/duplicate detection. Micropub
    # returns it in the Location header (captured as result["location"]).
    post_url = result.get("location", "") if isinstance(result, dict) else ""

    return _finalize_success(posted_id, claim_token, echo["id"], post_url=post_url)


def _send_matrix(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item into a Matrix room.

    The text event is the delivery: it decides success/failure. An attached
    image is a SECOND event (Matrix has no combined text+image message), so it
    is sent after the text and any image failure only logs — the item is
    already delivered and re-running the text would duplicate it.

    Transaction IDs are derived from the echo + item ID, so a retry of a send
    whose response was lost is de-duplicated by the homeserver instead of
    posting the item twice.
    """
    account, fail_result = _destination_account(
        "matrix_accounts", account_id, echo, posted_id, claim_token,
        "Matrix account", user_scoped=True,
    )
    if account is None:
        return fail_result

    base_url = account["base_url"] or account["homeserver"]
    access_token = decrypt_secret(account["access_token"])
    room_id = account["room_id"]

    # Image work happens BEFORE the text event so its slow I/O (fetch, alt
    # text, upload) does not sit between the text send and the claim's final
    # state. Any failure degrades to text-only, as on the other destinations.
    attach_image = _echo_attach_image(echo)

    mxc_uri = ""
    image_body = ""
    image_type = ""
    image_size = 0
    if attach_image:
        image_url = item.get("image_url", "")
        if image_url:
            try:
                image_result = fetch_image(image_url)
                if image_result:
                    img_bytes, img_type = image_result
                    if (
                        img_type in MATRIX_IMAGE_TYPES
                        and len(img_bytes) <= MATRIX_MAX_UPLOAD_BYTES
                    ):
                        # Feed-provided alt text wins; AI is the fallback.
                        alt_description = _resolve_alt_text(
                            echo,
                            item,
                            (item.get("image_alt") or "").strip(),
                            img_bytes,
                            img_type,
                        ) or ""
                        mxc_uri = matrix_upload_media(
                            base_url,
                            access_token,
                            img_bytes,
                            img_type,
                            filename=_matrix_image_filename(img_type),
                        )
                        image_body = alt_description or _matrix_image_filename(img_type)
                        image_type = img_type
                        image_size = len(img_bytes)
                    else:
                        logger.info(
                            "Echo %s: image unsupported for Matrix (type=%s, size=%d), posting text-only",
                            echo["id"],
                            img_type,
                            len(img_bytes),
                        )
                else:
                    logger.info(
                        "Echo %s: image fetch failed for item %s, posting text-only",
                        echo["id"],
                        item["id"],
                    )
            except Exception:
                logger.warning(
                    "Echo %s: Matrix image pipeline failed for item %s, posting text-only",
                    echo["id"],
                    item["id"],
                    exc_info=True,
                )
                mxc_uri = ""

    # Re-validate claim ownership immediately before the post: the image
    # pipeline above is slow network I/O, so the lease can lapse and another
    # worker reclaim the row. Same pattern as the other senders.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "Matrix"):
        return False

    try:
        # render_html echoes carry markup: the rendered output becomes the
        # formatted_body (re-sanitized at the send boundary — the template
        # may also interpolate raw-text variables) and the plain fallback is
        # the structure-preserving text conversion. Markup-only items (e.g.
        # a bare linked image) convert to an empty string, so the rendered
        # output itself backs the plain body up; Plain echoes keep the
        # legacy escape-and-linkify behavior.
        if _echo_render_html(echo):
            plain = html_to_text(content).strip() or content
            event_id = matrix_send_message(
                base_url,
                access_token,
                room_id,
                plain,
                matrix_transaction_id(echo["id"], item["id"]),
                formatted=sanitize_html(content),
            )
        else:
            event_id = matrix_send_message(
                base_url,
                access_token,
                room_id,
                content,
                matrix_transaction_id(echo["id"], item["id"]),
            )
    except MatrixAuthError as e:
        # Token rejected: retries cannot help until the user reconnects.
        logger.error("Echo %s: Matrix token rejected: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Matrix token rejected: {e}",
            permanent=True,
        )
    except MatrixPermissionError as e:
        # Kicked from the room, or no permission to post in it: also permanent
        # until the user fixes it on the Matrix side.
        logger.error("Echo %s: Matrix refused the post: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Matrix refused the post: {e}",
            permanent=True,
        )
    except MatrixError as e:
        logger.exception("Echo %s: Matrix post failed", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], f"Matrix delivery failed: {e}")
    except Exception:
        logger.exception("Echo %s: Matrix post failed unexpectedly", echo["id"])
        return _fail_post(posted_id, claim_token, echo["id"], "Matrix delivery failed")

    if mxc_uri:
        # The text already landed; a failed image must not fail the item.
        try:
            matrix_send_image(
                base_url,
                access_token,
                room_id,
                mxc_uri,
                image_body,
                image_type,
                image_size,
                matrix_transaction_id(echo["id"], item["id"], suffix="img"),
            )
        except Exception:
            logger.warning(
                "Echo %s: Matrix image event failed for item %s (text was delivered)",
                echo["id"],
                item["id"],
                exc_info=True,
            )

    return _finalize_success(
        posted_id,
        claim_token,
        echo["id"],
        post_url=matrix_permalink(room_id, event_id),
    )


def _matrix_image_filename(content_type: str) -> str:
    """A filename for the media upload, derived from the image type.

    Matrix clients show this when an image has no description, and some use
    the extension to decide how to render it.
    """
    ext = (content_type or "").rsplit("/", 1)[-1].lower()
    if ext == "jpeg":
        ext = "jpg"
    if not ext.isalnum():
        ext = "img"
    return f"feed-image.{ext}"


def _send_discord(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item to a Discord channel via its webhook.

    The message is one POST: ``content`` is the rendered template, and when
    image attachments are on and the item has one, a single embed carries the
    title, link, and image URL. Discord fetches the embed image itself, so
    there is no download/upload step on our side — an unusable image simply
    renders as text + link and the delivery still succeeds.

    A deleted or revoked webhook (401/404) is permanent until the user
    reconnects; rate limits and 5xx ride the transient retry pipeline.
    """
    account, fail_result = _destination_account(
        "discord_accounts", account_id, echo, posted_id, claim_token,
        "Discord account", user_scoped=True,
    )
    if account is None:
        return fail_result

    webhook_url = decrypt_secret(account["webhook_url"])

    attach_image = _echo_attach_image(echo)

    embed = None
    if attach_image:
        embed = discord_build_embed(
            item.get("title", ""),
            item.get("link", ""),
            item.get("image_url", ""),
        )

    # Discord work has no slow image pipeline that could lapse the lease, but
    # the claim is still re-validated so this path keeps the same contract as
    # the other senders.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "Discord"):
        return False

    try:
        discord_send_webhook(webhook_url, content, embed)
    except (DiscordAuthError, DiscordNotFoundError, DiscordBadRequestError) as e:
        # Revoked/deleted webhooks and rejected payloads cannot heal on
        # retry: permanent until the user reconnects or fixes the template.
        logger.error("Echo %s: Discord delivery refused: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Discord delivery refused: {e}",
            permanent=True,
        )
    except DiscordError as e:
        logger.exception("Echo %s: Discord post failed", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], f"Discord delivery failed: {e}"
        )
    except Exception:
        logger.exception("Echo %s: Discord post failed unexpectedly", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], "Discord delivery failed"
        )

    # Discord replies 204 with no message id (and a jump link would need the
    # guild id, which webhooks do not expose), so there is no post_url.
    return _finalize_success(posted_id, claim_token, echo["id"], post_url="")


def _send_telegram(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item to a Telegram chat via the Bot API.

    Templates without {{ content_html }} send with no parse_mode (Telegram
    auto-links URLs and hashtags; raw prose can never break entity parsing).
    Templates embedding {{ content_html }} send as parse_mode=HTML after the
    sanitized HTML is reduced to Telegram's supported tag subset.

    When image attachments are on and the item has one, the message goes out
    as sendPhoto with the text as caption (1024-char cap); a failed image
    fetch degrades to a text-only message. A bad token (401) is permanent
    until the user reconnects; rate limits and 5xx ride the transient retry
    pipeline.
    """
    account, fail_result = _destination_account(
        "telegram_accounts", account_id, echo, posted_id, claim_token,
        "Telegram account", user_scoped=True,
    )
    if account is None:
        return fail_result

    bot_token = decrypt_secret(account["bot_token"])
    chat_id = account["chat_id"]
    rich = "content_html" in (echo["template"] or "")

    attach_image = _echo_attach_image(echo)
    photo: tuple[bytes, str] | None = None
    if attach_image:
        entries = _item_image_entries(item, limit=1)
        if entries:
            image_result = fetch_image(entries[0]["url"])
            if image_result:
                photo = image_result

    # Telegram work has no slow multi-image pipeline that could lapse the
    # lease, but the claim is still re-validated so this path keeps the same
    # contract as the other senders.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "Telegram"):
        return False

    try:
        if photo is not None:
            caption, caption_mode = telegram_build_message(
                content or "", rich=rich, cap=MAX_CAPTION_CHARS
            )
            result = telegram_send_photo(
                bot_token, chat_id, photo[0], photo[1],
                caption=caption, parse_mode=caption_mode,
            )
        else:
            text, parse_mode = telegram_build_message(
                content or "", rich=rich, cap=MAX_MESSAGE_CHARS
            )
            result = telegram_send_message(bot_token, chat_id, text, parse_mode)
    except (TelegramAuthError, TelegramNotFoundError, TelegramBadRequestError) as e:
        # Bad tokens, chats the bot is no longer in, and rejected payloads
        # cannot heal on retry: permanent until the user reconnects or
        # fixes the template.
        logger.error("Echo %s: Telegram delivery refused: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Telegram delivery refused: {e}",
            permanent=True,
        )
    except TelegramError as e:
        logger.exception("Echo %s: Telegram post failed", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], f"Telegram delivery failed: {e}"
        )
    except Exception:
        logger.exception("Echo %s: Telegram post failed unexpectedly", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], "Telegram delivery failed"
        )

    # Public-channel messages get a t.me link; private chats/groups do not.
    return _finalize_success(
        posted_id, claim_token, echo["id"], post_url=telegram_message_url(result)
    )


def _send_webhook(
    echo,
    item: FeedItem,
    content: str,
    feed_name: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item to a generic webhook endpoint.

    The payload is one flat JSON object: ``text`` (the rendered template)
    plus every parsed item field, so receivers can map whatever shape they
    need. Auth failures (401/403), vanished endpoints (404/410), and rejected
    payloads (400/422) are permanent until the user fixes the account; rate
    limits and 5xx ride the transient retry pipeline.

    There is no per-message URL for a generic webhook, so post_url is "".
    """
    account, fail_result = _destination_account(
        "webhook_accounts", account_id, echo, posted_id, claim_token,
        "Webhook account", user_scoped=True,
    )
    if account is None:
        return fail_result

    headers = load_headers(account["headers"])
    body_template = (account["body_template"] if "body_template" in account.keys() else "") or ""
    if body_template:
        # Custom JSON body: the rendered template IS the payload. A render or
        # parse failure is a stored-config problem — permanent, like a
        # rejected payload, because retrying cannot change the template.
        try:
            payload = webhook_build_body_payload(body_template, item, feed_name=feed_name)
        except WebhookRejectedError as e:
            logger.error("Echo %s: webhook body template failed: %s", echo["id"], e)
            return _fail_post(
                posted_id,
                claim_token,
                echo["id"],
                f"Webhook delivery refused: {e}",
                permanent=True,
            )
    else:
        payload = webhook_build_payload(item, content, feed_name=feed_name)

    # Same contract as the other senders: re-validate the claim immediately
    # before the irreversible step.
    if not _guard_claim(posted_id, claim_token, echo["id"], item["id"], "webhook"):
        return False

    try:
        webhook_send_webhook(account["url"], headers, payload)
    except (WebhookAuthError, WebhookNotFoundError, WebhookRejectedError) as e:
        # Wrong credentials, a deleted endpoint, or a rejected payload shape
        # cannot heal on retry: permanent until the user reconnects.
        logger.error("Echo %s: webhook delivery refused: %s", echo["id"], e)
        return _fail_post(
            posted_id,
            claim_token,
            echo["id"],
            f"Webhook delivery refused: {e}",
            permanent=True,
        )
    except WebhookError as e:
        logger.exception("Echo %s: webhook post failed", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], f"Webhook delivery failed: {e}"
        )
    except Exception:
        logger.exception("Echo %s: webhook post failed unexpectedly", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], "Webhook delivery failed"
        )

    return _finalize_success(posted_id, claim_token, echo["id"], post_url="")


def _send_github(
    echo,
    item: FeedItem,
    content: str,
    account_id: int,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Dispatch one rendered item to a GitHub repository as a markdown file.

    The echo template IS the file body; the per-echo github_path template
    renders to the repo-relative file path (overwriting an existing file at
    that path via the contents API sha, so re-echoes and corrections land),
    and github_commit renders to the commit message. Auth failures (401,
    non-rate-limit 403), vanished repo/branch (404), and refused writes
    (409/422, including the size cap) are permanent until the user fixes the
    account; rate limits and 5xx ride the transient retry pipeline.

    Success records the file's html_url as post_url, so history links
    straight to the file on GitHub.
    """
    account, fail_result = _destination_account(
        "github_accounts", account_id, echo, posted_id, claim_token,
        "GitHub account", user_scoped=True,
    )
    if account is None:
        return bool(fail_result)

    try:
        token = decrypt_secret(account["token"])
    except ValueError:
        logger.exception(
            "Echo %s: github credential decryption failed", echo["id"]
        )
        return _fail_post(
            posted_id, claim_token, echo["id"],
            "GitHub credential could not be decrypted", permanent=True,
        )

    # feed_name resolves here (rather than through the registry signature,
    # like _send_webhook's special case) so every dispatch flow — fresh
    # items, drip flush, queue post-now — gets the same value.
    feed_name = ""
    with get_db() as db:
        feed = db.execute(
            "SELECT name FROM feeds WHERE id = ?", (echo["feed_id"],)
        ).fetchone()
    if feed is not None:
        feed_name = feed["name"] or ""

    # Render failures and the size cap are PERMANENT: a broken stored
    # template can never heal on retry, and truncating a backup file
    # silently corrupts it. The cap is checked BEFORE any network call.
    try:
        path = github_render_path(echo["github_path"], item, feed_name)
        message = github_build_commit_message(
            echo["github_commit"] if "github_commit" in echo.keys() else "",
            item, feed_name, path=path,
        )
    except Exception as e:
        logger.error(
            "Echo %s: github path/commit template failed: %s", echo["id"], e
        )
        return _fail_post(
            posted_id, claim_token, echo["id"],
            f"GitHub path or commit template failed: {e}", permanent=True,
        )
    if len(base64.b64encode(content.encode("utf-8"))) > GITHUB_MAX_FILE_BYTES:
        return _fail_post(
            posted_id, claim_token, echo["id"],
            f"GitHub file is over the {GITHUB_MAX_FILE_BYTES}-byte size cap",
            permanent=True,
        )

    # send_file owns the two-step round trip: the contents-API GET (idempotent,
    # it only reads the blob sha, so it may run before the claim guard), then
    # the claim guard, then the irreversible PUT. Keeping the guard inside
    # send_file means the shipped dispatch path is the path the tests drive.
    def _claim_guard() -> bool:
        return _guard_claim(
            posted_id, claim_token, echo["id"], item["id"], "github"
        )

    try:
        result = github_send_file(
            token, account["repo"], account["branch"], path, content, message,
            committer=github_build_committer(
                account["committer_name"], account["committer_email"]
            ),
            guard=_claim_guard,
        )
    except GitHubRateLimitError as e:
        return _fail_post(
            posted_id, claim_token, echo["id"], f"GitHub delivery rate-limited: {e}"
        )
    except (GitHubAuthError, GitHubNotFoundError, GitHubRejectedError) as e:
        logger.error("Echo %s: github delivery refused: %s", echo["id"], e)
        return _fail_post(
            posted_id, claim_token, echo["id"],
            f"GitHub delivery refused: {e}", permanent=True,
        )
    except GitHubError as e:
        logger.exception("Echo %s: github delivery failed", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], f"GitHub delivery failed: {e}"
        )
    except Exception:
        logger.exception("Echo %s: github delivery failed unexpectedly", echo["id"])
        return _fail_post(
            posted_id, claim_token, echo["id"], "GitHub delivery failed"
        )

    if result is None:
        # The claim lapsed between the GET and the PUT; another worker owns
        # the row, so the write was skipped rather than duplicated.
        return False

    return _finalize_success(
        posted_id, claim_token, echo["id"], post_url=result["html_url"]
    )


# Strategy registry: destination_type -> sender. All eight share the
# (echo, item, content, account_id, posted_id, claim_token) -> bool shape
# except _send_webhook, which additionally needs feed_name for its payload
# template — handled as a special case in _render_and_dispatch rather than
# forcing an unused parameter onto the other seven. Replaces an 8-arm
# if/elif chain that was hand-copied in at least four other places
# (add_echo/edit_echo's destination-field resolution, echoes.html's
# destination-type badge, and the two account-table registries in app.py
# and import_export.py) — this is the one place a destination TYPE maps to
# posting BEHAVIOR, as opposed to a table name or form field.
_DESTINATION_HANDLERS: dict[str, Callable[[dict, FeedItem, str, int, int, str], bool]] = {
    "mastodon": _send_mastodon,
    "email": _send_email_echo,
    "bluesky": _send_bluesky,
    "microblog": _send_microblog,
    "matrix": _send_matrix,
    "discord": _send_discord,
    "telegram": _send_telegram,
    "github": _send_github,
}


def _queue_for_digest(
    echo,
    item: FeedItem,
    content: str,
    posted_id: int,
    claim_token: str,
) -> bool:
    """Queue an item for digest delivery instead of sending immediately.

    Inserts into digest_items and marks the posted_item as 'queued' so the
    cursor advances and the item isn't retried. The digest flush job sends
    all queued items for this echo as one email.
    """
    echo_id = echo["id"]
    item_id = item["id"]

    with get_db() as db:
        db.execute(
            """INSERT INTO digest_items (echo_id, item_id, item_title, item_url, rendered_content)
               VALUES (?, ?, ?, ?, ?)
               ON CONFLICT(echo_id, item_id) DO NOTHING""",
            (echo_id, item_id, item.get("title", ""), item.get("link", ""), content),
        )

    # Mark the posted_item as 'queued' — the digest job will finalize it.
    # No record_success here: queueing is a hold, not a delivery, and clearing
    # the alert state would send a spurious "recovered" email and let a
    # genuinely broken SMTP path keep failing silently until the threshold
    # re-accumulates. flush_digests records the success once the mail is sent.
    # (_queue_for_drip already worked this way; see the note there.)
    ok = _update_post(posted_id, claim_token, "queued")
    if ok:
        logger.info(
            "Echo %s: item %s queued for digest delivery",
            echo_id,
            item_id,
        )
    return ok


def _requeue_drip_release_error(
    echo_id: int,
    item_id: str,
    item_json: str,
    attempts: int,
    posted_id: int,
    claim_token: str,
    exc: Exception,
) -> None:
    """Requeue a drip release whose dispatch raised an unexpected exception.

    Unlike _requeue_drip_failure, which runs after a completed dispatch and
    can rely on the row being 'failed', an escaped exception leaves the
    posted row 'pending' under this worker's claim. Requeueing must
    therefore clear that claim, and it must clear it in one transaction
    with the drip_items insert: if the row were requeued but the insert
    failed, the next flush would re-release the item and double-insert the
    queue row; if the insert landed but the row stayed pending, the item
    would sit unclaimable for the reclaim window while the queue row is
    dead (the queue join only matches status 'queued').

    The UPDATE fences on THIS worker's claim_token, not just 'pending':
    if the reclaim sweep re-claimed the row while the dispatch was stuck,
    the token has changed and the new owner's claim must not be clobbered
    (the queue row was never deleted on this path, so there is nothing to
    restore either).

    'attempts' is deliberately NOT incremented here. This path is an
    infrastructure fault, not a delivery failure; record_failure still
    fires so a persistently faulting release reaches the notify threshold
    instead of retrying silently every flush.
    """
    with get_db() as db:
        result = db.execute(
            """UPDATE posted_items
                  SET status = 'queued',
                      claimed_at = NULL,
                      claim_token = NULL,
                      next_retry_at = NULL,
                      posted_at = ?
                WHERE id = ? AND status = 'pending' AND claim_token = ?""",
            (_now(), posted_id, claim_token),
        )
        if result.rowcount != 1:
            # The claim was lost (another worker or the reclaim sweep owns
            # the row now); their outcome wins and the queue row survives.
            return
        db.execute(
            """INSERT INTO drip_items (echo_id, item_id, item_json, attempts)
               VALUES (?, ?, ?, ?)
               ON CONFLICT(echo_id, item_id) DO UPDATE SET attempts = excluded.attempts""",
            (echo_id, item_id, item_json, attempts),
        )
    record_failure(echo_id)
    logger.error(
        "Echo %s: drip release raised %s: %s; item %s requeued (attempt count held at %d)",
        echo_id,
        type(exc).__name__,
        exc,
        item_id,
        attempts,
    )


def _requeue_drip_failure(
    echo_id: int,
    item_id: str,
    item_json: str,
    attempts: int,
    posted_id: int,
) -> None:
    """Return a failed drip release to the queue, or gave_up at the cap.

    Retries run from the stored payload, so items that have rotated out
    of the feed are never lost the way a feed-based retry would lose
    them. Release attempts are counted on the drip_items row.
    """
    attempts += 1
    cap = max_attempts(echo_id)
    if cap > 0 and attempts >= cap:
        with get_db() as db:
            db.execute(
                """UPDATE posted_items
                      SET status = 'gave_up',
                          error_message = ?,
                          claimed_at = NULL,
                          claim_token = NULL,
                          posted_at = ?
                    WHERE id = ?""",
                (
                    f"Drip release gave up after {attempts} attempts",
                    _now(),
                    posted_id,
                ),
            )
        record_failure(echo_id)
        logger.error(
            "Echo %s: drip release gave up for item %s after %s attempts",
            echo_id,
            item_id,
            attempts,
        )
        return

    with get_db() as db:
        result = db.execute(
            """UPDATE posted_items
                  SET status = 'queued',
                      claimed_at = NULL,
                      claim_token = NULL,
                      next_retry_at = NULL,
                      posted_at = ?
                WHERE id = ? AND status = 'failed'""",
            (_now(), posted_id),
        )
        if result.rowcount != 1:
            # The row moved on (e.g. claimed by the retry sweep); the
            # normal path owns it now.
            return
        db.execute(
            """INSERT INTO drip_items (echo_id, item_id, item_json, attempts)
               VALUES (?, ?, ?, ?)
               ON CONFLICT(echo_id, item_id) DO UPDATE SET attempts = excluded.attempts""",
            (echo_id, item_id, item_json, attempts),
        )
    logger.info(
        "Echo %s: dripped item %s release failed; re-queued (attempt %d)",
        echo_id,
        item_id,
        attempts,
    )


def _discard_drip_backlog(echo_id: int, reason: str) -> None:
    """Finalize an echo's queued drip rows as gave_up and clear the queue.

    A disabled or deleted echo must not hold items indefinitely: on
    re-enable they would dump as stale posts, and the queue cap would
    keep silently dropping new items meanwhile. History records what was
    discarded and why.
    """
    with get_db() as db:
        db.execute(
            """UPDATE posted_items
                  SET status = 'gave_up',
                      error_message = ?,
                      claimed_at = NULL,
                      claim_token = NULL,
                      posted_at = ?
                WHERE echo_id = ? AND status = 'queued'""",
            (reason, _now(), echo_id),
        )
        db.execute("DELETE FROM drip_items WHERE echo_id = ?", (echo_id,))
    logger.info("Echo %s: %s", echo_id, reason)


def flush_queue() -> None:
    """Dispatch due items from the user-scheduled curation queue.

    Runs every 5 minutes wrapped in a database job lease. Claims due posts
    atomically, materializes a one-shot echo, dispatches with the stored content
    verbatim via process_echo, and updates status, posted_item_id, and error.
    """
    if not _acquire_job_lease("flush_queue", FLUSH_QUEUE_LEASE_SECONDS):
        return
    try:
        _flush_queue()
    finally:
        _release_job_lease("flush_queue")


def _flush_queue() -> None:
    now_ts = _now()
    # Re-claim rows stuck in 'sending' past a 15-minute TTL (crash recovery)
    ttl_expired = (datetime.now(timezone.utc) - timedelta(minutes=15)).strftime("%Y-%m-%d %H:%M:%S")
    with get_db() as db:
        db.execute(
            """
            UPDATE queued_posts
               SET status = 'queued', claim_token = NULL, claimed_at = NULL
             WHERE status = 'sending' AND claimed_at <= ?
            """,
            (ttl_expired,),
        )
        if settings.MULTI:
            due = db.execute(
                """
                SELECT q.id FROM queued_posts q
                  JOIN users u ON u.id = q.user_id
                 WHERE q.status = 'queued' AND q.scheduled_at <= ?
                   AND u.suspended = 0
                   AND NOT (u.plan = 'trial' AND u.trial_ends_at IS NOT NULL
                            AND u.trial_ends_at <= ?)
                 ORDER BY q.scheduled_at ASC
                 LIMIT 50
                """,
                (now_ts, now_ts),
            ).fetchall()
        else:
            due = db.execute(
                """
                SELECT id FROM queued_posts
                 WHERE status = 'queued' AND scheduled_at <= ?
                 ORDER BY scheduled_at ASC
                 LIMIT 50
                """,
                (now_ts,),
            ).fetchall()

    # Pre-compute per-user hourly post counts once per tick (hoisted out of the loop)
    user_hourly_counts: dict[int, int] = {}
    user_caps: dict[int, int] = {}
    if settings.MULTI:
        hour_ago = (datetime.now(timezone.utc) - timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")
        with get_db() as db:
            counts = db.execute(
                """
                SELECT e.user_id AS uid, COUNT(*) AS c
                  FROM posted_items pi JOIN echoes e ON pi.echo_id = e.id
                 WHERE pi.status = 'success' AND pi.posted_at >= ?
                 GROUP BY e.user_id
                """,
                (hour_ago,),
            ).fetchall()
            for r in counts:
                user_hourly_counts[r["uid"]] = r["c"]

    for r in due:
        qp_id = r["id"]
        claim_token = secrets.token_urlsafe(16)
        claim_time = _now()
        with get_db() as db:
            claimed = db.execute(
                """
                UPDATE queued_posts
                   SET status = 'sending', claim_token = ?, claimed_at = ?
                 WHERE id = ? AND status = 'queued' AND scheduled_at <= ?
                """,
                (claim_token, claim_time, qp_id, now_ts),
            )
            # In sqlite rowcount indicates updated rows; in PG rowcount also indicates it
            if claimed.rowcount != 1:
                continue
            row = db.execute(
                """
                SELECT q.*, f.name AS feed_name, f.deleted_at AS feed_deleted_at
                  FROM queued_posts q
                  LEFT JOIN feeds f ON q.feed_id = f.id
                 WHERE q.id = ?
                """,
                (qp_id,),
            ).fetchone()

        if not row:
            continue

        # Never send from a feed the user deleted (the feed's stored items
        # are purged on delete, so the item lookup below would come back
        # empty and dispatch a blank post). Finalize with a reason so the
        # row shows what happened instead of sitting stuck or sending.
        if row["feed_deleted_at"] is not None:
            with get_db() as db:
                db.execute(
                    "UPDATE queued_posts SET status = 'failed',"
                    " error_message = 'Feed deleted'"
                    " WHERE id = ? AND status = 'sending'",
                    (qp_id,),
                )
            logger.info(
                "Queue flush: post %s finalized — its feed was deleted", qp_id
            )
            continue

        # Enforce max_posts_per_hour for the queue (multi mode only, using hoisted counts)
        if settings.MULTI:
            uid = row["user_id"]
            if uid not in user_caps:
                with get_db() as db:
                    u = db.execute("SELECT plan FROM users WHERE id = ?", (uid,)).fetchone()
                    user_caps[uid] = plans.posts_per_hour_cap(u["plan"]) if u else 0
            cap = user_caps.get(uid, 0)
            if cap and user_hourly_counts.get(uid, 0) >= cap:
                logger.info(
                    "Queue flush: user %s hit posts_per_hour cap (%d/%d), skipping",
                    uid, user_hourly_counts.get(uid, 0), cap,
                )
                # Release the claim back to queued for next tick
                with get_db() as db:
                    db.execute(
                        "UPDATE queued_posts SET status = 'queued', claim_token = NULL, claimed_at = NULL"
                        " WHERE id = ? AND claim_token = ?",
                        (qp_id, claim_token),
                    )
                continue

        # Reuse the echo_id stored at enqueue time (idempotency key for crash recovery)
        echo_id = row["echo_id"]
        with get_db() as db:
            echo = db.execute("SELECT * FROM echoes WHERE id = ?", (echo_id,)).fetchone() if echo_id else None

        # Fallback: create echo if missing (e.g. pre-migration rows without echo_id)
        if not echo:
            now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
            with get_db() as db:
                new_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,
                        row["user_id"],
                    ),
                ).fetchone()["id"]
                echo = db.execute("SELECT * FROM echoes WHERE id = ?", (new_echo_id,)).fetchone()
            # Persist the echo_id so future retries reuse it
            with get_db() as db:
                db.execute(
                    "UPDATE queued_posts SET echo_id = ? WHERE id = ?",
                    (new_echo_id, qp_id),
                )
            echo_id = new_echo_id  # update the variable so the status lookup uses the right 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 feed_item_id is still present, load any existing images/metadata.
        # A missing row is NOT fatal: retention pruning removes feed_items
        # routinely, and the queue row is self-sufficient (it carries its own
        # materialized content).
        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 "",
                    })

        # Race insurance: the feed may have been deleted between the claim
        # SELECT above and dispatch (feed deletion finalizes 'queued' rows,
        # but this row is 'sending'). Re-check immediately before sending so
        # nothing ever posts from a deleted feed.
        with get_db() as db:
            feed_state = db.execute(
                "SELECT deleted_at FROM feeds WHERE id = ?", (row["feed_id"],)
            ).fetchone()
        if feed_state is None or feed_state["deleted_at"] is not None:
            with get_db() as db:
                db.execute(
                    "UPDATE queued_posts SET status = 'failed',"
                    " error_message = 'Feed deleted'"
                    " WHERE id = ? AND status = 'sending'",
                    (qp_id,),
                )
            continue

        try:
            ok = process_echo(
                echo,
                item,
                feed_name=row["feed_name"] or "",
                override_content=row["content"],
            )

            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

                new_attempt = row["attempt_count"] + 1
                if final_status == "failed":
                    # Auto-retry with backoff: up to 3 attempts, then terminal.
                    # On non-terminal attempts, clear the stale posted_items row
                    # so _claim_post can create a fresh one on the next try.
                    # On the terminal attempt, leave the posted_items row intact
                    # so the failure is visible in /history (audit trail).
                    if new_attempt < 3:
                        if pi:
                            db.execute("DELETE FROM posted_items WHERE id = ?", (pi["id"],))
                        backoff_minutes = new_attempt * 10
                        retry_at = (datetime.now(timezone.utc) + timedelta(minutes=backoff_minutes)).strftime("%Y-%m-%d %H:%M:%S")
                        db.execute(
                            "UPDATE queued_posts SET status = 'queued', error_message = ?,"
                            " attempt_count = ?, scheduled_at = ?, claim_token = NULL, claimed_at = NULL"
                            " WHERE id = ? AND claim_token = ?",
                            (err_msg, new_attempt, retry_at, qp_id, claim_token),
                        )
                    else:
                        # Terminal: keep the posted_items row for history, don't write posted_item_id
                        # since the row's status already records the failure with the error.
                        db.execute(
                            "UPDATE queued_posts SET status = 'failed', error_message = ?,"
                            " attempt_count = ? WHERE id = ? AND claim_token = ?",
                            (err_msg, new_attempt, qp_id, claim_token),
                        )
                else:
                    db.execute(
                        "UPDATE queued_posts SET status = 'sent', posted_item_id = ?, error_message = NULL,"
                        " attempt_count = ? WHERE id = ? AND claim_token = ?",
                        (pi_id, new_attempt, qp_id, claim_token),
                    )
                    # Increment the in-memory hourly counter so the cap
                    # enforcement sees this dispatch within the same tick.
                    user_hourly_counts[row["user_id"]] = user_hourly_counts.get(row["user_id"], 0) + 1
        except Exception as exc:
            logger.exception("Queue flush: error processing queued post %s", qp_id)
            new_attempt = row["attempt_count"] + 1
            if new_attempt < 3:
                with get_db() as db:
                    retry_at = (datetime.now(timezone.utc) + timedelta(minutes=new_attempt * 10)).strftime("%Y-%m-%d %H:%M:%S")
                    db.execute(
                        "UPDATE queued_posts SET status = 'queued', error_message = ?,"
                        " attempt_count = ?, scheduled_at = ?, claim_token = NULL, claimed_at = NULL"
                        " WHERE id = ? AND claim_token = ?",
                        (f"Dispatch exception: {exc}", new_attempt, retry_at, qp_id, claim_token),
                    )
            else:
                with get_db() as db:
                    db.execute(
                        "UPDATE queued_posts SET status = 'failed', error_message = ?,"
                        " attempt_count = ? WHERE id = ? AND claim_token = ?",
                        (f"Dispatch exception: {exc}", new_attempt, qp_id, claim_token),
                    )


def flush_drips() -> None:
    """Release queued drip items whose rate window has room.

    Called by the drip scheduler job every 10 minutes. Each held item is
    re-rendered with the echo's current template and dispatched through
    the normal per-destination senders, so images, alt text, and content
    warnings behave exactly like instant posts. Failed releases return to
    the queue and retry from the stored payload. Disabled or deleted
    echoes have their backlog discarded as gave_up. Removing the limit
    entirely drains the queue in bounded batches per flush.
    """
    if not _acquire_job_lease("flush_drips", FLUSH_DRIP_LEASE_SECONDS):
        return
    try:
        _flush_drips()
    finally:
        _release_job_lease("flush_drips")


def _flush_drips() -> None:
    with get_db() as db:
        pending = db.execute(
            """
            SELECT d.id AS drip_id, d.item_id, d.item_json, d.attempts, e.*,
                   f.name AS feed_name, f.deleted_at AS feed_deleted_at
              FROM drip_items d
              JOIN echoes e ON d.echo_id = e.id
              JOIN feeds f ON e.feed_id = f.id
             ORDER BY d.id ASC
            """
        ).fetchall()

    discarded: set[int] = set()
    released: dict[int, int] = {}

    for row in pending:
        echo_id = row["id"]
        if echo_id in discarded:
            continue

        try:
            _flush_one_drip(row, discarded, released)
        except Exception:
            # Isolate one row's failure from the rest of the batch: an
            # uncaught exception here (a transient DB error, an unexpected
            # data shape) must not skip every other echo's pending drip
            # item for the whole tick. Mirrors _flush_queue's per-row
            # isolation.
            logger.exception(
                "Drip flush: unhandled error for echo %s item %s",
                echo_id,
                row["item_id"],
            )
            continue


def _flush_one_drip(row, discarded: set[int], released: dict[int, int]) -> None:
    echo_id = row["id"]

    if row["deleted_at"] or row["feed_deleted_at"]:
        _discard_drip_backlog(echo_id, "Echo deleted; drip backlog discarded")
        discarded.add(echo_id)
        return
    if not row["enabled"]:
        _discard_drip_backlog(echo_id, "Echo disabled; drip backlog discarded")
        discarded.add(echo_id)
        return

    # Serialize with process_echo so concurrent workers cannot both
    # see an open window and exceed the hourly cap.
    with _drip_lock(echo_id):
        limit = _drip_limit(row)
        if limit > 0 and _drip_rate(echo_id) >= limit:
            return
        if limit <= 0 and released.get(echo_id, 0) >= DRIP_DOWNGRADE_BATCH:
            # Limit removed: drain the stale backlog in batches
            # instead of one burst.
            return

        try:
            item = json.loads(row["item_json"])
        except (json.JSONDecodeError, TypeError):
            logger.warning(
                "Drip queue: unreadable payload %s for echo %s; dropping",
                row["drip_id"],
                echo_id,
            )
            with get_db() as db:
                db.execute("DELETE FROM drip_items WHERE id = ?", (row["drip_id"],))
            return

        claimed = _reclaim_queued(echo_id, row["item_id"])
        if claimed is None:
            # The posted row is no longer queued; remove the orphaned
            # queue entry so it is not reconsidered forever.
            with get_db() as db:
                db.execute("DELETE FROM drip_items WHERE id = ?", (row["drip_id"],))
            return

        posted_id, claim_token = claimed
        logger.info(
            "Echo %s: releasing dripped item %s (%d in window)",
            echo_id,
            row["item_id"],
            _drip_rate(echo_id),
        )
        try:
            _render_and_dispatch(dict(row), item, row["feed_name"] or "", posted_id, claim_token)
        except Exception as exc:
            # _render_and_dispatch converts its own and the senders'
            # failures into posted_items status changes; an exception
            # escaping it is an infrastructure fault (DB error, a bug).
            # The drip_items row has not been deleted yet at this point,
            # so the item must be requeued explicitly — leaving it
            # 'pending' after the claim would strand it until the ten
            # minute reclaim sweep, which does not restore the queue row
            # and would burn attempt_count. Retrying is the safe
            # direction for an item queued for delivery: the send may
            # not have been attempted, and the drip cap bounds how much
            # a persistently faulting release can dump per window.
            _requeue_drip_release_error(
                echo_id,
                row["item_id"],
                row["item_json"],
                row["attempts"],
                posted_id,
                claim_token,
                exc,
            )
            return

        with get_db() as db:
            db.execute("DELETE FROM drip_items WHERE id = ?", (row["drip_id"],))

        if _row_state(echo_id, row["item_id"]) == "failed":
            _requeue_drip_failure(
                echo_id,
                row["item_id"],
                row["item_json"],
                row["attempts"],
                posted_id,
            )
        else:
            released[echo_id] = released.get(echo_id, 0) + 1


def _discard_orphaned_digest_items() -> None:
    """Finalize digest items whose echo or destination no longer qualifies.

    flush_digests only picks up enabled, non-deleted echoes with a live email
    account (INNER JOIN). Anything else was never sent and never cleaned up:
    deleting an echo (which sets enabled = 0) or its email account stranded
    its queued items forever, leaving posted_items rows stuck at 'queued' and
    digest_items growing without bound. Mirrors _discard_drip_backlog, which
    already did this for drip mode.

    Deliberately NOT triggered by enabled = 0 on its own. That is the pause
    toggle, a reversible user action, and discarding a pause's backlog is
    irrecoverable data loss from a normal click. A paused echo queues nothing
    new (process_echo skips it), so its backlog is frozen, not growing, and it
    goes out on re-enable. Drip discards on pause for a different reason that
    does not apply here: releasing a drip backlog means a burst of separate
    posts, and its queue cap silently drops new items while full. A digest is
    one batched email with no cap, so keeping it is what a user expects.
    """
    with get_db() as db:
        orphans = db.execute(
            """
            SELECT DISTINCT d.echo_id
              FROM digest_items d
              LEFT JOIN echoes e ON e.id = d.echo_id
              LEFT JOIN feeds f ON f.id = e.feed_id
              LEFT JOIN email_accounts ea
                     ON ea.id = e.destination_id AND e.destination_type = 'email'
             WHERE e.id IS NULL
                OR e.deleted_at IS NOT NULL
                OR f.id IS NULL
                OR f.deleted_at IS NOT NULL
                OR ea.id IS NULL
            """
        ).fetchall()

    for row in orphans:
        echo_id = row["echo_id"]
        try:
            reason = "Digest discarded: echo or destination is no longer active"
            with get_db() as db:
                db.execute(
                    """UPDATE posted_items
                          SET status = 'gave_up',
                              error_message = ?,
                              claimed_at = NULL,
                              claim_token = NULL,
                              posted_at = ?
                        WHERE echo_id = ? AND status = 'queued'""",
                    (reason, _now(), echo_id),
                )
                db.execute("DELETE FROM digest_items WHERE echo_id = ?", (echo_id,))
            logger.info("Echo %s: %s", echo_id, reason)
        except Exception:
            # Isolate one orphan row's failure from the rest: this sweep
            # runs before the per-echo loop in _flush_digests, so an
            # unhandled exception here would abort the whole digest tick
            # before any live tenant's digest was even queried — the same
            # bug class the per-row isolation elsewhere guards against.
            logger.exception(
                "Digest orphan sweep: unhandled error for echo %s", echo_id
            )
            continue


def flush_digests() -> None:
    """Send pending digest items as batched emails and finalize them.

    Called by the digest scheduler job at the configured interval.
    Groups digest_items by echo, sends one email per echo, then
    deletes the items from digest_items and updates posted_items
    from 'queued' to 'success'.
    """
    if not _acquire_job_lease("flush_digests", FLUSH_DIGEST_LEASE_SECONDS):
        return
    try:
        _flush_digests()
    finally:
        _release_job_lease("flush_digests")


def _flush_digests() -> None:
    # Sweep first: stranded items are invisible to the query below, so nothing
    # would ever clear them. The sweep itself is per-row isolated; wrapping
    # the invocation too keeps an unexpected fault there from aborting the
    # whole tick before any live tenant's digest is queried.
    try:
        _discard_orphaned_digest_items()
    except Exception:
        logger.exception("Digest orphan sweep: unhandled error before flush")

    with get_db() as db:
        # Find all echoes that have pending digest items
        pending_echoes = db.execute("""
            SELECT DISTINCT d.echo_id, e.destination_id, e.feed_id, e.user_id,
                   f.name as feed_name, ea.email as to_email
              FROM digest_items d
              JOIN echoes e ON d.echo_id = e.id
              JOIN feeds f ON e.feed_id = f.id
              JOIN email_accounts ea ON e.destination_id = ea.id
              JOIN users u ON u.id = e.user_id
             WHERE e.destination_type = 'email'
               AND e.delivery_mode = 'digest'
               AND e.enabled = 1
               AND f.deleted_at IS NULL
               AND e.deleted_at IS NULL
               AND u.suspended = 0
               AND NOT (u.plan = 'trial' AND u.trial_ends_at IS NOT NULL
                        AND u.trial_ends_at <= ?)
        """, (_now(),)).fetchall()

    if not pending_echoes:
        return

    for echo_row in pending_echoes:
        echo_id = echo_row["echo_id"]
        try:
            _flush_one_digest(echo_id, echo_row)
        except Exception:
            # Isolate one echo's failure from the rest of the batch: an
            # uncaught exception here (a transient DB error, an unexpected
            # data shape) must not skip every other tenant's pending digest
            # for the whole tick. Mirrors _flush_queue's per-row isolation.
            logger.exception("Digest flush: unhandled error for echo %s", echo_id)
            # record_failure feeds the notify threshold, so a permanently
            # broken echo (corrupt payload, persistent DB error) surfaces
            # to the user instead of failing silently every tick. The
            # send_email failure path in _flush_one_digest calls it too;
            # only exceptions OUTSIDE that guard were silent before.
            record_failure(echo_id)
            continue


def _build_digest_body(echo_row, items) -> tuple[str, str, list, list]:
    """Assemble a digest email's subject/body from queued rows.

    Returns (subject, body, sent_items, held_items). Items that fit the
    char cap go in the body; the rest are returned as held for the next
    flush. Malformed rows raise, which the caller's per-echo isolation
    contains. Extracted so tests can fault-inject body assembly (the step
    the narrow send_email try/except never covered).
    """
    # Build digest body incrementally so overflow can be detected
    # per-item: everything that fits goes out now, the rest stays
    # queued for the next flush. Silently truncating here reported
    # dropped content as delivered.
    date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    subject = f"FeedEcho Digest — {echo_row['feed_name']} — {date_str}"

    header = [f"FeedEcho Digest for {echo_row['feed_name']}", f"Date: {date_str}", ""]

    body_parts = list(header)
    used = sum(len(p) + 1 for p in body_parts)
    sent_items, held_items = [], []
    for i, item in enumerate(items, 1):
        title = item["item_title"] or item["item_url"] or f"Item {i}"
        lines = (
            f"{i}. {title}",
            f"   {item['rendered_content']}",
            "",
        )
        added = sum(len(ln) + 1 for ln in lines)
        if used + added > DIGEST_MAX_CHARS - DIGEST_NOTICE_RESERVE:
            held_items.append(item)
            continue
        body_parts.extend(lines)
        used += added
        sent_items.append(item)

    if held_items and not sent_items:
        # A single item larger than the whole cap: normally send it
        # truncated so one giant item cannot wedge the queue forever.
        # Degenerate case (a pathological title/URL that leaves no room
        # for any content): keep everything held instead of sending a
        # content-less title.
        first = held_items.pop(0)
        body_parts.append(f"1. {first['item_title'] or first['item_url'] or 'Item 1'}")
        budget = DIGEST_MAX_CHARS - DIGEST_NOTICE_RESERVE - used - len(body_parts[-1]) - 8
        if budget <= 0:
            held_items.insert(0, first)
        else:
            body_parts.append(f"   {first['rendered_content'][:budget]}…")
            sent_items.append(first)

    body = "\n".join(body_parts)
    if held_items:
        body += (
            f"\n\n[{len(held_items)} newer item{'s' if len(held_items) != 1 else ''}"
            " held for the next digest to stay under the size cap.]"
        )
    return subject, body, sent_items, held_items


def _flush_one_digest(echo_id: int, echo_row) -> None:
    with get_db() as db:
        items = db.execute(
            "SELECT * FROM digest_items WHERE echo_id = ? ORDER BY created_at ASC",
            (echo_id,),
        ).fetchall()

    if not items:
        return

    subject, body, sent_items, held_items = _build_digest_body(echo_row, items)

    if not sent_items:
        # Nothing fit this round (pathological content only): leave the
        # queue untouched rather than send an empty digest or drop the
        # item. Logged loudly so a wedged queue is discoverable.
        logger.warning(
            "Digest for echo %s held: an item exceeds the %d char cap",
            echo_id,
            DIGEST_MAX_CHARS,
        )
        return

    try:
        send_email(
            to_email=echo_row["to_email"],
            subject=subject,
            body=body,
            user_id=echo_row["user_id"],
        )
    except Exception as exc:
        logger.exception(
            "Digest flush failed for echo %s (%s, %d items)",
            echo_id,
            echo_row["feed_name"],
            len(items),
        )
        # Surface digest send failures to the user: mark the ATTEMPTED
        # rows (the ones in the body) 'failed' — without a retry time,
        # so the retry sweep never grabs them; only flush_digests owns
        # digest delivery — so the failure counter in
        # notify.record_failure can reach its threshold and alert on a
        # permanently broken SMTP config. Held items were never
        # attempted this round and stay 'queued'.
        with get_db() as db:
            for item in sent_items:
                db.execute(
                    """UPDATE posted_items
                          SET status = 'failed',
                              error_message = ?,
                              next_retry_at = NULL,
                              attempt_count = attempt_count + 1
                        WHERE echo_id = ? AND item_id = ?
                          AND status IN ('queued', 'failed')""",
                    (f"Digest send failed: {exc}", echo_id, item["item_id"]),
                )
        record_failure(echo_id)
        # Leave digest_items intact — the next successful flush sends them.
        return

    # Success — delete only the sent items and update posted_items to
    # 'success'; held items stay queued for the next flush.
    sent_item_ids = [item["item_id"] for item in sent_items]
    with get_db() as db:
        for item in sent_items:
            # Update posted_items to 'success'. 'failed' is included:
            # a prior flush may have marked the row failed on a send
            # error; a successful send now must still finalize it,
            # else the queue row below is deleted with the posted_item
            # stuck on 'failed'.
            db.execute(
                """UPDATE posted_items SET status = 'success', posted_at = ?
                   WHERE echo_id = ? AND item_id = ?
                     AND status IN ('queued', 'failed')""",
                (_now(), echo_id, item["item_id"]),
            )
        # Delete only the sent items, not any that may have arrived concurrently
        placeholders = ",".join("?" for _ in sent_item_ids)
        db.execute(
            f"DELETE FROM digest_items WHERE echo_id = ? AND item_id IN ({placeholders})",
            (echo_id, *sent_item_ids),
        )

    record_success(echo_id)
    logger.info(
        "Digest flushed for echo %s (%s): %d items sent, %d held",
        echo_id,
        echo_row["feed_name"],
        len(sent_items),
        len(held_items),
    )


def check_all_feeds() -> None:
    now = datetime.now(timezone.utc)
    with get_db() as db:
        if settings.MULTI:
            # Trial-pause + suspension gate at the sweep level too: a paused
            # user's feeds are not even considered "due", so their cursors
            # freeze (no fetch, no delivery, no retries) and nothing is
            # deleted. Single mode has no per-user state to check.
            feeds = db.execute(
                """
                SELECT f.id, f.name, f.poll_interval, f.last_fetched
                  FROM feeds f JOIN users u ON u.id = f.user_id
                 WHERE f.deleted_at IS NULL
                   AND u.suspended = 0
                   AND NOT (u.plan = 'trial' AND u.trial_ends_at IS NOT NULL
                            AND u.trial_ends_at <= ?)
                """,
                (_now(),),
            ).fetchall()
        else:
            feeds = db.execute(
                "SELECT id, name, poll_interval, last_fetched FROM feeds WHERE deleted_at IS NULL"
            ).fetchall()

    due = []
    for feed in feeds:
        if not feed["last_fetched"]:
            due.append(feed)
            continue
        try:
            # timestamp_str first: psycopg returns a datetime for
            # last_fetched, and datetime.replace("T", " ") raises TypeError,
            # which this except would swallow into "always due" — making
            # every feed poll on every tick and ignoring poll_interval.
            ts = timestamp_str(feed["last_fetched"])
            last = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S").replace(
                tzinfo=timezone.utc
            )
        except (ValueError, TypeError):
            # Malformed timestamp: treat as due rather than skipping forever
            due.append(feed)
            continue
        # poll_interval is clamped to >=1 at creation; a malformed or NULL
        # value can only come from direct DB writes, so degrade to the
        # default instead of crashing the whole run or hot-looping the feed.
        try:
            interval = max(1, int(feed["poll_interval"] or 15))
        except (ValueError, TypeError):
            interval = 15
        if last <= now - timedelta(minutes=interval):
            due.append(feed)

    for feed in due:
        try:
            check_feed(feed["id"])
        except Exception:
            logger.exception("Error checking feed %s (%s)", feed["id"], feed["name"])


def cleanup_pending_accounts() -> int:
    """Delete abandoned card-pending signups (hosted billing only).

    A card-pending account registered while billing is on but never finished
    Stripe Checkout: plan='trial' with the TRIAL_PENDING sentinel. It cannot
    reach the app at all, so once it is older than
    settings.PENDING_ACCOUNT_TTL_DAYS it is only a row, a scrypt hash, a
    verification email and admin-table noise. Deletion runs the same path as
    self-serve deletion (account-deletion hooks, then _hard_delete_user), so
    a Stripe customer created by the abandoned checkout does not outlive the
    row and no tenant data survives.

    Rows carrying a subscription id are SKIPPED and logged: a webhook that
    failed mid-flight must never get its paying user deleted. Overlay-
    registered guards (_pending_cleanup_guards) get the same veto first: the
    hosted billing module asks Stripe whether a supposedly-cardless account
    actually has a live subscription. A guard failure is fail-closed.

    Returns the number of accounts deleted (0 when not hosted billing).
    """
    if not (settings.MULTI and settings.BILLING_ENABLED):
        return 0
    # Imported lazily: app imports this module at startup.
    import app as app_module

    cutoff = (
        datetime.now(timezone.utc)
        - timedelta(days=settings.PENDING_ACCOUNT_TTL_DAYS)
    ).strftime("%Y-%m-%d %H:%M:%S")
    with get_db() as db:
        rows = db.execute(
            "SELECT id, email, trial_ends_at, stripe_subscription_id"
            " FROM users WHERE plan = 'trial' AND created_at < ?",
            (cutoff,),
        ).fetchall()
    # Sentinel check in Python: the stored form differs by dialect.
    pending = [row for row in rows if plans.trial_pending(row["trial_ends_at"])]

    deleted = 0
    for row in pending:
        uid = row["id"]
        if (row["stripe_subscription_id"] or "").strip():
            logger.warning(
                "Pending-signup cleanup: user %s has a subscription id; skipping",
                uid,
            )
            continue
        vetoed = False
        for guard in app_module._pending_cleanup_guards:
            try:
                guard(uid)
            except app_module.AccountDeletionAbort as exc:
                logger.warning(
                    "Pending-signup cleanup: guard vetoed user %s (%s)", uid, exc
                )
                vetoed = True
                break
            except Exception:  # noqa: BLE001 — fail closed, never guess
                logger.exception(
                    "Pending-signup cleanup: guard failed for user %s; skipping", uid
                )
                vetoed = True
                break
        if vetoed:
            continue
        for hook in app_module._account_deletion_hooks:
            try:
                hook(uid)
            except app_module.AccountDeletionAbort as exc:
                logger.warning(
                    "Pending-signup cleanup: hook vetoed user %s (%s)", uid, exc
                )
                vetoed = True
                break
            except Exception:  # noqa: BLE001 — mirrors the delete route
                logger.exception(
                    "Pending-signup cleanup: deletion hook failed for user %s", uid
                )
        if vetoed:
            continue
        # One account failing must not abort the batch: an unhandled error
        # here would re-hit the same row every run and stop cleaning later
        # accounts.
        try:
            with get_db() as db:
                app_module._hard_delete_user(db, uid)
        except Exception:  # noqa: BLE001
            logger.exception(
                "Pending-signup cleanup: deleting user %s failed; continuing", uid
            )
            continue
        deleted += 1
        logger.info(
            "Pending-signup cleanup: deleted user %s (%s)", uid, row["email"]
        )
    return deleted


def start_scheduler() -> None:
    global scheduler
    if scheduler is not None:
        return

    scheduler = BackgroundScheduler()
    scheduler.add_job(
        check_all_feeds,
        trigger=IntervalTrigger(minutes=2),
        id="check_all_feeds",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.start()
    scheduler.add_job(check_all_feeds, "date", id="startup_check", replace_existing=True)

    # Digest flush job — runs hourly, checks for pending digest items
    scheduler.add_job(
        flush_digests,
        trigger=IntervalTrigger(hours=1),
        id="flush_digests",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Run a digest flush shortly after startup
    scheduler.add_job(flush_digests, "date", id="startup_digest_flush", replace_existing=True)

    # Drip flush job — releases queued items as their rate windows allow
    scheduler.add_job(
        flush_drips,
        trigger=IntervalTrigger(minutes=10),
        id="flush_drips",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    # Run a drip flush shortly after startup
    scheduler.add_job(flush_drips, "date", id="startup_drip_flush", replace_existing=True)

    # Queue flush job — dispatches human-scheduled curation queue posts
    scheduler.add_job(
        flush_queue,
        trigger=IntervalTrigger(minutes=5),
        id="flush_queue",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(flush_queue, "date", id="startup_queue_flush", replace_existing=True)

    # Abandoned-signup cleanup — deletes card-pending accounts past the TTL
    # (no-op unless multi mode with billing enabled).
    scheduler.add_job(
        cleanup_pending_accounts,
        trigger=IntervalTrigger(hours=6),
        id="cleanup_pending_accounts",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(
        cleanup_pending_accounts,
        "date",
        id="startup_pending_cleanup",
        replace_existing=True,
    )


def stop_scheduler() -> None:
    global scheduler
    if scheduler is not None:
        scheduler.shutdown(wait=False)
        scheduler = None