scheduler.py
Rama master · réplica de lectura del repositorio original.
"""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