database.py
Rama master · réplica de lectura del repositorio original.
"""Database layer — SQLite (single mode) or PostgreSQL (multi mode).
Every query in the codebase is written with ``?`` placeholders and runs
unchanged on both backends: :func:`qmark` translates ``?`` to ``%s`` when
the dialect is postgres, and ``get_db`` yields a connection whose
``execute`` performs that translation. Rows support ``row["col"]`` on both
backends (sqlite3.Row / psycopg dict_row).
"""
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
import settings
from settings import DB_PATH
from security import decrypt_secret, hash_secret
def dialect() -> str:
"""'postgres' when running multi-tenant against DATABASE_URL, else 'sqlite'."""
return "postgres" if (settings.MULTI and settings.DATABASE_URL) else "sqlite"
def qmark(sql: str) -> str:
"""Translate ``?`` placeholders to ``%s`` for Postgres.
Single-quoted string literals are skipped: ``?`` inside them is left
alone, ``%`` is escaped to ``%%`` (psycopg3 treats bare ``%`` in
literals as a placeholder), and SQL-standard doubled quotes
(``'it''s'``) are handled so a ``?`` after an escaped quote stays
inside the literal. No-op when the dialect is sqlite.
Deliberately out of scope: ``E'...'`` backslash escapes and
dollar-quoted strings — FeedEcho's SQL never uses them.
"""
if dialect() != "postgres":
return sql
out = []
i = 0
n = len(sql)
while i < n:
ch = sql[i]
if ch != "'":
out.append("%s" if ch == "?" else ch)
i += 1
continue
# Enter a string literal: copy verbatim until the closing quote.
out.append(ch)
i += 1
while i < n:
if sql[i] == "'":
if i + 1 < n and sql[i + 1] == "'": # doubled quote escape
out.append("''")
i += 2
continue
out.append("'")
i += 1
break
out.append("%%" if sql[i] == "%" else sql[i])
i += 1
return "".join(out)
def as_utc_naive(value) -> datetime | None:
"""Normalize a stored TIMESTAMP value to a naive UTC datetime.
The read-side counterpart to :func:`qmark`. sqlite hands TIMESTAMP
columns back as the TEXT that was written; psycopg hands them back as
``datetime`` objects. Anything that parses or compares a stored
timestamp in Python must normalize first, or it works on sqlite and
raises ``TypeError`` on Postgres only.
Returns None when the value is empty or unparseable.
Invariant: every writer stores UTC in these TIMESTAMP columns (app and
scheduler code generate UTC strings via ``scheduler._now()``; sqlite
``CURRENT_TIMESTAMP`` is UTC). Naive values are therefore treated as
UTC. Do not add writers that store local time, and prefer binding an
explicit ``_now()`` over Postgres ``CURRENT_TIMESTAMP``, which resolves
in the session time zone rather than UTC.
"""
if not value:
return None
if isinstance(value, datetime):
dt = value
else:
try:
dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError:
return None
if dt.tzinfo is not None:
dt = dt.astimezone(timezone.utc).replace(tzinfo=None)
return dt
def timestamp_str(value) -> str:
"""Normalize a stored TIMESTAMP value to 'YYYY-MM-DD HH:MM:SS' (UTC).
For code that compares a stored timestamp against the app's own UTC
timestamp strings. Returns "" when the value is missing or
unparseable, so callers can treat "no timestamp" as falsy.
"""
dt = as_utc_naive(value)
return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else ""
class _PgConnection:
"""Thin adapter over psycopg3 so callers keep writing db.execute(sql, ?)."""
def __init__(self, conn):
self._conn = conn
def execute(self, sql, params=()):
return self._conn.execute(qmark(sql), params)
def commit(self):
self._conn.commit()
def rollback(self):
self._conn.rollback()
def close(self):
self._conn.close()
def _pg_connect():
try:
import psycopg
except ImportError as exc:
raise RuntimeError(
"PostgreSQL mode requires the psycopg package: "
"pip install 'feedecho[postgres]'"
) from exc
conn = psycopg.connect(
settings.DATABASE_URL,
row_factory=psycopg.rows.dict_row,
# Pin the session timezone to UTC at the handshake itself (not via a
# post-connect SET, which runs inside psycopg3's implicit transaction
# and reverts to the server default on any ROLLBACK, leaving later
# statements on the same connection skewed).
options="-c timezone=UTC",
)
# Postgres's CURRENT_TIMESTAMP/NOW() resolve in the session/server
# timezone before being cast into our naive TIMESTAMP columns (unlike
# sqlite's CURRENT_TIMESTAMP, which is always UTC) — without this, a
# non-UTC server timezone silently skews every stored timestamp
# relative to as_utc_naive()/timestamp_str()'s UTC assumption.
return conn
@contextmanager
def get_db():
"""Yield a connection, committing on clean exit, rolling back on error.
Postgres connections commit/rollback via psycopg's transaction
context manager; SQLite connections use WAL mode (set once during
init_db) with busy_timeout=30s for write contention.
"""
if dialect() == "postgres":
conn = _pg_connect()
try:
with conn: # commits on success, rolls back on exception
yield _PgConnection(conn)
finally:
conn.close()
return
conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys=ON")
conn.execute("PRAGMA busy_timeout=30000")
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def _column_names(db, table_name: str) -> set[str]:
if dialect() == "postgres":
rows = db.execute(
"SELECT column_name AS name FROM information_schema.columns"
" WHERE table_name = ?",
(table_name,),
).fetchall()
return {row["name"] for row in rows}
return {row["name"] for row in db.execute(f"PRAGMA table_info({table_name})")}
def _has_unique_on(db, table_name: str, columns: set[str]) -> bool:
"""True if the table has a UNIQUE index covering exactly `columns`.
SQLite-only helper: detects inline UNIQUE constraints (autoindexes)
so migrations can be idempotent across restarts.
"""
for idx in db.execute(f"PRAGMA index_list({table_name})"):
if not idx["unique"]:
continue
cols = {r["name"] for r in db.execute(f"PRAGMA index_info({idx['name']})")}
if cols == columns:
return True
return False
def _add_column_if_missing(
db,
table_name: str,
column_name: str,
column_definition: str,
) -> None:
if dialect() == "postgres":
db.execute(
f"ALTER TABLE {table_name} ADD COLUMN IF NOT EXISTS "
f"{column_name} {column_definition}"
)
return
if column_name not in _column_names(db, table_name):
db.execute(
f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_definition}"
)
def _dedupe_discord_webhook_hashes(db) -> None:
"""Collapse duplicate (user_id, webhook_url_hash) rows before the unique
index is built.
The old UNIQUE(user_id, webhook_url) was on the ciphertext, so a prior
racy dedup could theoretically have stored the same plaintext webhook
twice under two ciphertexts — the backfill would then give them the same
hash and CREATE UNIQUE INDEX would fail. Keep the lowest id per group.
"""
dupes = db.execute(
"SELECT user_id, webhook_url_hash, MIN(id) AS keep_id"
" FROM discord_accounts WHERE webhook_url_hash != ''"
" GROUP BY user_id, webhook_url_hash HAVING COUNT(*) > 1"
).fetchall()
for d in dupes:
db.execute(
"DELETE FROM discord_accounts"
" WHERE user_id = ? AND webhook_url_hash = ? AND id != ?",
(d["user_id"], d["webhook_url_hash"], d["keep_id"]),
)
def _dedupe_accounts_for_unique_index(db) -> None:
"""Collapse duplicate (user_id, instance, username) account rows before
the unique index is built.
Pre-upgrade ``oauth_callback`` and ``add_account`` were plain INSERTs on
every reconnect, so long-lived installs can already hold multiple rows
for the same logical account; CREATE UNIQUE INDEX over them would crash
init_db at boot (the v1.39.0 Discord-hash failure class). Keep the
highest id per group — the most recent reconnect carries the freshest
token and display name — repoint the references that make rows distinct
accounts (echoes + queued_posts by destination), then delete the older
rows. Runs best-effort: the index creation below is the hard gate.
"""
dupes = db.execute(
"SELECT user_id, instance, username, MAX(id) AS keep_id"
" FROM accounts"
" GROUP BY user_id, instance, username HAVING COUNT(*) > 1"
).fetchall()
for d in dupes:
keep_id = d["keep_id"]
stale = db.execute(
"SELECT id FROM accounts"
" WHERE user_id = ? AND instance = ? AND username = ? AND id != ?",
(d["user_id"], d["instance"], d["username"], keep_id),
).fetchall()
for s in stale:
old_id = s["id"]
# Echoes that would collide after the repoint (an echo already
# points at the surviving account from the same feed): merge post
# history into the surviving echo, then drop the duplicate — a
# blind repoint would double-post every item to the destination.
collisions = db.execute(
"SELECT o.id AS old_echo_id, k.id AS keep_echo_id"
" FROM echoes o JOIN echoes k"
" ON k.user_id = o.user_id AND k.feed_id = o.feed_id"
" AND k.destination_type = o.destination_type"
" AND k.destination_id = ?"
" AND k.deleted_at IS NULL"
" WHERE o.destination_type = 'mastodon' AND o.destination_id = ?"
" AND o.deleted_at IS NULL",
(keep_id, old_id),
).fetchall()
for c in collisions:
db.execute(
"UPDATE posted_items SET echo_id = ? WHERE echo_id = ?",
(c["keep_echo_id"], c["old_echo_id"]),
)
db.execute("DELETE FROM echoes WHERE id = ?", (c["old_echo_id"],))
# Duplicate queued jobs for the same item, pointing at the old
# account while a queued job for the survivor already exists.
db.execute(
"DELETE FROM queued_posts"
" WHERE destination_type = 'mastodon' AND destination_id = ?"
" AND status = 'queued'"
" AND EXISTS (SELECT 1 FROM queued_posts q2 WHERE q2.feed_id = queued_posts.feed_id"
" AND q2.item_id = queued_posts.item_id"
" AND q2.destination_type = 'mastodon'"
" AND q2.destination_id = ? AND q2.status = 'queued')",
(old_id, keep_id),
)
db.execute(
"UPDATE echoes SET destination_id = ?"
" WHERE destination_type = 'mastodon' AND destination_id = ?",
(keep_id, old_id),
)
db.execute(
"UPDATE queued_posts SET destination_id = ?"
" WHERE destination_type = 'mastodon' AND destination_id = ?",
(keep_id, old_id),
)
db.execute("DELETE FROM accounts WHERE id = ?", (old_id,))
def _dedupe_feeds_for_unique_index(db) -> None:
"""Collapse duplicate active (user_id, url) feed rows before the unique
index is built.
Same pre-upgrade plain-INSERT exposure as accounts. Keep the lowest id
per group — the original feed, with its poll state and item history —
repoint echoes/queued_posts, merge feed_items (delete items that would
collide on the destination feed's (feed_id, item_id) unique index, then
repoint the rest), and soft-delete the duplicate rows so the partial
index accepts them while posted_items history stays intact.
"""
dupes = db.execute(
"SELECT user_id, url, MIN(id) AS keep_id"
" FROM feeds WHERE deleted_at IS NULL"
" GROUP BY user_id, url HAVING COUNT(*) > 1"
).fetchall()
for d in dupes:
keep_id = d["keep_id"]
stale = db.execute(
"SELECT id FROM feeds"
" WHERE user_id = ? AND url = ? AND deleted_at IS NULL AND id != ?",
(d["user_id"], d["url"], keep_id),
).fetchall()
for s in stale:
old_id = s["id"]
# Echoes that would collide after the repoint (the surviving feed
# already has an echo to the same destination): preserve their
# post history by moving posted_items to the surviving echo, then
# drop the duplicate echo. A blind repoint here would leave two
# identical echoes on one feed and every item would post twice.
collisions = db.execute(
"SELECT o.id AS old_echo_id, k.id AS keep_echo_id"
" FROM echoes o JOIN echoes k"
" ON k.user_id = o.user_id AND k.feed_id = ?"
" AND k.destination_type = o.destination_type"
" AND k.destination_id = o.destination_id"
" AND k.deleted_at IS NULL"
" WHERE o.feed_id = ? AND o.deleted_at IS NULL",
(keep_id, old_id),
).fetchall()
for c in collisions:
db.execute(
"UPDATE posted_items SET echo_id = ? WHERE echo_id = ?",
(c["keep_echo_id"], c["old_echo_id"]),
)
db.execute("DELETE FROM echoes WHERE id = ?", (c["old_echo_id"],))
# Duplicate queued jobs for the same item+destination (both feeds
# queued the item before the dedupe): keep the survivor's row.
db.execute(
"DELETE FROM queued_posts WHERE feed_id = ? AND status = 'queued'"
" AND EXISTS (SELECT 1 FROM queued_posts q2 WHERE q2.feed_id = ?"
" AND q2.item_id = queued_posts.item_id"
" AND q2.destination_type = queued_posts.destination_type"
" AND q2.destination_id = queued_posts.destination_id"
" AND q2.status = 'queued')",
(old_id, keep_id),
)
db.execute(
"UPDATE echoes SET feed_id = ? WHERE feed_id = ?",
(keep_id, old_id),
)
db.execute(
"UPDATE queued_posts SET feed_id = ? WHERE feed_id = ?",
(keep_id, old_id),
)
db.execute(
"DELETE FROM feed_items"
" WHERE feed_id = ? AND item_id IN"
" (SELECT item_id FROM feed_items WHERE feed_id = ?)",
(old_id, keep_id),
)
db.execute(
"UPDATE feed_items SET feed_id = ? WHERE feed_id = ?",
(keep_id, old_id),
)
db.execute(
"UPDATE feeds SET deleted_at = CURRENT_TIMESTAMP WHERE id = ?",
(old_id,),
)
def prune_feed_items(db, feed_id: int, limit: int | None = None) -> None:
"""Delete a feed's oldest items beyond ``limit``, newest kept.
Starred rows are exempt from general retention: they are never deleted
by the unstarred cap and do not count toward it. If
``settings.READER_MAX_STARRED_PER_FEED`` is non-zero, a separate cap trims
oldest starred items beyond that safety threshold.
Ordering is ``published_at IS NULL, published_at DESC, id DESC`` so NULL
dates are pruned first on BOTH dialects: bare ``published_at DESC`` sorts
NULLs last on sqlite but first on Postgres, which would keep the wrong
rows. ``published_at IS NULL`` (0 before 1) normalizes that.
"""
cap = limit if limit is not None else settings.READER_MAX_ITEMS_PER_FEED
if cap > 0:
db.execute(
"""
DELETE FROM feed_items
WHERE feed_id = ?
AND starred = 0
AND id NOT IN (
SELECT id FROM feed_items
WHERE feed_id = ? AND starred = 0
ORDER BY published_at IS NULL, published_at DESC, id DESC
LIMIT ?
)
""",
(feed_id, feed_id, cap),
)
starred_cap = getattr(settings, "READER_MAX_STARRED_PER_FEED", 0)
if starred_cap > 0:
db.execute(
"""
DELETE FROM feed_items
WHERE feed_id = ?
AND starred = 1
AND id NOT IN (
SELECT id FROM feed_items
WHERE feed_id = ? AND starred = 1
ORDER BY published_at IS NULL, published_at DESC, id DESC
LIMIT ?
)
""",
(feed_id, feed_id, starred_cap),
)
def init_db() -> None:
"""Create and migrate the schema for the active dialect."""
if dialect() == "postgres":
init_db_postgres()
else:
init_db_sqlite()
def init_db_sqlite() -> None:
"""Create and migrate the SQLite application schema."""
with get_db() as db:
db.execute("PRAGMA journal_mode=WAL")
db.execute("""
CREATE TABLE IF NOT EXISTS accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
username TEXT DEFAULT '',
instance TEXT NOT NULL,
access_token TEXT NOT NULL,
booster_enabled INTEGER NOT NULL DEFAULT 0,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
account_columns = _column_names(db, "accounts")
if "username" not in account_columns:
db.execute("ALTER TABLE accounts ADD COLUMN username TEXT DEFAULT ''")
rows = db.execute("SELECT id, name FROM accounts").fetchall()
import re
for row in rows:
match = re.search(r"\(([^)]+)\)$", row["name"] or "")
username = (
match.group(1)
if match
else (row["name"] or "unknown")
)
db.execute(
"UPDATE accounts SET username = ? WHERE id = ?",
(username, row["id"]),
)
_add_column_if_missing(db, "accounts", "booster_enabled", "INTEGER NOT NULL DEFAULT 0")
db.execute("""
CREATE TABLE IF NOT EXISTS saved_searches (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 1,
name TEXT NOT NULL,
query TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_saved_searches_user_name
ON saved_searches(user_id, LOWER(name))
""")
db.execute("""
CREATE TABLE IF NOT EXISTS folders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 1,
name TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_folders_user_name
ON folders(user_id, LOWER(name))
""")
db.execute("""
CREATE TABLE IF NOT EXISTS feeds (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
url TEXT NOT NULL,
feed_type TEXT DEFAULT 'rss',
poll_interval INTEGER DEFAULT 15,
last_fetched TIMESTAMP,
last_item_id TEXT,
lease_token TEXT,
lease_expires_at TIMESTAMP,
paused INTEGER NOT NULL DEFAULT 0,
read_enabled INTEGER NOT NULL DEFAULT 0,
ai_alt_override INTEGER NOT NULL DEFAULT 0,
mute_keywords TEXT DEFAULT '',
last_error TEXT,
etag TEXT,
last_modified TEXT,
poke_token TEXT,
last_poked_at TIMESTAMP,
deleted_at TIMESTAMP,
user_id INTEGER NOT NULL DEFAULT 1,
folder_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
_add_column_if_missing(db, "feeds", "lease_token", "TEXT")
_add_column_if_missing(db, "feeds", "lease_expires_at", "TIMESTAMP")
_add_column_if_missing(db, "feeds", "paused", "INTEGER NOT NULL DEFAULT 0")
_add_column_if_missing(db, "feeds", "read_enabled", "INTEGER NOT NULL DEFAULT 0")
# Per-feed toggle: when set, AI-generated alt text wins over the
# feed-provided alt text on image echoes (default off = feed alt wins).
_add_column_if_missing(db, "feeds", "ai_alt_override", "INTEGER NOT NULL DEFAULT 0")
_add_column_if_missing(db, "feeds", "mute_keywords", "TEXT DEFAULT ''")
_add_column_if_missing(db, "feeds", "last_error", "TEXT")
_add_column_if_missing(db, "feeds", "folder_id", "INTEGER")
_add_column_if_missing(db, "feeds", "etag", "TEXT")
_add_column_if_missing(db, "feeds", "last_modified", "TEXT")
_add_column_if_missing(db, "feeds", "poke_token", "TEXT")
_add_column_if_missing(db, "feeds", "last_poked_at", "TIMESTAMP")
# Soft-delete marker: feeds are never hard-deleted by the app so that
# echo configuration and posted-item history survive as an audit trail.
_add_column_if_missing(db, "feeds", "deleted_at", "TIMESTAMP")
db.execute("""
CREATE TABLE IF NOT EXISTS feed_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
title TEXT,
link TEXT,
summary TEXT,
content TEXT,
content_text TEXT,
content_link TEXT,
content_html TEXT,
categories TEXT,
author TEXT,
image_url TEXT,
image_alt TEXT,
image_urls TEXT,
enclosure_url TEXT,
published_at TIMESTAMP,
is_read INTEGER NOT NULL DEFAULT 0,
starred INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (feed_id) REFERENCES feeds(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_feed_items_feed_item
ON feed_items(feed_id, item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feed_items_feed_pub
ON feed_items(feed_id, published_at DESC)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feed_items_feed_read
ON feed_items(feed_id, is_read)
""")
_add_column_if_missing(db, "feed_items", "content_text", "TEXT")
# v1.66.0: sanitized HTML passthrough for {{ content_html }}.
_add_column_if_missing(db, "feed_items", "content_html", "TEXT")
# v1.73.2: RSS/Atom category domains for {{ categories }} (JSON dict).
_add_column_if_missing(db, "feed_items", "categories", "TEXT")
_add_column_if_missing(db, "feed_items", "image_url", "TEXT")
_add_column_if_missing(db, "feed_items", "image_alt", "TEXT")
_add_column_if_missing(db, "feed_items", "image_urls", "TEXT")
_add_column_if_missing(db, "feed_items", "enclosure_url", "TEXT")
_add_column_if_missing(db, "feed_items", "starred", "INTEGER NOT NULL DEFAULT 0")
echo_columns = _column_names(db, "echoes")
if (
echo_columns
and "account_id" in echo_columns
and "destination_type" not in echo_columns
):
db.execute("DROP TABLE echoes")
db.execute("""
CREATE TABLE IF NOT EXISTS echoes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
feed_id INTEGER NOT NULL,
destination_type TEXT NOT NULL DEFAULT 'mastodon',
destination_id INTEGER NOT NULL,
template TEXT NOT NULL DEFAULT '{{ title }} {{ link }}',
visibility TEXT DEFAULT 'public',
enabled INTEGER DEFAULT 1,
user_id INTEGER NOT NULL DEFAULT 1,
filter_keywords TEXT DEFAULT '',
filter_mode TEXT NOT NULL DEFAULT 'exclude',
content_warning TEXT DEFAULT '',
attach_image INTEGER NOT NULL DEFAULT 0,
render_html INTEGER NOT NULL DEFAULT 0,
use_markdown INTEGER NOT NULL DEFAULT 0,
delivery_mode TEXT NOT NULL DEFAULT 'instant',
drip_limit INTEGER NOT NULL DEFAULT 0,
one_shot INTEGER NOT NULL DEFAULT 0,
booster_enabled INTEGER NOT NULL DEFAULT 0,
deleted_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (feed_id) REFERENCES feeds(id) ON DELETE CASCADE
)
""")
_add_column_if_missing(db, "echoes", "filter_keywords", "TEXT DEFAULT ''")
_add_column_if_missing(
db, "echoes", "filter_mode", "TEXT NOT NULL DEFAULT 'exclude'"
)
_add_column_if_missing(db, "echoes", "content_warning", "TEXT DEFAULT ''")
_add_column_if_missing(
db, "echoes", "attach_image", "INTEGER NOT NULL DEFAULT 0"
)
_add_column_if_missing(
db, "echoes", "delivery_mode", "TEXT NOT NULL DEFAULT 'instant'"
)
_add_column_if_missing(
db, "echoes", "drip_limit", "INTEGER NOT NULL DEFAULT 0"
)
_add_column_if_missing(db, "echoes", "deleted_at", "TIMESTAMP")
_add_column_if_missing(db, "echoes", "one_shot", "INTEGER NOT NULL DEFAULT 0")
_add_column_if_missing(db, "echoes", "booster_enabled", "INTEGER NOT NULL DEFAULT 0")
# v1.67.0: email echoes may render the body as HTML ({{ content_html }}).
_add_column_if_missing(
db, "echoes", "render_html", "INTEGER NOT NULL DEFAULT 0"
)
# Mastodon/GoToSocial: send content_type=text/markdown when set.
_add_column_if_missing(
db, "echoes", "use_markdown", "INTEGER NOT NULL DEFAULT 0"
)
# GitHub destination: per-echo file path and commit message templates
# ('' = not a github echo / default commit message).
_add_column_if_missing(
db, "echoes", "github_path", "TEXT NOT NULL DEFAULT ''"
)
_add_column_if_missing(
db, "echoes", "github_commit", "TEXT NOT NULL DEFAULT ''"
)
db.execute("""
CREATE TABLE IF NOT EXISTS digest_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_title TEXT,
item_url TEXT,
rendered_content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(echo_id, item_id),
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_digest_items_echo
ON digest_items(echo_id)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS drip_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_json TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
queued_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(echo_id, item_id),
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_drip_items_echo
ON drip_items(echo_id)
""")
# Migrate drip_items tables created by the initial 1.11.0 schema
# draft, which predated the attempts column.
_add_column_if_missing(
db, "drip_items", "attempts", "INTEGER NOT NULL DEFAULT 0"
)
db.execute("""
CREATE TABLE IF NOT EXISTS email_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, email)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS bluesky_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
handle TEXT NOT NULL,
app_password TEXT NOT NULL,
did TEXT DEFAULT '',
pds TEXT DEFAULT '',
access_jwt TEXT DEFAULT '',
refresh_jwt TEXT DEFAULT '',
session_expires_at TIMESTAMP,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, handle)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS microblog_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
uid TEXT NOT NULL,
token TEXT NOT NULL,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, uid)
)
""")
# One row per room: the same token usually posts to several rooms, and
# an echo targets exactly one room. base_url is the resolved client API
# base (well-known delegation), homeserver the value the user typed.
db.execute("""
CREATE TABLE IF NOT EXISTS matrix_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
homeserver TEXT NOT NULL,
base_url TEXT NOT NULL DEFAULT '',
access_token TEXT NOT NULL,
matrix_user_id TEXT DEFAULT '',
room_id TEXT NOT NULL,
room_alias TEXT DEFAULT '',
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, homeserver, room_id)
)
""")
# One row per Discord webhook URL: the URL is the credential and the
# posting endpoint in one. channel_id comes from Discord's metadata
# endpoint and is display-only. UNIQUE(user_id, webhook_url) makes
# reconnecting the same webhook (to refresh the name) an in-place
# update instead of a duplicate row.
db.execute("""
CREATE TABLE IF NOT EXISTS discord_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
webhook_url TEXT NOT NULL,
webhook_url_hash TEXT DEFAULT '',
channel_id TEXT DEFAULT '',
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# The webhook URL is encrypted at rest (non-deterministic Fernet), so
# the dedup key is a deterministic digest of the plaintext URL.
_add_column_if_missing(
db, "discord_accounts", "webhook_url_hash", "TEXT DEFAULT ''"
)
for _row in db.execute(
"SELECT id, webhook_url FROM discord_accounts"
" WHERE webhook_url_hash = '' OR webhook_url_hash IS NULL"
).fetchall():
db.execute(
"UPDATE discord_accounts SET webhook_url_hash = ? WHERE id = ?",
(hash_secret(decrypt_secret(_row["webhook_url"])), _row["id"]),
)
_dedupe_discord_webhook_hashes(db)
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_discord_user_hash"
" ON discord_accounts(user_id, webhook_url_hash)"
)
# One row per Telegram chat: the bot token is the credential, the
# chat id the target. UNIQUE(user_id, chat_id) makes reconnecting
# the same chat (e.g. after a token rotation) an in-place update
# instead of a duplicate row. The token is encrypted at rest.
db.execute("""
CREATE TABLE IF NOT EXISTS telegram_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
bot_token TEXT NOT NULL,
chat_id TEXT NOT NULL,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_telegram_user_chat"
" ON telegram_accounts(user_id, chat_id)"
)
# One row per generic webhook endpoint: url is the target, headers a
# JSON object of custom HTTP headers (credentials live there — never
# rendered back). UNIQUE(user_id, url) makes reconnecting the same
# endpoint an in-place update instead of a duplicate row.
db.execute("""
CREATE TABLE IF NOT EXISTS webhook_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
url TEXT NOT NULL,
headers TEXT NOT NULL DEFAULT '{}',
body_template TEXT NOT NULL DEFAULT '',
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, url)
)
""")
# One row per target repository: the token is the credential
# (encrypted at rest by the app), repo is normalized owner/repo.
# UNIQUE(user_id, repo, branch) makes reconnecting the same target
# an in-place update instead of a duplicate row.
db.execute("""
CREATE TABLE IF NOT EXISTS github_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
repo TEXT NOT NULL,
branch TEXT NOT NULL DEFAULT 'main',
token TEXT NOT NULL,
committer_name TEXT DEFAULT '',
committer_email TEXT DEFAULT '',
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_github_user_repo_branch"
" ON github_accounts(user_id, repo, branch)"
)
db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL DEFAULT '',
plan TEXT NOT NULL DEFAULT 'trial',
trial_ends_at TIMESTAMP,
email_verified INTEGER NOT NULL DEFAULT 0,
suspended INTEGER NOT NULL DEFAULT 0,
is_admin INTEGER NOT NULL DEFAULT 0,
session_epoch INTEGER NOT NULL DEFAULT 0,
stripe_customer_id TEXT DEFAULT '',
stripe_subscription_id TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Single-tenant placeholder: all data belongs to user 1 when the
# app runs without accounts (FEEDECHO_MODE=single).
db.execute("INSERT OR IGNORE INTO users (id, email) VALUES (1, 'local')")
# Owned tables carry user_id. Existing single-tenant databases
# backfill to user 1 via the column default.
for table in ("accounts", "feeds", "echoes", "email_accounts", "bluesky_accounts", "microblog_accounts", "matrix_accounts", "discord_accounts", "telegram_accounts", "webhook_accounts", "github_accounts"):
_add_column_if_missing(
db, table, "user_id", "INTEGER NOT NULL DEFAULT 1"
)
# Admin flag for hosted mode (A2); never auto-granted.
_add_column_if_missing(db, "users", "is_admin", "INTEGER NOT NULL DEFAULT 0")
# Session epoch: bumped on password reset, invalidating all prior
# session cookies for the user. Added after the users table shipped,
# so sqlite databases created before it need the backfill too — the
# multi-mode auth middleware and login both SELECT this column, and
# a self-hoster switching an existing sqlite DB to multi mode would
# otherwise hit "no such column: session_epoch".
_add_column_if_missing(db, "users", "session_epoch", "INTEGER NOT NULL DEFAULT 0")
# Migrate single-column UNIQUE constraints to per-user composite
# constraints so different tenants can use the same destination
# email/handle. Identical behavior in single mode (user_id is
# always 1). Recreate-table pattern, guarded by _has_unique_on so
# it is idempotent across restarts.
if not _has_unique_on(db, "email_accounts", {"user_id", "email"}):
db.execute("DROP TABLE IF EXISTS email_accounts_legacy")
db.execute(
"ALTER TABLE email_accounts RENAME TO email_accounts_legacy"
)
db.execute("""
CREATE TABLE email_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, email)
)
""")
db.execute(
"INSERT INTO email_accounts (id, name, email, user_id, created_at)"
" SELECT id, name, email, user_id, created_at"
" FROM email_accounts_legacy"
)
db.execute("DROP TABLE email_accounts_legacy")
if not _has_unique_on(db, "bluesky_accounts", {"user_id", "handle"}):
db.execute("DROP TABLE IF EXISTS bluesky_accounts_legacy")
db.execute(
"ALTER TABLE bluesky_accounts RENAME TO bluesky_accounts_legacy"
)
db.execute("""
CREATE TABLE bluesky_accounts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
handle TEXT NOT NULL,
app_password TEXT NOT NULL,
did TEXT DEFAULT '',
pds TEXT DEFAULT '',
access_jwt TEXT DEFAULT '',
refresh_jwt TEXT DEFAULT '',
session_expires_at TIMESTAMP,
user_id INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, handle)
)
""")
db.execute(
"INSERT INTO bluesky_accounts (id, name, handle, app_password,"
" did, pds, access_jwt, refresh_jwt, session_expires_at,"
" user_id, created_at)"
" SELECT id, name, handle, app_password, did, pds, access_jwt,"
" refresh_jwt, session_expires_at, user_id, created_at"
" FROM bluesky_accounts_legacy"
)
db.execute("DROP TABLE bluesky_accounts_legacy")
db.execute("""
CREATE TABLE IF NOT EXISTS settings (
user_id INTEGER NOT NULL DEFAULT 1,
key TEXT NOT NULL,
value TEXT,
PRIMARY KEY (user_id, key)
)
""")
if "user_id" not in _column_names(db, "settings"):
# Migrate legacy single-key settings tables: backfill to user 1.
# The guard keeps the app startable if a partial migration or
# backup restore left a settings_legacy table behind.
db.execute("DROP TABLE IF EXISTS settings_legacy")
db.execute("ALTER TABLE settings RENAME TO settings_legacy")
db.execute("""
CREATE TABLE settings (
user_id INTEGER NOT NULL DEFAULT 1,
key TEXT NOT NULL,
value TEXT,
PRIMARY KEY (user_id, key)
)
""")
db.execute(
"INSERT INTO settings (user_id, key, value) SELECT 1, key, value FROM settings_legacy"
)
db.execute("DROP TABLE settings_legacy")
db.execute("""
CREATE TABLE IF NOT EXISTS posted_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_title TEXT,
item_url TEXT,
status TEXT NOT NULL,
error_message TEXT,
claimed_at TIMESTAMP,
claim_token TEXT,
attempt_count INTEGER NOT NULL DEFAULT 0,
post_url TEXT,
next_retry_at TIMESTAMP,
posted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
_add_column_if_missing(db, "posted_items", "claimed_at", "TIMESTAMP")
_add_column_if_missing(db, "posted_items", "claim_token", "TEXT")
_add_column_if_missing(db, "posted_items", "post_url", "TEXT")
_add_column_if_missing(
db,
"posted_items",
"attempt_count",
"INTEGER NOT NULL DEFAULT 0",
)
_add_column_if_missing(db, "posted_items", "next_retry_at", "TIMESTAMP")
db.execute("""
CREATE TABLE IF NOT EXISTS queued_posts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL DEFAULT 1,
feed_item_id INTEGER,
item_id TEXT NOT NULL,
feed_id INTEGER NOT NULL,
destination_type TEXT NOT NULL,
destination_id INTEGER NOT NULL,
content TEXT NOT NULL,
visibility TEXT DEFAULT 'public',
attach_image INTEGER NOT NULL DEFAULT 0,
image_alt TEXT,
scheduled_at TIMESTAMP NOT NULL,
status TEXT NOT NULL DEFAULT 'queued',
claim_token TEXT,
claimed_at TIMESTAMP,
attempt_count INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
posted_item_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_queued_posts_due
ON queued_posts(status, scheduled_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_queued_posts_user
ON queued_posts(user_id, scheduled_at)
""")
_add_column_if_missing(db, "queued_posts", "echo_id", "INTEGER")
db.execute("""
CREATE TABLE IF NOT EXISTS queue_settings (
user_id INTEGER NOT NULL,
destination_key TEXT NOT NULL,
queue_interval_minutes INTEGER NOT NULL DEFAULT 240,
queue_active_hours TEXT DEFAULT '09:00-21:00',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, destination_key)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS oauth_apps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instance TEXT NOT NULL UNIQUE,
client_id TEXT NOT NULL,
client_secret TEXT NOT NULL,
website TEXT,
redirect_uris TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Apps registered before these were recorded carry NULL, which forces
# one re-registration: the post link stops pointing at the old
# example.com placeholder, and a drifted callback URL gets refreshed.
_add_column_if_missing(db, "oauth_apps", "website", "TEXT")
_add_column_if_missing(db, "oauth_apps", "redirect_uris", "TEXT")
db.execute("""
CREATE TABLE IF NOT EXISTS oauth_states (
nonce TEXT PRIMARY KEY,
instance TEXT NOT NULL,
session_binding TEXT NOT NULL,
expires_at TIMESTAMP NOT NULL,
user_id INTEGER,
consumed_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""")
_add_column_if_missing(db, "oauth_states", "user_id", "INTEGER")
# v1.65.0: per-webhook custom JSON body template ('' = default payload).
_add_column_if_missing(
db, "webhook_accounts", "body_template", "TEXT NOT NULL DEFAULT ''"
)
db.execute("""
CREATE TABLE IF NOT EXISTS invite_codes (
code TEXT PRIMARY KEY,
created_by INTEGER,
used_by INTEGER,
used_at TIMESTAMP,
revoked INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_echo
ON posted_items(echo_id, posted_at DESC)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_posted_items_echo_item
ON posted_items(echo_id, item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_item_id
ON posted_items(item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_reclaim
ON posted_items(status, claimed_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_echoes_feed
ON echoes(feed_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feeds_lease
ON feeds(lease_expires_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_oauth_states_expiry
ON oauth_states(expires_at)
""")
# DB-level dedup backing for import_export.py's check-then-insert
# logic: without this, two concurrent requests (e.g. two imports, or
# an import racing a manual add) for the same user+url/instance+
# username can both pass the SELECT before either INSERT commits,
# producing duplicate rows. A unique index (rather than an inline
# UNIQUE + table rebuild) applies retroactively to already-deployed
# databases without a risky recreate-table migration; run after
# user_id/username are guaranteed to exist on both tables above.
# Partial (WHERE deleted_at IS NULL): feeds are soft-deleted and
# never purged, and import_export.py's own dedup lookup already
# ignores deleted_at IS NOT NULL rows (see get_feeds/its INSERT
# dedup query) — a non-partial index would otherwise permanently
# block re-adding a feed URL after it was deleted.
# Pre-upgrade inserts were unguarded (oauth_callback re-connected
# via plain INSERT), so already-deployed databases can hold exact
# duplicate rows; dedupe them first or CREATE UNIQUE INDEX fails at
# boot (the v1.39.0 Discord-hash failure class).
_dedupe_feeds_for_unique_index(db)
_dedupe_accounts_for_unique_index(db)
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_feeds_user_url
ON feeds(user_id, url)
WHERE deleted_at IS NULL
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feeds_poke_token
ON feeds(poke_token)
WHERE poke_token IS NOT NULL AND deleted_at IS NULL
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_accounts_user_instance_username
ON accounts(user_id, instance, username)
""")
# Best-effort cleanup of expired/consumed state rows.
db.execute("""
DELETE FROM oauth_states
WHERE consumed_at IS NOT NULL
OR expires_at <= datetime('now', '-1 day')
""")
_init_shared_tables(db)
def _init_shared_tables(db) -> None:
"""Deployment-wide tables, identical on both dialects.
system_settings holds admin email/SMTP config; email_tokens holds
single-use email-flow tokens (verification, password reset).
scheduler_leases ensures only one process runs each flush job.
Deliberately a separate table from per-user `settings`: system values
(verification/reset SMTP) belong to the deployment, not to any tenant,
and must be unreadable/unwritable by tenant-scoped settings routes.
"""
db.execute("""
CREATE TABLE IF NOT EXISTS system_settings (
key TEXT PRIMARY KEY,
value TEXT
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS scheduler_leases (
job_name TEXT PRIMARY KEY,
instance_id TEXT NOT NULL,
expires_at TIMESTAMP NOT NULL
)
""")
if dialect() == "postgres":
db.execute("""
CREATE TABLE IF NOT EXISTS email_tokens (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL,
token_hash TEXT NOT NULL,
purpose TEXT NOT NULL,
expires_at TIMESTAMP NOT NULL,
consumed_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
else:
db.execute("""
CREATE TABLE IF NOT EXISTS email_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
token_hash TEXT NOT NULL,
purpose TEXT NOT NULL,
expires_at TIMESTAMP NOT NULL,
consumed_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_email_tokens_user
ON email_tokens(user_id, purpose)
""")
# At most ONE unconsumed token per (user, purpose): the partial unique
# index serializes concurrent issue_token calls at the DB level on both
# dialects (the issuer retries once on conflict).
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_email_tokens_active
ON email_tokens(user_id, purpose)
WHERE consumed_at IS NULL
""")
def init_db_postgres() -> None:
"""Create the PostgreSQL application schema (multi-tenant).
Mirrors init_db_sqlite at the post-migration state: fresh Postgres
installs get the final schema directly. No singleton user row — real
accounts come from /register. Keep the table set in sync with
init_db_sqlite; the schema-parity test enforces it.
"""
with get_db() as db:
db.execute("""
CREATE TABLE IF NOT EXISTS users (
id BIGSERIAL PRIMARY KEY,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL DEFAULT '',
plan TEXT NOT NULL DEFAULT 'trial',
trial_ends_at TIMESTAMP,
email_verified INTEGER NOT NULL DEFAULT 0,
suspended INTEGER NOT NULL DEFAULT 0,
is_admin INTEGER NOT NULL DEFAULT 0,
session_epoch INTEGER NOT NULL DEFAULT 0,
stripe_customer_id TEXT DEFAULT '',
stripe_subscription_id TEXT DEFAULT '',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS webhook_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
url TEXT NOT NULL,
headers TEXT NOT NULL DEFAULT '{}',
body_template TEXT NOT NULL DEFAULT '',
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, url)
)
""")
# GitHub destination — see init_db_sqlite for the column notes.
db.execute("""
CREATE TABLE IF NOT EXISTS github_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
repo TEXT NOT NULL,
branch TEXT NOT NULL DEFAULT 'main',
token TEXT NOT NULL,
committer_name TEXT DEFAULT '',
committer_email TEXT DEFAULT '',
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_github_user_repo_branch"
" ON github_accounts(user_id, repo, branch)"
)
db.execute("""
CREATE TABLE IF NOT EXISTS accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
username TEXT DEFAULT '',
instance TEXT NOT NULL,
access_token TEXT NOT NULL,
booster_enabled INTEGER NOT NULL DEFAULT 0,
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Pre-existing hosted databases created before `username` shipped
# need the same backfill-from-`name` treatment as sqlite: fresh
# installs get the column from CREATE TABLE above, but that's a
# no-op on a table that already exists without it.
account_columns = _column_names(db, "accounts")
if "username" not in account_columns:
db.execute("ALTER TABLE accounts ADD COLUMN username TEXT DEFAULT ''")
rows = db.execute("SELECT id, name FROM accounts").fetchall()
import re
for row in rows:
match = re.search(r"\(([^)]+)\)$", row["name"] or "")
username = (
match.group(1)
if match
else (row["name"] or "unknown")
)
db.execute(
"UPDATE accounts SET username = ? WHERE id = ?",
(username, row["id"]),
)
_add_column_if_missing(db, "accounts", "booster_enabled", "INTEGER NOT NULL DEFAULT 0")
db.execute("""
CREATE TABLE IF NOT EXISTS saved_searches (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL DEFAULT 1,
name TEXT NOT NULL,
query TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_saved_searches_user_name
ON saved_searches(user_id, LOWER(name))
""")
db.execute("""
CREATE TABLE IF NOT EXISTS folders (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL DEFAULT 1,
name TEXT NOT NULL,
position INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_folders_user_name
ON folders(user_id, LOWER(name))
""")
db.execute("""
CREATE TABLE IF NOT EXISTS feeds (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
url TEXT NOT NULL,
feed_type TEXT DEFAULT 'rss',
poll_interval INTEGER DEFAULT 15,
last_fetched TIMESTAMP,
last_item_id TEXT,
lease_token TEXT,
lease_expires_at TIMESTAMP,
paused INTEGER NOT NULL DEFAULT 0,
read_enabled INTEGER NOT NULL DEFAULT 0,
ai_alt_override INTEGER NOT NULL DEFAULT 0,
mute_keywords TEXT DEFAULT '',
last_error TEXT,
etag TEXT,
last_modified TEXT,
poke_token TEXT,
last_poked_at TIMESTAMP,
deleted_at TIMESTAMP,
user_id BIGINT NOT NULL DEFAULT 1,
folder_id INTEGER,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Reader toggle for already-deployed hosted databases (fresh installs
# get the column from the CREATE TABLE above).
_add_column_if_missing(db, "feeds", "read_enabled", "INTEGER NOT NULL DEFAULT 0")
# Per-feed toggle: when set, AI-generated alt text wins over the
# feed-provided alt text on image echoes (default off = feed alt wins).
_add_column_if_missing(db, "feeds", "ai_alt_override", "INTEGER NOT NULL DEFAULT 0")
_add_column_if_missing(db, "feeds", "mute_keywords", "TEXT DEFAULT ''")
_add_column_if_missing(db, "feeds", "last_error", "TEXT")
_add_column_if_missing(db, "feeds", "folder_id", "BIGINT")
_add_column_if_missing(db, "feeds", "etag", "TEXT")
_add_column_if_missing(db, "feeds", "last_modified", "TEXT")
_add_column_if_missing(db, "feeds", "poke_token", "TEXT")
_add_column_if_missing(db, "feeds", "last_poked_at", "TIMESTAMP")
_add_column_if_missing(db, "feeds", "lease_token", "TEXT")
_add_column_if_missing(db, "feeds", "lease_expires_at", "TIMESTAMP")
_add_column_if_missing(db, "feeds", "paused", "INTEGER NOT NULL DEFAULT 0")
# Soft-delete marker: feeds are never hard-deleted by the app so that
# echo configuration and posted-item history survive as an audit trail.
_add_column_if_missing(db, "feeds", "deleted_at", "TIMESTAMP")
db.execute("""
CREATE TABLE IF NOT EXISTS feed_items (
id BIGSERIAL PRIMARY KEY,
feed_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
title TEXT,
link TEXT,
summary TEXT,
content TEXT,
content_text TEXT,
content_link TEXT,
content_html TEXT,
categories TEXT,
author TEXT,
image_url TEXT,
image_alt TEXT,
image_urls TEXT,
enclosure_url TEXT,
published_at TIMESTAMP,
is_read INTEGER NOT NULL DEFAULT 0,
starred INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (feed_id) REFERENCES feeds(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_feed_items_feed_item
ON feed_items(feed_id, item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feed_items_feed_pub
ON feed_items(feed_id, published_at DESC)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feed_items_feed_read
ON feed_items(feed_id, is_read)
""")
_add_column_if_missing(db, "feed_items", "content_text", "TEXT")
# v1.66.0: sanitized HTML passthrough for {{ content_html }}.
_add_column_if_missing(db, "feed_items", "content_html", "TEXT")
# v1.73.2: RSS/Atom category domains for {{ categories }} (JSON dict).
_add_column_if_missing(db, "feed_items", "categories", "TEXT")
_add_column_if_missing(db, "feed_items", "image_url", "TEXT")
_add_column_if_missing(db, "feed_items", "image_alt", "TEXT")
_add_column_if_missing(db, "feed_items", "image_urls", "TEXT")
_add_column_if_missing(db, "feed_items", "enclosure_url", "TEXT")
_add_column_if_missing(db, "feed_items", "starred", "INTEGER NOT NULL DEFAULT 0")
db.execute("""
CREATE TABLE IF NOT EXISTS echoes (
id BIGSERIAL PRIMARY KEY,
feed_id INTEGER NOT NULL,
destination_type TEXT NOT NULL DEFAULT 'mastodon',
destination_id INTEGER NOT NULL,
template TEXT NOT NULL DEFAULT '{{ title }} {{ link }}',
visibility TEXT DEFAULT 'public',
enabled INTEGER DEFAULT 1,
user_id BIGINT NOT NULL DEFAULT 1,
filter_keywords TEXT DEFAULT '',
filter_mode TEXT NOT NULL DEFAULT 'exclude',
content_warning TEXT DEFAULT '',
attach_image INTEGER NOT NULL DEFAULT 0,
render_html INTEGER NOT NULL DEFAULT 0,
use_markdown INTEGER NOT NULL DEFAULT 0,
delivery_mode TEXT NOT NULL DEFAULT 'instant',
drip_limit INTEGER NOT NULL DEFAULT 0,
one_shot INTEGER NOT NULL DEFAULT 0,
booster_enabled INTEGER NOT NULL DEFAULT 0,
deleted_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (feed_id) REFERENCES feeds(id) ON DELETE CASCADE
)
""")
_add_column_if_missing(db, "echoes", "one_shot", "INTEGER NOT NULL DEFAULT 0")
_add_column_if_missing(db, "echoes", "booster_enabled", "INTEGER NOT NULL DEFAULT 0")
# v1.67.0: email echoes may render the body as HTML ({{ content_html }}).
_add_column_if_missing(
db, "echoes", "render_html", "INTEGER NOT NULL DEFAULT 0"
)
_add_column_if_missing(
db, "echoes", "use_markdown", "INTEGER NOT NULL DEFAULT 0"
)
# GitHub destination: per-echo file path and commit message templates.
_add_column_if_missing(
db, "echoes", "github_path", "TEXT NOT NULL DEFAULT ''"
)
_add_column_if_missing(
db, "echoes", "github_commit", "TEXT NOT NULL DEFAULT ''"
)
_add_column_if_missing(db, "echoes", "filter_keywords", "TEXT DEFAULT ''")
_add_column_if_missing(
db, "echoes", "filter_mode", "TEXT NOT NULL DEFAULT 'exclude'"
)
_add_column_if_missing(db, "echoes", "content_warning", "TEXT DEFAULT ''")
_add_column_if_missing(
db, "echoes", "attach_image", "INTEGER NOT NULL DEFAULT 0"
)
_add_column_if_missing(
db, "echoes", "delivery_mode", "TEXT NOT NULL DEFAULT 'instant'"
)
_add_column_if_missing(
db, "echoes", "drip_limit", "INTEGER NOT NULL DEFAULT 0"
)
_add_column_if_missing(db, "echoes", "deleted_at", "TIMESTAMP")
db.execute("""
CREATE TABLE IF NOT EXISTS digest_items (
id BIGSERIAL PRIMARY KEY,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_title TEXT,
item_url TEXT,
rendered_content TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(echo_id, item_id),
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_digest_items_echo
ON digest_items(echo_id)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS drip_items (
id BIGSERIAL PRIMARY KEY,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_json TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
queued_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(echo_id, item_id),
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_drip_items_echo
ON drip_items(echo_id)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS email_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL,
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, email)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS bluesky_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
handle TEXT NOT NULL,
app_password TEXT NOT NULL,
did TEXT DEFAULT '',
pds TEXT DEFAULT '',
access_jwt TEXT DEFAULT '',
refresh_jwt TEXT DEFAULT '',
session_expires_at TIMESTAMP,
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, handle)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS microblog_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
uid TEXT NOT NULL,
token TEXT NOT NULL,
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, uid)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS matrix_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
homeserver TEXT NOT NULL,
base_url TEXT NOT NULL DEFAULT '',
access_token TEXT NOT NULL,
matrix_user_id TEXT DEFAULT '',
room_id TEXT NOT NULL,
room_alias TEXT DEFAULT '',
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, homeserver, room_id)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS discord_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
webhook_url TEXT NOT NULL,
webhook_url_hash TEXT DEFAULT '',
channel_id TEXT DEFAULT '',
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
_add_column_if_missing(
db, "discord_accounts", "webhook_url_hash", "TEXT DEFAULT ''"
)
for _row in db.execute(
"SELECT id, webhook_url FROM discord_accounts"
" WHERE webhook_url_hash = '' OR webhook_url_hash IS NULL"
).fetchall():
db.execute(
"UPDATE discord_accounts SET webhook_url_hash = ? WHERE id = ?",
(hash_secret(decrypt_secret(_row["webhook_url"])), _row["id"]),
)
_dedupe_discord_webhook_hashes(db)
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_discord_user_hash"
" ON discord_accounts(user_id, webhook_url_hash)"
)
db.execute("""
CREATE TABLE IF NOT EXISTS telegram_accounts (
id BIGSERIAL PRIMARY KEY,
name TEXT NOT NULL,
bot_token TEXT NOT NULL,
chat_id TEXT NOT NULL,
user_id BIGINT NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_telegram_user_chat"
" ON telegram_accounts(user_id, chat_id)"
)
db.execute("""
CREATE TABLE IF NOT EXISTS settings (
user_id BIGINT NOT NULL DEFAULT 1,
key TEXT NOT NULL,
value TEXT,
PRIMARY KEY (user_id, key)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS posted_items (
id BIGSERIAL PRIMARY KEY,
echo_id INTEGER NOT NULL,
item_id TEXT NOT NULL,
item_title TEXT,
item_url TEXT,
status TEXT NOT NULL,
error_message TEXT,
claimed_at TIMESTAMP,
claim_token TEXT,
attempt_count INTEGER NOT NULL DEFAULT 0,
post_url TEXT,
next_retry_at TIMESTAMP,
posted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (echo_id) REFERENCES echoes(id) ON DELETE CASCADE
)
""")
_add_column_if_missing(db, "posted_items", "claimed_at", "TIMESTAMP")
_add_column_if_missing(db, "posted_items", "claim_token", "TEXT")
_add_column_if_missing(db, "posted_items", "post_url", "TEXT")
_add_column_if_missing(
db,
"posted_items",
"attempt_count",
"INTEGER NOT NULL DEFAULT 0",
)
_add_column_if_missing(db, "posted_items", "next_retry_at", "TIMESTAMP")
db.execute("""
CREATE TABLE IF NOT EXISTS queued_posts (
id BIGSERIAL PRIMARY KEY,
user_id BIGINT NOT NULL DEFAULT 1,
feed_item_id BIGINT,
item_id TEXT NOT NULL,
feed_id BIGINT NOT NULL,
destination_type TEXT NOT NULL,
destination_id BIGINT NOT NULL,
content TEXT NOT NULL,
visibility TEXT DEFAULT 'public',
attach_image INTEGER NOT NULL DEFAULT 0,
image_alt TEXT,
scheduled_at TIMESTAMP NOT NULL,
status TEXT NOT NULL DEFAULT 'queued',
claim_token TEXT,
claimed_at TIMESTAMP,
attempt_count INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
posted_item_id BIGINT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_queued_posts_due
ON queued_posts(status, scheduled_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_queued_posts_user
ON queued_posts(user_id, scheduled_at)
""")
_add_column_if_missing(db, "queued_posts", "echo_id", "INTEGER")
db.execute("""
CREATE TABLE IF NOT EXISTS queue_settings (
user_id BIGINT NOT NULL,
destination_key TEXT NOT NULL,
queue_interval_minutes INTEGER NOT NULL DEFAULT 240,
queue_active_hours TEXT DEFAULT '09:00-21:00',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (user_id, destination_key)
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS oauth_apps (
id BIGSERIAL PRIMARY KEY,
instance TEXT NOT NULL UNIQUE,
client_id TEXT NOT NULL,
client_secret TEXT NOT NULL,
website TEXT,
redirect_uris TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# See the sqlite branch: NULL columns force one re-registration so
# existing installs stop advertising the example.com placeholder.
_add_column_if_missing(db, "oauth_apps", "website", "TEXT")
_add_column_if_missing(db, "oauth_apps", "redirect_uris", "TEXT")
db.execute("""
CREATE TABLE IF NOT EXISTS oauth_states (
nonce TEXT PRIMARY KEY,
instance TEXT NOT NULL,
session_binding TEXT NOT NULL,
expires_at TIMESTAMP NOT NULL,
user_id BIGINT,
consumed_at TIMESTAMP,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE TABLE IF NOT EXISTS invite_codes (
code TEXT PRIMARY KEY,
created_by BIGINT,
used_by BIGINT,
used_at TIMESTAMP,
revoked INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_echo
ON posted_items(echo_id, posted_at DESC)
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_posted_items_echo_item
ON posted_items(echo_id, item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_item_id
ON posted_items(item_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_posted_items_reclaim
ON posted_items(status, claimed_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_echoes_feed
ON echoes(feed_id)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feeds_lease
ON feeds(lease_expires_at)
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_oauth_states_expiry
ON oauth_states(expires_at)
""")
# DB-level dedup backing for import_export.py's check-then-insert
# logic: without this, two concurrent requests (e.g. two imports, or
# an import racing a manual add) for the same user+url/instance+
# username can both pass the SELECT before either INSERT commits,
# producing duplicate rows. A unique index (rather than an inline
# UNIQUE + table rebuild) applies retroactively to already-deployed
# databases without a risky recreate-table migration; run after
# user_id/username are guaranteed to exist on both tables above.
# Partial (WHERE deleted_at IS NULL): feeds are soft-deleted and
# never purged, and import_export.py's own dedup lookup already
# ignores deleted_at IS NOT NULL rows (see get_feeds/its INSERT
# dedup query) — a non-partial index would otherwise permanently
# block re-adding a feed URL after it was deleted.
# Pre-upgrade inserts were unguarded (oauth_callback re-connected
# via plain INSERT), so already-deployed databases can hold exact
# duplicate rows; dedupe them first or CREATE UNIQUE INDEX fails at
# boot (the v1.39.0 Discord-hash failure class).
_dedupe_feeds_for_unique_index(db)
_dedupe_accounts_for_unique_index(db)
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_feeds_user_url
ON feeds(user_id, url)
WHERE deleted_at IS NULL
""")
db.execute("""
CREATE INDEX IF NOT EXISTS idx_feeds_poke_token
ON feeds(poke_token)
WHERE poke_token IS NOT NULL AND deleted_at IS NULL
""")
db.execute("""
CREATE UNIQUE INDEX IF NOT EXISTS idx_accounts_user_instance_username
ON accounts(user_id, instance, username)
""")
# Best-effort cleanup of expired/consumed state rows.
db.execute("""
DELETE FROM oauth_states
WHERE consumed_at IS NOT NULL
OR expires_at <= NOW() - INTERVAL '1 day'
""")
# Admin flag for hosted mode (A2); never auto-granted.
_add_column_if_missing(db, "users", "is_admin", "INTEGER NOT NULL DEFAULT 0")
# Session epoch: bumped on password reset, invalidating all prior
# session cookies for the user.
_add_column_if_missing(db, "users", "session_epoch", "INTEGER NOT NULL DEFAULT 0")
# v1.65.0: per-webhook custom JSON body template ('' = default payload).
_add_column_if_missing(
db, "webhook_accounts", "body_template", "TEXT NOT NULL DEFAULT ''"
)
_init_shared_tables(db)
init_db()