from .public_urls import magent_public_url """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") row["public_url"] = magent_public_url(row["public_url"]) 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: values = {**values, "public_url": magent_public_url(values.get("public_url", ""))} 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()) config["public_url"] = magent_public_url(config["public_url"]) 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,))]