244 lines
14 KiB
Python
244 lines
14 KiB
Python
"""Durable consent, schedule and delivery records for personal email recaps."""
|
|
|
|
import hashlib
|
|
import secrets
|
|
import sqlite3
|
|
import uuid
|
|
from contextlib import closing, contextmanager
|
|
from datetime import datetime
|
|
|
|
from .. import db
|
|
from .monthly_reports import shift_month
|
|
from . import email_queue
|
|
|
|
|
|
def init_schema(conn: sqlite3.Connection) -> None:
|
|
for statement in (
|
|
"""CREATE TABLE IF NOT EXISTS email_recap_settings (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1), enabled INTEGER NOT NULL DEFAULT 0,
|
|
day INTEGER NOT NULL DEFAULT 2, hour INTEGER NOT NULL DEFAULT 9,
|
|
public_url TEXT NOT NULL DEFAULT '', next_send_at REAL)""",
|
|
"INSERT OR IGNORE INTO email_recap_settings (id) VALUES (1)",
|
|
"""CREATE TABLE IF NOT EXISTS email_recap_subscriptions (
|
|
user_id INTEGER PRIMARY KEY, state TEXT NOT NULL, email TEXT NOT NULL,
|
|
identity_source TEXT NOT NULL, identity_id TEXT NOT NULL, version TEXT NOT NULL,
|
|
confirmation_hash TEXT UNIQUE, confirmation_expires REAL, requested_at REAL NOT NULL,
|
|
confirmed_at REAL, unsubscribe_token TEXT NOT NULL UNIQUE)""",
|
|
"""CREATE TABLE IF NOT EXISTS email_recap_deliveries (
|
|
id TEXT PRIMARY KEY, dedupe_key TEXT NOT NULL UNIQUE, user_id INTEGER NOT NULL,
|
|
month TEXT NOT NULL, kind TEXT NOT NULL, email TEXT NOT NULL,
|
|
subscription_version TEXT NOT NULL, public_url TEXT NOT NULL,
|
|
state TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0,
|
|
created_at REAL NOT NULL, updated_at REAL NOT NULL, next_attempt_at REAL NOT NULL,
|
|
claim TEXT, lease_until REAL, detail TEXT NOT NULL DEFAULT '')""",
|
|
"CREATE INDEX IF NOT EXISTS idx_email_recap_queue ON email_recap_deliveries (state, next_attempt_at)",
|
|
"""CREATE TRIGGER IF NOT EXISTS email_recap_account_changed AFTER UPDATE OF email, is_blocked ON users
|
|
WHEN LOWER(TRIM(COALESCE(NEW.email, ''))) != LOWER(TRIM(COALESCE(OLD.email, '')))
|
|
OR NEW.is_blocked = 1
|
|
BEGIN UPDATE email_recap_subscriptions SET state = 'off', confirmation_hash = NULL,
|
|
confirmed_at = NULL WHERE user_id = NEW.id; END""",
|
|
"""CREATE TRIGGER IF NOT EXISTS email_recap_account_deleted AFTER DELETE ON users
|
|
BEGIN DELETE FROM email_recap_subscriptions WHERE user_id = OLD.id;
|
|
UPDATE email_recap_deliveries SET state = 'cancelled', detail = 'Account removed.'
|
|
WHERE user_id = OLD.id AND state IN ('queued', 'retry', 'preparing'); END""",
|
|
"""CREATE TRIGGER IF NOT EXISTS email_recap_identity_changed AFTER UPDATE ON jellyfin_user_links
|
|
WHEN NEW.jellyfin_user_id != OLD.jellyfin_user_id OR NEW.source != OLD.source
|
|
OR NEW.local_user_id != OLD.local_user_id
|
|
BEGIN UPDATE email_recap_subscriptions SET state = 'off', confirmation_hash = NULL,
|
|
confirmed_at = NULL WHERE user_id = OLD.local_user_id; END""",
|
|
"""CREATE TRIGGER IF NOT EXISTS email_recap_identity_deleted AFTER DELETE ON jellyfin_user_links
|
|
BEGIN UPDATE email_recap_subscriptions SET state = 'off', confirmation_hash = NULL,
|
|
confirmed_at = NULL WHERE user_id = OLD.local_user_id; END""",
|
|
):
|
|
conn.execute(statement)
|
|
columns = {row[1] for row in conn.execute('PRAGMA table_info(email_recap_subscriptions)')}
|
|
if 'automatic_monthly' not in columns:
|
|
conn.execute('ALTER TABLE email_recap_subscriptions ADD COLUMN automatic_monthly INTEGER NOT NULL DEFAULT 1')
|
|
|
|
|
|
@contextmanager
|
|
def transaction():
|
|
with closing(db._connect()) as conn, conn:
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("BEGIN IMMEDIATE")
|
|
yield conn
|
|
|
|
|
|
def read_one(sql: str, args=()) -> dict | None:
|
|
with closing(db._connect()) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute(sql, args).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def settings() -> dict:
|
|
row = read_one("SELECT * FROM email_recap_settings WHERE id = 1")
|
|
return {key: (bool(value) if key == "enabled" else value) for key, value in row.items() if key != "id"}
|
|
|
|
|
|
def next_due(now: datetime, day: int, hour: int) -> datetime:
|
|
due = shift_month(now, 0).replace(day=day, hour=hour)
|
|
return due if due > now else shift_month(now, 1).replace(day=day, hour=hour)
|
|
|
|
|
|
def save_settings(values: dict, now: datetime) -> dict:
|
|
with transaction() as conn:
|
|
old = dict(conn.execute("SELECT * FROM email_recap_settings WHERE id = 1").fetchone())
|
|
changed = any(old[key] != values[key] for key in ("day", "hour", "public_url"))
|
|
due = old["next_send_at"]
|
|
if not values["enabled"]:
|
|
due = None
|
|
elif not old["enabled"] or changed:
|
|
due = next_due(now, values["day"], values["hour"]).timestamp()
|
|
conn.execute("UPDATE email_recap_settings SET enabled=?, day=?, hour=?, public_url=?, next_send_at=? WHERE id=1",
|
|
(values["enabled"], values["day"], values["hour"], values["public_url"], due))
|
|
if not values["enabled"] or changed:
|
|
conn.execute("""UPDATE email_recap_deliveries SET state='cancelled', detail='Schedule paused or changed.', updated_at=?
|
|
WHERE kind='scheduled' AND state IN ('queued', 'retry', 'preparing')""", (now.timestamp(),))
|
|
return settings()
|
|
|
|
|
|
def subscription(user_id: int) -> dict | None:
|
|
return read_one("SELECT * FROM email_recap_subscriptions WHERE user_id=?", (user_id,))
|
|
|
|
|
|
def disable(user_id: int) -> None:
|
|
with transaction() as conn:
|
|
conn.execute("UPDATE email_recap_subscriptions SET state='off', confirmation_hash=NULL, confirmed_at=NULL WHERE user_id=?", (user_id,))
|
|
conn.execute("""UPDATE email_recap_deliveries SET state='cancelled', detail='Email recaps turned off.'
|
|
WHERE user_id=? AND state IN ('queued', 'retry', 'preparing')""", (user_id,))
|
|
|
|
|
|
def request_confirmation(user: dict, source: str, identity: str, now: float, automatic_monthly: bool = True) -> str:
|
|
token = secrets.token_urlsafe(32)
|
|
with transaction() as conn:
|
|
old = conn.execute("SELECT * FROM email_recap_subscriptions WHERE user_id=?", (user["id"],)).fetchone()
|
|
if old and old["requested_at"] > now - 300:
|
|
raise ValueError("Please wait five minutes before requesting another confirmation email.")
|
|
conn.execute("""INSERT INTO email_recap_subscriptions
|
|
(user_id, state, email, identity_source, identity_id, version, confirmation_hash,
|
|
confirmation_expires, requested_at, confirmed_at, unsubscribe_token)
|
|
VALUES (?, 'pending', ?, ?, ?, ?, ?, ?, ?, NULL, ?)
|
|
ON CONFLICT(user_id) DO UPDATE SET state='pending', email=excluded.email,
|
|
identity_source=excluded.identity_source, identity_id=excluded.identity_id, version=excluded.version,
|
|
confirmation_hash=excluded.confirmation_hash, confirmation_expires=excluded.confirmation_expires,
|
|
requested_at=excluded.requested_at, confirmed_at=NULL, unsubscribe_token=excluded.unsubscribe_token""",
|
|
(user["id"], user["email"].strip(), source, identity, uuid.uuid4().hex,
|
|
hashlib.sha256(token.encode()).hexdigest(), now + 86400, now, secrets.token_urlsafe(32)))
|
|
conn.execute('UPDATE email_recap_subscriptions SET automatic_monthly=? WHERE user_id=?', (automatic_monthly, user['id']))
|
|
return token
|
|
|
|
|
|
def token_subscription(token: str, action: str) -> dict | None:
|
|
if action == "confirm":
|
|
return read_one("SELECT * FROM email_recap_subscriptions WHERE confirmation_hash=?",
|
|
(hashlib.sha256(token.encode()).hexdigest(),))
|
|
return read_one("SELECT * FROM email_recap_subscriptions WHERE unsubscribe_token=?", (token,))
|
|
|
|
|
|
def confirm(sub: dict, now: float) -> bool:
|
|
with transaction() as conn:
|
|
# Recheck address and blocked state in the same transaction as the consent write.
|
|
result = conn.execute("""UPDATE email_recap_subscriptions SET state='enabled', confirmed_at=?, confirmation_hash=NULL
|
|
WHERE user_id=? AND version=? AND state='pending' AND confirmation_expires>?
|
|
AND EXISTS (SELECT 1 FROM users WHERE users.id=user_id AND is_blocked=0
|
|
AND LOWER(TRIM(users.email))=LOWER(TRIM(email_recap_subscriptions.email)))""",
|
|
(now, sub["user_id"], sub["version"], now))
|
|
return result.rowcount == 1
|
|
|
|
|
|
def _enqueue(conn, sub: dict, month: str, kind: str, key: str, public_url: str, now: float) -> str:
|
|
delivery_id = uuid.uuid4().hex
|
|
conn.execute("""INSERT OR IGNORE INTO email_recap_deliveries
|
|
(id, dedupe_key, user_id, month, kind, email, subscription_version, public_url,
|
|
created_at, updated_at, next_attempt_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(delivery_id, key, sub["user_id"], month, kind, sub["email"], sub["version"], public_url, now, now, now))
|
|
return conn.execute("SELECT id FROM email_recap_deliveries WHERE dedupe_key=?", (key,)).fetchone()[0]
|
|
|
|
|
|
def enqueue_test(sub: dict, month: str, request_id: str, public_url: str, now: float, kind: str = "test") -> str:
|
|
key = f"{kind}:{sub['user_id']}:{request_id}"
|
|
with transaction() as conn:
|
|
existing = conn.execute("SELECT id,month,subscription_version FROM email_recap_deliveries WHERE dedupe_key=?", (key,)).fetchone()
|
|
if existing:
|
|
if existing['month'] != month or existing['subscription_version'] != sub['version']:
|
|
raise ValueError('This send request was already used. Refresh before requesting another report.')
|
|
return existing[0]
|
|
recent = conn.execute("SELECT 1 FROM email_recap_deliveries WHERE user_id=? AND kind IN ('test','on_demand') AND created_at>?",
|
|
(sub["user_id"], now - 300)).fetchone()
|
|
if recent:
|
|
raise ValueError("Please wait five minutes between report emails.")
|
|
return _enqueue(conn, sub, month, kind, key, public_url, now)
|
|
|
|
|
|
def enqueue_due(now: datetime) -> int:
|
|
with transaction() as conn:
|
|
config = dict(conn.execute("SELECT * FROM email_recap_settings WHERE id=1").fetchone())
|
|
if not config["enabled"] or not config["next_send_at"] or config["next_send_at"] > now.timestamp():
|
|
return 0
|
|
# After long downtime, send only the latest due recap; never backfill a pile of old emails.
|
|
due = shift_month(now, 0).replace(day=config["day"], hour=config["hour"])
|
|
if due > now:
|
|
due = shift_month(now, -1).replace(day=config["day"], hour=config["hour"])
|
|
month = shift_month(due, -1).strftime("%Y-%m")
|
|
subs = conn.execute("SELECT * FROM email_recap_subscriptions WHERE state='enabled' AND automatic_monthly=1 AND confirmed_at<=?", (due.timestamp(),)).fetchall()
|
|
before = conn.total_changes
|
|
for sub in subs:
|
|
_enqueue(conn, dict(sub), month, "scheduled", f"scheduled:{sub['user_id']}:{month}", config["public_url"], now.timestamp())
|
|
count = conn.total_changes - before
|
|
conn.execute("UPDATE email_recap_settings SET next_send_at=? WHERE id=1",
|
|
(next_due(now, config["day"], config["hour"]).timestamp(),))
|
|
return count
|
|
|
|
|
|
def claim_delivery(now: float) -> dict | None:
|
|
with transaction() as conn:
|
|
return email_queue.claim(conn, "email_recap_deliveries", now)
|
|
|
|
|
|
def begin_sending(delivery: dict, now: float) -> bool:
|
|
with transaction() as conn:
|
|
# Consent may have changed while the report or SMTP connection was being prepared.
|
|
result = conn.execute("""UPDATE email_recap_deliveries SET state='sending', updated_at=?, lease_until=?
|
|
WHERE id=? AND claim=? AND state='preparing'
|
|
AND EXISTS (SELECT 1 FROM email_recap_subscriptions s JOIN users u ON u.id=s.user_id
|
|
JOIN jellyfin_user_links j ON j.local_user_id=u.id AND j.source=s.identity_source
|
|
WHERE s.user_id=email_recap_deliveries.user_id AND s.state='enabled'
|
|
AND s.version=email_recap_deliveries.subscription_version AND u.is_blocked=0
|
|
AND (email_recap_deliveries.kind!='scheduled' OR s.automatic_monthly=1)
|
|
AND LOWER(TRIM(u.email))=LOWER(TRIM(s.email)) AND j.jellyfin_user_id=s.identity_id)
|
|
AND EXISTS (SELECT 1 FROM email_recap_settings c WHERE c.id=1 AND c.public_url=email_recap_deliveries.public_url
|
|
AND (email_recap_deliveries.kind IN ('test','on_demand') OR c.enabled=1))""", (now, now + 1800, delivery["id"], delivery["claim"]))
|
|
return result.rowcount == 1
|
|
|
|
|
|
def finish(delivery: dict, state: str, detail: str, now: float, delay: int = 0) -> None:
|
|
with transaction() as conn:
|
|
email_queue.finish(conn, "email_recap_deliveries", delivery, state, detail, now, delay)
|
|
|
|
|
|
def history(limit: int = 50, offset: int = 0) -> dict:
|
|
with closing(db._connect()) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
rows = conn.execute("""SELECT d.id, d.month, d.kind, d.email, d.state, d.attempts, d.created_at, d.updated_at,
|
|
d.next_attempt_at, d.detail, u.username FROM email_recap_deliveries d LEFT JOIN users u ON u.id=d.user_id
|
|
ORDER BY d.created_at DESC, d.id LIMIT ? OFFSET ?""", (limit, offset)).fetchall()
|
|
total = conn.execute("SELECT COUNT(*) FROM email_recap_deliveries").fetchone()[0]
|
|
subscribers = conn.execute("SELECT COUNT(*) FROM email_recap_subscriptions WHERE state='enabled'").fetchone()[0]
|
|
return {"deliveries": [dict(row) for row in rows], "total": total, "subscribers": subscribers}
|
|
|
|
|
|
def set_automatic(user_id: int, enabled: bool):
|
|
with transaction() as conn:
|
|
conn.execute('UPDATE email_recap_subscriptions SET automatic_monthly=? WHERE user_id=?', (enabled, user_id))
|
|
if not enabled:
|
|
conn.execute("""UPDATE email_recap_deliveries SET state='cancelled',detail='Automatic monthly emails turned off.'
|
|
WHERE user_id=? AND kind='scheduled' AND state IN ('queued','retry','preparing')""", (user_id,))
|
|
|
|
|
|
def personal_history(user_id: int) -> list[dict]:
|
|
with closing(db._connect()) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
return [dict(row) for row in conn.execute("""SELECT id,month,kind,state,created_at,detail
|
|
FROM email_recap_deliveries WHERE user_id=? ORDER BY created_at DESC,id DESC LIMIT 5""", (user_id,))]
|