348 lines
21 KiB
Python
348 lines
21 KiB
Python
"""Independent newsletter consent and immutable edition snapshots using the shared email queue."""
|
||
|
||
import hashlib
|
||
import json
|
||
import secrets
|
||
import uuid
|
||
from contextlib import closing
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from .. import db
|
||
from . import email_queue
|
||
from .recap_store import read_one, transaction
|
||
|
||
|
||
class Conflict(ValueError):
|
||
pass
|
||
|
||
|
||
def init_schema(conn):
|
||
for sql in (
|
||
"""CREATE TABLE IF NOT EXISTS newsletter_settings (
|
||
id INTEGER PRIMARY KEY CHECK(id=1), enabled INTEGER NOT NULL DEFAULT 0,
|
||
weekday INTEGER NOT NULL DEFAULT 4, hour INTEGER NOT NULL DEFAULT 9, limit_titles INTEGER NOT NULL DEFAULT 12,
|
||
public_url TEXT NOT NULL DEFAULT '', intro TEXT NOT NULL DEFAULT '', revision INTEGER NOT NULL DEFAULT 1,
|
||
next_send_at REAL, generation_claim TEXT, generation_until REAL, generation_attempts INTEGER NOT NULL DEFAULT 0,
|
||
last_error TEXT NOT NULL DEFAULT '')""",
|
||
"INSERT OR IGNORE INTO newsletter_settings (id, public_url) SELECT 1, public_url FROM email_recap_settings WHERE id=1",
|
||
"""CREATE TABLE IF NOT EXISTS newsletter_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 newsletter_editions (
|
||
id TEXT PRIMARY KEY, subject TEXT NOT NULL, intro TEXT NOT NULL, content_json TEXT NOT NULL,
|
||
revision INTEGER NOT NULL DEFAULT 1, state TEXT NOT NULL DEFAULT 'draft', origin TEXT NOT NULL DEFAULT 'manual',
|
||
weekly_key TEXT UNIQUE, send_at REAL, created_at REAL NOT NULL, updated_at REAL NOT NULL, created_by TEXT NOT NULL)""",
|
||
"""CREATE TABLE IF NOT EXISTS newsletter_versions (
|
||
edition_id TEXT NOT NULL, revision INTEGER NOT NULL, content_json TEXT NOT NULL,
|
||
PRIMARY KEY (edition_id, revision))""",
|
||
"""CREATE TABLE IF NOT EXISTS newsletter_deliveries (
|
||
id TEXT PRIMARY KEY, dedupe_key TEXT NOT NULL UNIQUE, user_id INTEGER NOT NULL,
|
||
edition_id TEXT NOT NULL, edition_revision INTEGER 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_newsletter_queue ON newsletter_deliveries (state, next_attempt_at)",
|
||
"""CREATE TRIGGER IF NOT EXISTS newsletter_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 newsletter_subscriptions SET state='off', confirmation_hash=NULL, confirmed_at=NULL WHERE user_id=NEW.id; END""",
|
||
"""CREATE TRIGGER IF NOT EXISTS newsletter_account_deleted AFTER DELETE ON users
|
||
BEGIN DELETE FROM newsletter_subscriptions WHERE user_id=OLD.id;
|
||
UPDATE newsletter_deliveries SET state='cancelled', detail='Account removed.'
|
||
WHERE user_id=OLD.id AND state IN ('queued','retry','preparing'); END""",
|
||
"""CREATE TRIGGER IF NOT EXISTS newsletter_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 newsletter_subscriptions SET state='off', confirmation_hash=NULL, confirmed_at=NULL WHERE user_id=OLD.local_user_id; END""",
|
||
"""CREATE TRIGGER IF NOT EXISTS newsletter_identity_deleted AFTER DELETE ON jellyfin_user_links
|
||
BEGIN UPDATE newsletter_subscriptions SET state='off', confirmation_hash=NULL, confirmed_at=NULL WHERE user_id=OLD.local_user_id; END""",
|
||
):
|
||
conn.execute(sql)
|
||
|
||
|
||
def settings() -> dict:
|
||
result = read_one('SELECT * FROM newsletter_settings WHERE id=1')
|
||
result['enabled'] = bool(result['enabled'])
|
||
return result
|
||
|
||
|
||
def public_settings() -> dict:
|
||
return {key: value for key, value in settings().items() if key in
|
||
{'enabled', 'weekday', 'hour', 'limit_titles', 'public_url', 'intro', 'revision', 'next_send_at', 'last_error'}}
|
||
|
||
|
||
def next_due(now: datetime, weekday: int, hour: int) -> datetime:
|
||
now = now.astimezone(timezone.utc)
|
||
due = now.replace(hour=hour, minute=0, second=0, microsecond=0) + timedelta(days=(weekday - now.weekday()) % 7)
|
||
return due if due > now else due + timedelta(days=7)
|
||
|
||
|
||
def save_settings(values: dict, now: datetime):
|
||
with transaction() as conn:
|
||
old = dict(conn.execute('SELECT * FROM newsletter_settings WHERE id=1').fetchone())
|
||
if old['revision'] != values['revision']:
|
||
raise Conflict('The newsletter settings changed. Refresh before saving.')
|
||
due = next_due(now, values['weekday'], values['hour']).timestamp() if values['enabled'] else None
|
||
conn.execute("""UPDATE newsletter_settings SET enabled=?, weekday=?, hour=?, limit_titles=?, public_url=?, intro=?,
|
||
revision=revision+1, next_send_at=?, generation_claim=NULL, generation_until=NULL, generation_attempts=0, last_error='' WHERE id=1""",
|
||
(values['enabled'], values['weekday'], values['hour'], values['limit_titles'], values['public_url'], values['intro'], due))
|
||
if not values['enabled'] or any(old[key] != values[key] for key in ('weekday', 'hour', 'public_url')):
|
||
conn.execute("UPDATE newsletter_editions SET state='cancelled', updated_at=? WHERE origin='weekly' AND state IN ('scheduled','queued')", (now.timestamp(),))
|
||
conn.execute("""UPDATE newsletter_deliveries SET state='cancelled', detail='Weekly schedule paused or changed.'
|
||
WHERE state IN ('queued','retry','preparing') AND kind='edition'
|
||
AND edition_id IN (SELECT id FROM newsletter_editions WHERE state='cancelled')""")
|
||
return public_settings()
|
||
|
||
|
||
def subscription(user_id):
|
||
return read_one('SELECT * FROM newsletter_subscriptions WHERE user_id=?', (user_id,))
|
||
|
||
|
||
def disable(user_id):
|
||
with transaction() as conn:
|
||
conn.execute("UPDATE newsletter_subscriptions SET state='off', confirmation_hash=NULL, confirmed_at=NULL WHERE user_id=?", (user_id,))
|
||
conn.execute("UPDATE newsletter_deliveries SET state='cancelled', detail='Newsletter subscription turned off.' WHERE user_id=? AND state IN ('queued','retry','preparing')", (user_id,))
|
||
|
||
|
||
def request_confirmation(user, source, identity, now):
|
||
token = secrets.token_urlsafe(32)
|
||
with transaction() as conn:
|
||
old = conn.execute('SELECT requested_at FROM newsletter_subscriptions WHERE user_id=?', (user['id'],)).fetchone()
|
||
if old and old[0] > now - 300:
|
||
raise Conflict('Please wait five minutes before requesting another confirmation.')
|
||
conn.execute("""INSERT INTO newsletter_subscriptions (user_id,state,email,identity_source,identity_id,version,
|
||
confirmation_hash,confirmation_expires,requested_at,unsubscribe_token) VALUES (?,'pending',?,?,?,?,?,?,?,?)
|
||
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)))
|
||
return token
|
||
|
||
|
||
def token_subscription(token, action):
|
||
if action == 'confirm':
|
||
return read_one('SELECT * FROM newsletter_subscriptions WHERE confirmation_hash=?', (hashlib.sha256(token.encode()).hexdigest(),))
|
||
return read_one('SELECT * FROM newsletter_subscriptions WHERE unsubscribe_token=?', (token,))
|
||
|
||
|
||
def confirm(sub, now):
|
||
with transaction() as conn:
|
||
result = conn.execute("""UPDATE newsletter_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 u JOIN jellyfin_user_links j ON j.local_user_id=u.id
|
||
WHERE u.id=newsletter_subscriptions.user_id AND u.is_blocked=0
|
||
AND LOWER(TRIM(u.email))=LOWER(TRIM(newsletter_subscriptions.email))
|
||
AND j.source=identity_source AND j.jellyfin_user_id=identity_id)""", (now, sub['user_id'], sub['version'], now))
|
||
return result.rowcount == 1
|
||
|
||
|
||
def unpack(row):
|
||
if row is None:
|
||
return None
|
||
result = dict(row)
|
||
result['content'] = json.loads(result.pop('content_json'))
|
||
return result
|
||
|
||
|
||
def edition(identity):
|
||
return unpack(read_one('SELECT * FROM newsletter_editions WHERE id=?', (identity,)))
|
||
|
||
|
||
def create_edition(content, subject, intro, creator, now):
|
||
identity = uuid.uuid4().hex
|
||
with transaction() as conn:
|
||
conn.execute('''INSERT INTO newsletter_editions (id,subject,intro,content_json,created_at,updated_at,created_by)
|
||
VALUES (?,?,?,?,?,?,?)''', (identity, subject, intro, json.dumps(content), now, now, creator))
|
||
return edition(identity)
|
||
|
||
|
||
def editable(conn, identity, revision):
|
||
row = conn.execute('SELECT * FROM newsletter_editions WHERE id=?', (identity,)).fetchone()
|
||
if not row or row['revision'] != revision:
|
||
raise Conflict('This edition changed. Reload it before continuing.')
|
||
if row['state'] != 'draft':
|
||
raise Conflict('This edition is already scheduled or finished. Create a new draft to make changes.')
|
||
return unpack(row)
|
||
|
||
|
||
def update_edition(identity, revision, subject, intro, selections, now):
|
||
with transaction() as conn:
|
||
old = editable(conn, identity, revision)
|
||
titles = old['content']['titles']
|
||
selected = {entry['id']: entry for entry in selections}
|
||
if len(selected) != len(selections) or set(selected) != {entry['id'] for entry in titles}:
|
||
raise Conflict('The title selection does not match this draft. Reload the edition.')
|
||
if sum(bool(entry['selected']) for entry in selections) > 24 or sum(bool(entry['featured']) for entry in selections) > 3:
|
||
raise Conflict('Choose up to 24 titles and three featured picks.')
|
||
if any(entry['featured'] and not entry['selected'] for entry in selections):
|
||
raise Conflict('Featured picks must be included in the edition.')
|
||
for entry in titles:
|
||
entry.update(selected=selected[entry['id']]['selected'], featured=selected[entry['id']]['featured'])
|
||
conn.execute('UPDATE newsletter_editions SET subject=?,intro=?,content_json=?,revision=revision+1,updated_at=? WHERE id=?',
|
||
(subject, intro, json.dumps(old['content']), now, identity))
|
||
return edition(identity)
|
||
|
||
|
||
def snapshot(conn, row):
|
||
data = {**row['content'], 'subject': row['subject'], 'intro': row['intro']}
|
||
# Store only included titles; retries of a test retain the exact saved version.
|
||
data['titles'] = [entry for entry in data['titles'] if entry['selected']]
|
||
conn.execute('INSERT OR IGNORE INTO newsletter_versions (edition_id,revision,content_json) VALUES (?,?,?)',
|
||
(row['id'], row['revision'], json.dumps(data)))
|
||
|
||
|
||
def version(delivery):
|
||
row = read_one('SELECT content_json FROM newsletter_versions WHERE edition_id=? AND revision=?', (delivery['edition_id'], delivery['edition_revision']))
|
||
return json.loads(row['content_json']) if row else None
|
||
|
||
|
||
def publish(identity, revision, send_at, now):
|
||
with transaction() as conn:
|
||
previous = conn.execute('SELECT revision,state FROM newsletter_editions WHERE id=?', (identity,)).fetchone()
|
||
if previous and previous['revision'] == revision and previous['state'] in {'scheduled', 'queued', 'complete'}:
|
||
return edition(identity)
|
||
row = editable(conn, identity, revision)
|
||
if not any(entry['selected'] for entry in row['content']['titles']) and not row['intro'].strip():
|
||
raise Conflict('Add an announcement or select a title before sending.')
|
||
snapshot(conn, row)
|
||
conn.execute("UPDATE newsletter_editions SET state='scheduled',send_at=?,updated_at=? WHERE id=?", (send_at, now, identity))
|
||
return edition(identity)
|
||
|
||
|
||
def cancel(identity, now):
|
||
with transaction() as conn:
|
||
conn.execute("UPDATE newsletter_editions SET state='cancelled',updated_at=? WHERE id=? AND state IN ('draft','scheduled','queued')", (now, identity))
|
||
conn.execute("UPDATE newsletter_deliveries SET state='cancelled',detail='Edition cancelled.',updated_at=? WHERE edition_id=? AND state IN ('queued','retry','preparing')", (now, identity))
|
||
return edition(identity)
|
||
|
||
|
||
def _enqueue(conn, sub, row, kind, key, public_url, now):
|
||
identity = uuid.uuid4().hex
|
||
conn.execute('''INSERT OR IGNORE INTO newsletter_deliveries (id,dedupe_key,user_id,edition_id,edition_revision,kind,email,
|
||
subscription_version,public_url,created_at,updated_at,next_attempt_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)''',
|
||
(identity, key, sub['user_id'], row['id'], row['revision'], kind, sub['email'], sub['version'], public_url, now, now, now))
|
||
return conn.execute('SELECT id FROM newsletter_deliveries WHERE dedupe_key=?', (key,)).fetchone()[0]
|
||
|
||
|
||
def enqueue_test(sub, identity, revision, request_id, public_url, now):
|
||
key = f"test:{sub['user_id']}:{request_id}"
|
||
with transaction() as conn:
|
||
previous = conn.execute('SELECT id,edition_id,edition_revision FROM newsletter_deliveries WHERE dedupe_key=?', (key,)).fetchone()
|
||
if previous:
|
||
if previous['edition_id'] != identity or previous['edition_revision'] != revision:
|
||
raise Conflict('This test request was already used for another saved version.')
|
||
return previous['id']
|
||
row = unpack(conn.execute('SELECT * FROM newsletter_editions WHERE id=? AND revision=?', (identity, revision)).fetchone())
|
||
if not row or row['state'] == 'cancelled':
|
||
raise Conflict('This edition changed or was cancelled. Reload it first.')
|
||
if conn.execute("SELECT 1 FROM newsletter_deliveries WHERE user_id=? AND kind='test' AND created_at>?", (sub['user_id'], now-300)).fetchone():
|
||
raise Conflict('Please wait five minutes between newsletter test emails.')
|
||
snapshot(conn, row)
|
||
return _enqueue(conn, sub, row, 'test', key, public_url, now)
|
||
|
||
|
||
def enqueue_due(now):
|
||
with transaction() as conn:
|
||
config = conn.execute('SELECT * FROM newsletter_settings WHERE id=1').fetchone()
|
||
rows = conn.execute("SELECT * FROM newsletter_editions WHERE state='scheduled' AND send_at<=?", (now,)).fetchall()
|
||
for raw in rows:
|
||
row = unpack(raw)
|
||
subs = conn.execute("SELECT * FROM newsletter_subscriptions WHERE state='enabled' AND confirmed_at<=?", (row['send_at'],)).fetchall()
|
||
for sub in subs:
|
||
_enqueue(conn, sub, row, 'edition', f"edition:{row['id']}:{sub['user_id']}", config['public_url'], now)
|
||
conn.execute("UPDATE newsletter_editions SET state=?,updated_at=? WHERE id=?", ('queued' if subs else 'complete', now, row['id']))
|
||
|
||
|
||
def claim_delivery(now):
|
||
with transaction() as conn:
|
||
return email_queue.claim(conn, 'newsletter_deliveries', now)
|
||
|
||
|
||
def begin_sending(delivery, now):
|
||
with transaction() as conn:
|
||
result = conn.execute("""UPDATE newsletter_deliveries SET state='sending',updated_at=?,lease_until=?
|
||
WHERE id=? AND claim=? AND state='preparing'
|
||
AND EXISTS (SELECT 1 FROM newsletter_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=newsletter_deliveries.user_id AND s.state='enabled'
|
||
AND s.version=newsletter_deliveries.subscription_version AND u.is_blocked=0
|
||
AND LOWER(TRIM(u.email))=LOWER(TRIM(s.email)) AND j.jellyfin_user_id=s.identity_id)
|
||
AND EXISTS (SELECT 1 FROM newsletter_settings WHERE id=1 AND public_url=newsletter_deliveries.public_url)
|
||
AND EXISTS (SELECT 1 FROM newsletter_editions e WHERE e.id=newsletter_deliveries.edition_id AND e.state!='cancelled')""",
|
||
(now, now+1800, delivery['id'], delivery['claim']))
|
||
return result.rowcount == 1
|
||
|
||
|
||
def finish(delivery, state, detail, now, delay=0):
|
||
with transaction() as conn:
|
||
email_queue.finish(conn, 'newsletter_deliveries', delivery, state, detail, now, delay)
|
||
|
||
|
||
def finish_editions(now):
|
||
with transaction() as conn:
|
||
conn.execute("""UPDATE newsletter_editions SET state='complete',updated_at=? WHERE state='queued'
|
||
AND NOT EXISTS (SELECT 1 FROM newsletter_deliveries d WHERE d.edition_id=newsletter_editions.id
|
||
AND d.kind='edition' AND d.state IN ('queued','preparing','sending','retry'))""", (now,))
|
||
|
||
|
||
def claim_weekly(now: datetime):
|
||
with transaction() as conn:
|
||
config = dict(conn.execute('SELECT * FROM newsletter_settings WHERE id=1').fetchone())
|
||
stamp = now.timestamp()
|
||
if not config['enabled'] or not config['next_send_at'] or config['next_send_at'] > stamp or (config['generation_until'] or 0) > stamp:
|
||
return None
|
||
claim = uuid.uuid4().hex
|
||
conn.execute('UPDATE newsletter_settings SET generation_claim=?,generation_until=?,generation_attempts=generation_attempts+1 WHERE id=1', (claim, stamp+600))
|
||
due = next_due(now, config['weekday'], config['hour']) - timedelta(days=7)
|
||
return {**config, 'generation_claim': claim, 'due': due, 'generation_attempts': config['generation_attempts']+1}
|
||
|
||
|
||
def complete_weekly(config, content, now: datetime, failure=''):
|
||
with transaction() as conn:
|
||
current = conn.execute('SELECT * FROM newsletter_settings WHERE id=1').fetchone()
|
||
if not current['enabled'] or current['revision'] != config['revision'] or current['generation_claim'] != config['generation_claim']:
|
||
return
|
||
if failure:
|
||
retry = config['generation_attempts'] < 3
|
||
conn.execute('''UPDATE newsletter_settings SET generation_claim=NULL,generation_until=?,last_error=?,next_send_at=?,
|
||
generation_attempts=? WHERE id=1''', (now.timestamp()+300 if retry else None, failure,
|
||
current['next_send_at'] if retry else next_due(now, config['weekday'], config['hour']).timestamp(),
|
||
config['generation_attempts'] if retry else 0))
|
||
return
|
||
identity = uuid.uuid4().hex
|
||
due = config['due']
|
||
empty = not content['titles']
|
||
conn.execute('''INSERT OR IGNORE INTO newsletter_editions
|
||
(id,subject,intro,content_json,state,origin,weekly_key,send_at,created_at,updated_at,created_by)
|
||
VALUES (?,?,?,?,?,'weekly',?,?,?,?,?)''',
|
||
(identity, f"What’s new on Grizzlyflix · {due.strftime('%d %b %Y')}", config['intro'], json.dumps(content),
|
||
'skipped' if empty else 'scheduled', due.isoformat(), due.timestamp(), now.timestamp(), now.timestamp(), 'Weekly schedule'))
|
||
row = unpack(conn.execute('SELECT * FROM newsletter_editions WHERE weekly_key=?', (due.isoformat(),)).fetchone())
|
||
if not empty:
|
||
snapshot(conn, row)
|
||
conn.execute('''UPDATE newsletter_settings SET next_send_at=?,generation_claim=NULL,generation_until=NULL,
|
||
generation_attempts=0,last_error=? WHERE id=1''',
|
||
(next_due(now, config['weekday'], config['hour']).timestamp(), 'No new arrivals for the weekly edition; no email was queued.' if empty else ''))
|
||
|
||
|
||
def overview(offset=0):
|
||
with closing(db._connect()) as conn:
|
||
import sqlite3
|
||
conn.row_factory = sqlite3.Row
|
||
rows = conn.execute('SELECT * FROM newsletter_editions ORDER BY created_at DESC,id LIMIT 30').fetchall()
|
||
editions = []
|
||
for raw in rows:
|
||
row = unpack(raw)
|
||
content = row.pop('content')
|
||
row.update(period_start=content['period_start'], period_end=content['period_end'], titles=sum(entry['selected'] for entry in content['titles']))
|
||
editions.append(row)
|
||
deliveries = conn.execute('''SELECT d.id,d.edition_id,e.subject,d.kind,d.email,d.state,d.attempts,d.updated_at,d.next_attempt_at,
|
||
d.detail,u.username FROM newsletter_deliveries d LEFT JOIN users u ON u.id=d.user_id
|
||
LEFT JOIN newsletter_editions e ON e.id=d.edition_id ORDER BY d.created_at DESC,d.id LIMIT 50 OFFSET ?''', (offset,)).fetchall()
|
||
subscribers = conn.execute("SELECT COUNT(*) FROM newsletter_subscriptions WHERE state='enabled'").fetchone()[0]
|
||
total = conn.execute('SELECT COUNT(*) FROM newsletter_deliveries').fetchone()[0]
|
||
return {'editions': editions, 'deliveries': [dict(row) for row in deliveries], 'subscribers': subscribers, 'total': total}
|