diff --git a/backend/app/db.py b/backend/app/db.py
index 0b6a584..d2787f9 100644
--- a/backend/app/db.py
+++ b/backend/app/db.py
@@ -735,6 +735,8 @@ def init_db() -> None:
pass
from .services.recap_store import init_schema as init_recap_schema
init_recap_schema(conn)
+ from .services.newsletter_store import init_schema as init_newsletter_schema
+ init_newsletter_schema(conn)
_backfill_auth_providers()
ensure_admin_user()
_backfill_request_repairs()
diff --git a/backend/app/main.py b/backend/app/main.py
index 022552f..0459ceb 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -30,9 +30,11 @@ from .routers.operations import router as operations_router
from .routers.insights import router as insights_router
from .routers.identities import router as identities_router
from .routers.recaps import router as recaps_router
+from .routers.newsletters import router as newsletters_router
from .services.jellyfin_sync import run_daily_jellyfin_sync
from .services.issue_resolution import run_issue_confirmation_loop
from .services.email_recaps import run_email_recap_loop
+from .services.newsletters import run_newsletter_loop
from .services.operation_progress import (
begin_operation,
finish_operation,
@@ -270,6 +272,7 @@ async def startup() -> None:
_launch_background_task("db-cleanup", run_daily_db_cleanup)
_launch_background_task("issue-confirmation", run_issue_confirmation_loop)
_launch_background_task("email-recaps", run_email_recap_loop)
+ _launch_background_task("newsletters", run_newsletter_loop)
logger.info("startup complete")
@@ -288,3 +291,4 @@ app.include_router(operations_router)
app.include_router(insights_router)
app.include_router(identities_router)
app.include_router(recaps_router)
+app.include_router(newsletters_router)
diff --git a/backend/app/routers/newsletters.py b/backend/app/routers/newsletters.py
new file mode 100644
index 0000000..54b0c7a
--- /dev/null
+++ b/backend/app/routers/newsletters.py
@@ -0,0 +1,191 @@
+import time
+from datetime import datetime, timezone
+from typing import Literal
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from pydantic import Field, field_validator
+
+from ..auth import get_current_user, require_admin
+from ..runtime import get_runtime_settings
+from ..services import newsletters as service, newsletter_store as store, newsletter_catalog as catalog
+from .recaps import StrictPayload, Preference, RecapSettings, TokenAction, no_cache
+
+router = APIRouter(tags=['newsletters'], dependencies=[Depends(no_cache)])
+
+
+class Settings(StrictPayload):
+ enabled: bool
+ weekday: int = Field(ge=0, le=6)
+ hour: int = Field(ge=0, le=23)
+ limit_titles: int = Field(ge=1, le=24)
+ public_url: str = Field(max_length=500)
+ intro: str = Field(default='', max_length=2000)
+ revision: int = Field(ge=1)
+ _url = field_validator('public_url')(RecapSettings.origin_only.__func__)
+
+
+class NewDraft(StrictPayload):
+ days: Literal[7, 14, 30] = 7
+
+
+class Selection(StrictPayload):
+ id: str = Field(pattern=r'^[a-f0-9]{32}$')
+ selected: bool
+ featured: bool
+
+
+class Version(StrictPayload):
+ revision: int = Field(ge=1)
+
+
+class EditionUpdate(Version):
+ subject: str = Field(min_length=1, max_length=150)
+ intro: str = Field(default='', max_length=2000)
+ titles: list[Selection] = Field(max_length=60)
+
+ @field_validator('subject')
+ @classmethod
+ def subject_line(cls, value):
+ value = value.strip()
+ if not value or any(ord(char) < 32 or ord(char) == 127 for char in value):
+ raise ValueError('Use a single, non-empty subject line.')
+ return value
+
+
+class Test(Version):
+ request_id: UUID
+
+
+class Publish(Version):
+ send_at: datetime | None = None
+
+
+def fail(exc):
+ if isinstance(exc, service.NewsletterError):
+ raise HTTPException(exc.status, exc.detail) from exc
+ if isinstance(exc, store.Conflict):
+ raise HTTPException(429 if 'five minutes' in str(exc) else 409, str(exc)) from exc
+ raise HTTPException(502, str(exc) if isinstance(exc, catalog.CatalogError) else 'Jellyfin took too long to prepare this edition. Please try again.') from exc
+
+
+@router.get('/profile/newsletters')
+def preference(user: dict = Depends(get_current_user)):
+ try:
+ return service.preferences(user)
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.put('/profile/newsletters')
+async def set_preference(payload: Preference, user: dict = Depends(get_current_user)):
+ try:
+ if payload.enabled:
+ return await service.subscribe(user)
+ store.disable(service.account_for(user)['id'])
+ return service.preferences(user)
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.post('/newsletter-subscription/check')
+def check_token(payload: TokenAction):
+ try:
+ return service.token_action(payload.token, payload.action)
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.post('/newsletter-subscription/confirm')
+def confirm_token(payload: TokenAction):
+ try:
+ return service.token_action(payload.token, payload.action, apply=True)
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.get('/admin/newsletters')
+def overview(offset: int = Query(default=0, ge=0, le=1_000_000), user: dict = Depends(require_admin)):
+ ready, detail = service.delivery_ready()
+ return {'settings': store.public_settings(), 'ready': ready, 'detail': detail,
+ 'playback_url': service.playback_url(get_runtime_settings()), **store.overview(offset)}
+
+
+@router.put('/admin/newsletters')
+def settings(payload: Settings, user: dict = Depends(require_admin)):
+ try:
+ ready, detail = service.delivery_ready(payload.public_url)
+ if payload.enabled and not ready:
+ raise service.NewsletterError(detail)
+ return store.save_settings(payload.model_dump(), datetime.now(timezone.utc))
+ except (service.NewsletterError, store.Conflict) as exc:
+ fail(exc)
+
+
+@router.post('/admin/newsletters/drafts', status_code=201)
+async def create_draft(payload: NewDraft, user: dict = Depends(require_admin)):
+ try:
+ return await service.create_draft(user, payload.days)
+ except (service.NewsletterError, catalog.CatalogError, TimeoutError) as exc:
+ fail(exc)
+
+
+@router.get('/admin/newsletters/editions/{identity}')
+def edition(identity: UUID, user: dict = Depends(require_admin)):
+ try:
+ return service.require_edition(identity.hex)
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.put('/admin/newsletters/editions/{identity}')
+def update_edition(identity: UUID, payload: EditionUpdate, user: dict = Depends(require_admin)):
+ try:
+ return store.update_edition(identity.hex, payload.revision, payload.subject, payload.intro,
+ [entry.model_dump() for entry in payload.titles], time.time())
+ except store.Conflict as exc:
+ fail(exc)
+
+
+@router.post('/admin/newsletters/editions/{identity}/preview')
+async def preview(identity: UUID, payload: Version, user: dict = Depends(require_admin)):
+ try:
+ return await service.preview(identity.hex, payload.revision)
+ except (service.NewsletterError, catalog.CatalogError, TimeoutError) as exc:
+ fail(exc)
+
+
+@router.post('/admin/newsletters/editions/{identity}/test', status_code=202)
+def send_test(identity: UUID, payload: Test, user: dict = Depends(require_admin)):
+ try:
+ return service.queue_test(user, identity.hex, payload.revision, str(payload.request_id))
+ except (service.NewsletterError, store.Conflict) as exc:
+ fail(exc)
+
+
+@router.post('/admin/newsletters/editions/{identity}/publish', status_code=202)
+def publish(identity: UUID, payload: Publish, user: dict = Depends(require_admin)):
+ try:
+ return service.publish(identity.hex, payload.revision, payload.send_at)
+ except (service.NewsletterError, store.Conflict) as exc:
+ fail(exc)
+
+
+@router.post('/admin/newsletters/editions/{identity}/cancel')
+def cancel(identity: UUID, user: dict = Depends(require_admin)):
+ try:
+ service.require_edition(identity.hex)
+ return store.cancel(identity.hex, time.time())
+ except service.NewsletterError as exc:
+ fail(exc)
+
+
+@router.get('/admin/newsletters/artwork/{identity}')
+async def artwork(identity: UUID, user: dict = Depends(require_admin)):
+ runtime = get_runtime_settings()
+ if not runtime.jellyfin_base_url or not runtime.jellyfin_api_key:
+ raise HTTPException(404, 'Artwork unavailable')
+ content = await catalog.poster(runtime, identity.hex)
+ if not content:
+ raise HTTPException(404, 'Artwork unavailable')
+ return Response(content=content, media_type='image/jpeg', headers={'Cache-Control': 'private, max-age=600'})
diff --git a/backend/app/services/email_queue.py b/backend/app/services/email_queue.py
new file mode 100644
index 0000000..dc27eb3
--- /dev/null
+++ b/backend/app/services/email_queue.py
@@ -0,0 +1,33 @@
+"""Shared claim and completion rules for the two durable email queues."""
+
+import uuid
+
+
+def queue_table(table: str) -> str:
+ if table not in {"email_recap_deliveries", "newsletter_deliveries"}:
+ raise ValueError("Unknown email queue")
+ return table
+
+
+def claim(conn, table: str, now: float) -> dict | None:
+ table = queue_table(table)
+ conn.execute(f"""UPDATE {table} SET state='unknown', detail='Delivery interrupted after sending began; check the mail server.', updated_at=?
+ WHERE state='sending' AND lease_until=3 THEN 'failed' ELSE 'retry' END,
+ next_attempt_at=?, updated_at=?, detail='Email preparation interrupted.'
+ WHERE state='preparing' AND lease_until datetime | None:
+ try:
+ result = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
+ return result.replace(tzinfo=timezone.utc) if result.tzinfo is None else result.astimezone(timezone.utc)
+ except (ValueError, TypeError):
+ return None
+
+
+async def get_json(client, runtime, path, params=None):
+ try:
+ response = await client.get(runtime.jellyfin_base_url.rstrip('/') + path,
+ headers={'X-Emby-Token': runtime.jellyfin_api_key}, params=params)
+ response.raise_for_status()
+ return response.json()
+ except (httpx.HTTPError, ValueError) as exc:
+ raise CatalogError('Jellyfin is temporarily unavailable. Please try again.') from exc
+
+
+def group_arrivals(items: list[dict], start: datetime, end: datetime) -> list[dict]:
+ groups = {}
+ seen = set()
+ for row in items:
+ identity = item_id(row.get('Id'))
+ added = date(row.get('DateCreated'))
+ if (not identity or identity in seen or not added or not start <= added < end
+ or row.get('LocationType') == 'Virtual' or row.get('IsPlaceHolder')):
+ continue
+ kind = row.get('Type')
+ if kind not in {'Movie', 'Episode'}:
+ continue
+ parent = item_id(row.get('SeriesId')) if kind == 'Episode' else identity
+ if not parent:
+ continue
+ seen.add(identity)
+ title = str((row.get('SeriesName') if kind == 'Episode' else row.get('Name')) or '').strip()
+ if not title:
+ continue
+ entry = groups.setdefault(parent, {'id': parent, 'type': 'series' if kind == 'Episode' else 'movie',
+ 'title': title[:250], 'year': row.get('ProductionYear') if kind == 'Movie' else None,
+ 'overview': str(row.get('Overview') or '')[:500] if kind == 'Movie' else '',
+ 'added_at': added.isoformat(), 'has_artwork': False, 'items': [], 'selected': False, 'featured': False})
+ entry['added_at'] = max(entry['added_at'], added.isoformat())
+ entry['has_artwork'] |= bool(row.get('SeriesPrimaryImageTag') if kind == 'Episode' else (row.get('ImageTags') or {}).get('Primary'))
+ entry['items'].append({'id': identity, 'season': row.get('ParentIndexNumber') if kind == 'Episode' else None,
+ 'number': row.get('IndexNumber') if kind == 'Episode' else None})
+ return sorted(groups.values(), key=lambda row: (row['added_at'], row['id']), reverse=True)
+
+
+async def collect(runtime, start: datetime, end: datetime, limit: int = 12) -> dict:
+ if not runtime.jellyfin_base_url or not runtime.jellyfin_api_key:
+ raise CatalogError('Connect Jellyfin before collecting new arrivals.')
+ rows, seen = [], set()
+ exhausted = False
+ async with httpx.AsyncClient(timeout=20) as client:
+ info = await get_json(client, runtime, '/System/Info')
+ server_id = item_id(info.get('Id')) if isinstance(info, dict) else None
+ if not server_id:
+ raise CatalogError('Jellyfin did not return its server identity.')
+ for offset in range(0, MAX_ITEMS, PAGE_SIZE):
+ payload = await get_json(client, runtime, '/Items', {'Recursive': 'true', 'IncludeItemTypes': 'Movie,Episode',
+ 'SortBy': 'DateCreated,SortName', 'SortOrder': 'Descending', 'Fields': 'DateCreated,Overview',
+ 'EnableUserData': 'false', 'IsMissing': 'false', 'IsPlaceHolder': 'false', 'Limit': PAGE_SIZE, 'StartIndex': offset})
+ if not isinstance(payload, dict) or not isinstance(payload.get('Items'), list):
+ raise CatalogError('Jellyfin returned an incomplete arrival list.')
+ page = payload['Items']
+ total = payload.get('TotalRecordCount')
+ if not isinstance(total, int) or total < offset + len(page):
+ raise CatalogError('Jellyfin returned an incomplete arrival count.')
+ for row in page:
+ if not isinstance(row, dict) or not item_id(row.get('Id')) or not date(row.get('DateCreated')):
+ raise CatalogError('Jellyfin returned an arrival without a valid identity or added date.')
+ identity = item_id(row['Id'])
+ if identity in seen:
+ raise CatalogError('The library changed during collection. Refresh arrivals to try again.')
+ seen.add(identity)
+ if rows and date(row['DateCreated']) > date(rows[-1]['DateCreated']):
+ raise CatalogError('The library changed during collection. Refresh arrivals to try again.')
+ rows.append(row)
+ if (not page or len(page) < PAGE_SIZE) and offset + len(page) < total:
+ raise CatalogError('Jellyfin returned an incomplete arrival page.')
+ if not page or any(date(row['DateCreated']) < start for row in page) or offset + len(page) >= total:
+ exhausted = True
+ break
+ if not exhausted:
+ raise CatalogError('More than 5,000 recent items were found. Choose a shorter arrival period; no partial edition was created.')
+ titles = group_arrivals(rows, start, end)
+ total = len(titles)
+ titles = titles[:MAX_TITLES]
+ for index, title in enumerate(titles):
+ title['selected'] = index < limit
+ return {'source': source_key(runtime.jellyfin_base_url), 'server_id': server_id,
+ 'period_start': start.isoformat(), 'period_end': end.isoformat(), 'total_titles': total, 'titles': titles}
+
+
+async def for_recipient(runtime, content: dict, jellyfin_id: str) -> dict:
+ """Scope every ID lookup to a view Jellyfin permits this user to browse.
+
+ Jellyfin 10.11's AddUserToQuery skips its default library filter when ItemIds
+ is present. UserId alone is insufficient; ParentId supplies the allowed scope.
+ """
+ if not item_id(jellyfin_id):
+ raise CatalogError('The recipient does not have a valid Jellyfin identity.')
+ selected = [entry for entry in content['titles'] if entry['selected']]
+ ids = sorted({identity for entry in selected for identity in [entry['id'], *(item['id'] for item in entry['items'])]})
+ allowed = set()
+ async with httpx.AsyncClient(timeout=20) as client:
+ info = await get_json(client, runtime, '/System/Info')
+ if not isinstance(info, dict) or source_key(runtime.jellyfin_base_url) != content['source'] or item_id(info.get('Id')) != content['server_id']:
+ raise CatalogError('The Jellyfin server changed. Create a new edition for the current library.')
+ user = await get_json(client, runtime, '/Users/' + jellyfin_id)
+ if not isinstance(user, dict) or item_id(user.get('Id')) != item_id(jellyfin_id) or not isinstance(user.get('Policy'), dict):
+ raise CatalogError('Could not verify the recipient’s Jellyfin account.')
+ if user['Policy'].get('IsDisabled') or user['Policy'].get('EnableMediaPlayback') is False:
+ return {**content, 'titles': [], 'recipient_disabled': True}
+ views = await get_json(client, runtime, '/UserViews', {'UserId': jellyfin_id, 'IncludeHidden': 'true', 'IncludeExternalContent': 'false'})
+ if not isinstance(views, dict) or not isinstance(views.get('Items'), list) or len(views['Items']) > 32:
+ raise CatalogError('Could not check the recipient’s library access.')
+ for view in views['Items']:
+ parent = item_id(view.get('Id')) if isinstance(view, dict) else None
+ if not parent:
+ raise CatalogError('Jellyfin returned a library without a valid identity.')
+ for offset in range(0, len(ids), 100):
+ chunk = ids[offset:offset + 100]
+ payload = await get_json(client, runtime, '/Items', {'UserId': jellyfin_id, 'ParentId': parent, 'Ids': ','.join(chunk),
+ 'Recursive': 'true', 'Limit': len(chunk), 'EnableUserData': 'false', 'EnableImages': 'false',
+ 'IsMissing': 'false', 'IsPlaceHolder': 'false'})
+ if not isinstance(payload, dict) or not isinstance(payload.get('Items'), list):
+ raise CatalogError('Could not check the recipient’s library access.')
+ allowed.update(item_id(item.get('Id')) for item in payload['Items'] if isinstance(item, dict))
+ titles = []
+ for entry in selected:
+ accessible = [item for item in entry['items'] if item['id'] in allowed]
+ if entry['id'] in allowed and accessible:
+ titles.append({**entry, 'items': accessible})
+ return {**content, 'titles': titles}
+
+
+async def poster(runtime, identity: str) -> bytes | None:
+ if not item_id(identity):
+ return None
+ key = (source_key(runtime.jellyfin_base_url), hashlib.sha256(runtime.jellyfin_api_key.encode()).hexdigest(), identity)
+ async with _poster_lock:
+ cached = _posters.get(key)
+ if cached and cached[0] > time.monotonic():
+ _posters.move_to_end(key)
+ return cached[1]
+ result = None
+ try:
+ async with httpx.AsyncClient(timeout=10) as client:
+ async with client.stream('GET', runtime.jellyfin_base_url.rstrip('/') + f'/Items/{identity}/Images/Primary',
+ headers={'X-Emby-Token': runtime.jellyfin_api_key}, params={'maxWidth': 160, 'maxHeight': 240, 'quality': 82, 'format': 'Jpg'}) as response:
+ response.raise_for_status()
+ data = bytearray()
+ async for chunk in response.aiter_bytes():
+ data.extend(chunk)
+ if len(data) > 512 * 1024:
+ raise ValueError('Poster too large')
+ with Image.open(io.BytesIO(data)) as image:
+ if image.width * image.height > 4_000_000:
+ raise ValueError('Poster dimensions too large')
+ image.thumbnail((160, 240))
+ target = io.BytesIO()
+ image.convert('RGB').save(target, format='JPEG', quality=82)
+ result = target.getvalue()
+ except (httpx.HTTPError, ValueError, OSError, Image.DecompressionBombError):
+ pass
+ _posters[key] = (time.monotonic() + (1800 if result else 60), result)
+ while len(_posters) > 128:
+ _posters.popitem(last=False)
+ return result
+
+
+async def posters(runtime, content: dict) -> dict:
+ titles = [entry for entry in content['titles'] if entry['selected'] and entry['has_artwork']]
+ results = await asyncio.gather(*(poster(runtime, entry['id']) for entry in titles))
+ return {entry['id']: data for entry, data in zip(titles, results) if data}
diff --git a/backend/app/services/newsletter_email.py b/backend/app/services/newsletter_email.py
new file mode 100644
index 0000000..a9191a1
--- /dev/null
+++ b/backend/app/services/newsletter_email.py
@@ -0,0 +1,74 @@
+import base64
+import html
+from urllib.parse import urlencode
+
+from .recap_email import document
+
+
+def description(entry):
+ if entry['type'] == 'movie':
+ return f"Movie · {entry['year']}" if entry.get('year') else 'Movie'
+ seasons = sorted({item['season'] for item in entry['items'] if isinstance(item.get('season'), int)})
+ count = len(entry['items'])
+ labels = ', '.join('Specials' if value == 0 else str(value) for value in seasons[:8])
+ suffix = f" · {'Season' if len(seasons) == 1 else 'Seasons'} {labels}" if labels else ''
+ return f"{count} new {'episode' if count == 1 else 'episodes'}{suffix}"
+
+
+def render_confirmation(username, url):
+ intro = f"Hi {username}, confirm your email to receive new arrivals, featured picks and announcements from Grizzlyflix."
+ return {'subject': 'Confirm your Grizzlyflix newsletter subscription',
+ 'body_text': f'{intro}\n\nConfirm newsletter subscription: {url}\n\nThis link expires in 24 hours. If you did not request this, ignore this email.',
+ 'body_html': document(title='Your next watch starts here.', intro=intro,
+ content='
A weekly look at new movies and TV updates, with posters and links to watch.
',
+ action='Confirm newsletter subscription', url=url, kicker='NEW ON GRIZZLYFLIX',
+ footer='This link expires in 24 hours. If you did not request this, ignore this email.')}
+
+
+def render(content, images, public_url, playback_url, unsubscribe_url, *, preview=False, test=False):
+ esc = html.escape
+ titles = [entry for entry in content['titles'] if entry['selected']]
+ body, lines, attachments = [], [], []
+ intro = str(content.get('intro') or '').strip()
+ if intro:
+ body.append(f'
{esc(intro).replace(chr(10), " ")}
')
+ lines += [intro, '']
+ sections = [('Featured picks', [entry for entry in titles if entry['featured']]),
+ ('New movies', [entry for entry in titles if not entry['featured'] and entry['type'] == 'movie']),
+ ('Fresh episodes', [entry for entry in titles if not entry['featured'] and entry['type'] == 'series'])]
+ for heading, entries in sections:
+ if not entries:
+ continue
+ body.append(f'
{heading}
')
+ lines += [heading, '']
+ for entry in entries:
+ watch = playback_url + '/web/index.html#!/details?' + urlencode({'id': entry['id'], 'serverId': content['server_id']})
+ image_data = images.get(entry['id'])
+ cid = f"newsletter-{entry['id']}@magent"
+ if image_data:
+ source = 'data:image/jpeg;base64,' + base64.b64encode(image_data).decode() if preview else 'cid:' + cid
+ poster = f''
+ if not preview:
+ attachments.append({'cid': cid, 'data': image_data})
+ else:
+ poster = f'
{"TV" if entry["type"] == "series" else "MOVIE"}
'
+ details = description(entry)
+ overview = str(entry.get('overview') or '')[:180]
+ copy = f'
{esc(overview)}
' if overview and entry['featured'] else ''
+ body.append(f'''
''')
+ lines += [entry['title'], details, watch, '']
+ if not titles:
+ body.append('
Your next discovery is waiting in Grizzlyflix.
')
+ period = f"{content['period_start'][:10]} to {content['period_end'][:10]} · UTC"
+ footer = f'You subscribed to the Grizzlyflix newsletter. Arrivals recorded by Jellyfin · {esc(period)} Unsubscribe from newsletters · Email preferences'
+ subject = ('[Test] ' if test else '') + content['subject']
+ return {'subject': subject, 'body_text': '\n'.join([subject, '', *lines, f'Browse Grizzlyflix: {playback_url}', '',
+ f'Arrivals recorded by Jellyfin: {period}', f'Unsubscribe from newsletters: {unsubscribe_url}',
+ f'Email preferences: {public_url}/profile#newsletters']),
+ 'body_html': document(title='What’s new on Grizzlyflix',
+ intro=('This is your test edition. ' if test else '') + 'New stories for your watchlist. Find your next movie or catch up on fresh episodes.',
+ content=''.join(body), action='Explore Grizzlyflix', url=playback_url, footer=footer, kicker='YOUR NEXT WATCH'),
+ 'inline_images': attachments}
diff --git a/backend/app/services/newsletter_store.py b/backend/app/services/newsletter_store.py
new file mode 100644
index 0000000..b5f812a
--- /dev/null
+++ b/backend/app/services/newsletter_store.py
@@ -0,0 +1,347 @@
+"""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}
diff --git a/backend/app/services/newsletters.py b/backend/app/services/newsletters.py
new file mode 100644
index 0000000..6c3a494
--- /dev/null
+++ b/backend/app/services/newsletters.py
@@ -0,0 +1,271 @@
+"""Weekly new-arrival newsletters, manual editions and separate opt-in delivery."""
+
+import asyncio
+import logging
+import time
+import uuid
+from datetime import datetime, timedelta, timezone
+from urllib.parse import urlencode, urlsplit
+
+from .. import db
+from ..runtime import get_runtime_settings
+from . import email_recaps, newsletter_catalog as catalog, newsletter_email as template, newsletter_store as store
+from . import recap_email as mail, recap_store
+from .invite_email import smtp_email_config_ready
+from .jellyfin_identity import linked_user_id, source_key
+
+logger = logging.getLogger(__name__)
+NewsletterError = email_recaps.RecapError
+
+
+def playback_url(runtime) -> str:
+ value = str(runtime.jellyfin_public_url or '').strip().rstrip('/')
+ try:
+ parsed = urlsplit(value)
+ if parsed.scheme in {'https', 'http'} and parsed.hostname and not (parsed.username or parsed.password or parsed.query or parsed.fragment) and not any(c.isspace() or c in '<>"\\' for c in value):
+ return value
+ except ValueError:
+ pass
+ return ''
+
+
+def delivery_ready(public_url=None):
+ config = store.settings()
+ if not (public_url if public_url is not None else config['public_url']):
+ return False, 'Set the public Magent address for newsletter email links.'
+ runtime = get_runtime_settings()
+ if not runtime.jellyfin_base_url or not runtime.jellyfin_api_key:
+ return False, 'Connect Jellyfin to collect new arrivals.'
+ if not playback_url(runtime):
+ return False, 'Set the public Jellyfin address in Jellyfin settings for Watch links.'
+ ready, detail = smtp_email_config_ready()
+ if not ready:
+ return ready, detail
+ if not email_recaps.worker_enabled():
+ return False, 'Background automation is paused on this server.'
+ return True, 'Newsletter delivery is configured.'
+
+
+def account_for(user):
+ account = db.get_user_by_username(user.get('username', ''))
+ if not account or account.get('is_blocked') or account.get('is_expired'):
+ raise NewsletterError('This account cannot receive newsletters.', 403)
+ return account
+
+
+def active_subscription(account):
+ sub = store.subscription(account['id'])
+ if sub and sub['state'] != 'off' and not email_recaps.binding_matches(sub, account):
+ store.disable(account['id'])
+ sub = store.subscription(account['id'])
+ return sub
+
+
+def preferences(user):
+ account = account_for(user)
+ sub = active_subscription(account)
+ runtime = get_runtime_settings()
+ ready, detail = delivery_ready()
+ linked = bool(linked_user_id(account['username'], runtime.jellyfin_base_url))
+ email = mail.valid_email(account.get('email'))
+ config = store.settings()
+ state = sub['state'] if sub else 'off'
+ if state == 'pending' and sub['confirmation_expires'] <= time.time():
+ state = 'expired'
+ return {'state': state, 'email': account.get('email'), 'can_subscribe': ready and linked and bool(email),
+ 'detail': detail if not ready else 'Save a valid profile email address.' if not email else
+ 'Link your Jellyfin account so newsletter titles match your library access.' if not linked else 'New arrivals and featured picks, in your inbox.',
+ 'schedule_enabled': config['enabled'], 'next_send_at': config['next_send_at'], 'weekday': config['weekday'], 'hour': config['hour'],
+ 'resend_after': sub['requested_at'] + 300 if sub else None}
+
+
+async def subscribe(user):
+ account = account_for(user)
+ preference = preferences(user)
+ if preference['state'] == 'enabled':
+ return preference
+ if not preference['can_subscribe']:
+ raise NewsletterError(preference['detail'])
+ runtime = get_runtime_settings()
+ try:
+ token = store.request_confirmation(account, source_key(runtime.jellyfin_base_url),
+ linked_user_id(account['username'], runtime.jellyfin_base_url), time.time())
+ except store.Conflict as exc:
+ raise NewsletterError(str(exc), 429) from exc
+ # The click supplies separate newsletter consent. Reuse a still-valid confirmed address if available.
+ recap = recap_store.subscription(account['id'])
+ if recap and recap['state'] == 'enabled' and email_recaps.binding_matches(recap, account):
+ if store.confirm(store.subscription(account['id']), time.time()):
+ return {**preferences(user), 'message': 'Newsletter subscription is on, using your confirmed profile email.'}
+ config = store.settings()
+ url = config['public_url'] + '/newsletter-subscription#' + urlencode({'action': 'confirm', 'token': token})
+ try:
+ await asyncio.to_thread(mail.send_email, account['email'].strip(), template.render_confirmation(account['username'], url),
+ mail.message_id(uuid.uuid4().hex, config['public_url']))
+ except mail.DeliveryError as exc:
+ raise NewsletterError('Could not confirm delivery of the verification email. Check your inbox; another can be requested in five minutes.', 502) from exc
+ return {**preferences(user), 'message': 'Check your inbox and confirm within 24 hours to turn on newsletters.'}
+
+
+def token_action(token, action, apply=False):
+ sub = store.token_subscription(token, action)
+ if not sub:
+ raise NewsletterError('This newsletter link is invalid or has already been used. Open Profile to manage your subscription.', 410)
+ if action == 'unsubscribe':
+ if apply:
+ store.disable(sub['user_id'])
+ return {'action': action, 'state': 'off' if apply or sub['state'] == 'off' else 'ready'}
+ account = db.get_user_by_id(sub['user_id'])
+ if sub['state'] != 'pending' or sub['confirmation_expires'] <= time.time() or not email_recaps.binding_matches(sub, account):
+ raise NewsletterError('This confirmation expired or your account changed. Request a new newsletter link in Profile.', 410)
+ if apply and not store.confirm(sub, time.time()):
+ raise NewsletterError('This confirmation is no longer available. Request a new newsletter link in Profile.', 410)
+ return {'action': action, 'state': 'enabled' if apply else 'ready'}
+
+
+async def collect(start, end, limit):
+ runtime = get_runtime_settings()
+ result = await asyncio.wait_for(catalog.collect(runtime, start, end, limit), timeout=180)
+ return {**result, 'playback_url': playback_url(runtime)}
+
+
+async def create_draft(user, days):
+ end = datetime.now(timezone.utc)
+ config = store.settings()
+ content = await collect(end - timedelta(days=days), end, config['limit_titles'])
+ return store.create_edition(content, f"What’s new on Grizzlyflix · {end.strftime('%d %b %Y')}", config['intro'], user['username'], end.timestamp())
+
+
+def require_edition(identity, revision=None):
+ row = store.edition(identity)
+ if not row:
+ raise NewsletterError('Newsletter edition not found.', 404)
+ if revision is not None and row['revision'] != revision:
+ raise NewsletterError('This edition changed. Reload it before continuing.')
+ return row
+
+
+async def preview(identity, revision):
+ row = require_edition(identity, revision)
+ runtime = get_runtime_settings()
+ config = store.settings()
+ if not config['public_url'] or not playback_url(runtime):
+ raise NewsletterError('Set the public Magent and Jellyfin addresses before previewing.')
+ if row['content']['source'] != source_key(runtime.jellyfin_base_url) or row['content']['playback_url'] != playback_url(runtime):
+ raise NewsletterError('The Jellyfin connection or public address changed. Create a fresh draft.')
+ content = {**row['content'], 'subject': row['subject'], 'intro': row['intro']}
+ images = await asyncio.wait_for(catalog.posters(runtime, content), timeout=90)
+ rendered = template.render(content, images, config['public_url'], content['playback_url'], config['public_url'] + '/profile#newsletters', preview=True)
+ rendered.pop('inline_images')
+ return {'id': row['id'], 'revision': row['revision'], **rendered}
+
+
+def queue_test(user, identity, revision, request_id):
+ ready, detail = delivery_ready()
+ if not ready:
+ raise NewsletterError(detail)
+ account = account_for(user)
+ sub = active_subscription(account)
+ if not sub or sub['state'] != 'enabled':
+ raise NewsletterError('Subscribe to newsletters and confirm your email in Profile before sending yourself a test.')
+ delivery_id = store.enqueue_test(sub, identity, revision, request_id, store.settings()['public_url'], time.time())
+ return {'id': delivery_id, 'message': 'Test queued for your confirmed newsletter email. Delivery history will show the result.'}
+
+
+def publish(identity, revision, send_at):
+ ready, detail = delivery_ready()
+ if not ready:
+ raise NewsletterError(detail)
+ row = require_edition(identity, revision)
+ runtime = get_runtime_settings()
+ if row['content']['source'] != source_key(runtime.jellyfin_base_url) or row['content']['playback_url'] != playback_url(runtime):
+ raise NewsletterError('The Jellyfin connection changed. Create a fresh draft before sending.')
+ now = datetime.now(timezone.utc)
+ when = now if send_at is None else send_at
+ if when.tzinfo is None:
+ raise NewsletterError('Choose a send time with an explicit timezone.', 422)
+ when = when.astimezone(timezone.utc)
+ if send_at is not None and not now + timedelta(seconds=30) <= when <= now + timedelta(days=90):
+ raise NewsletterError('Schedule the edition at least 30 seconds ahead and within the next 90 days.', 422)
+ return store.publish(identity, revision, when.timestamp(), now.timestamp())
+
+
+def eligible(delivery):
+ account = db.get_user_by_id(delivery['user_id'])
+ sub = active_subscription(account) if account else None
+ ready, _ = delivery_ready()
+ if not ready or not sub or sub['state'] != 'enabled' or sub['version'] != delivery['subscription_version'] or sub['email'] != delivery['email'] or not email_recaps.binding_matches(sub, account) or store.settings()['public_url'] != delivery['public_url']:
+ raise mail.DeliveryCancelled()
+ row = store.edition(delivery['edition_id'])
+ if not row or row['state'] == 'cancelled':
+ raise mail.DeliveryCancelled()
+ return account, sub
+
+
+async def process_delivery(delivery):
+ state, detail, delay = 'failed', 'Could not prepare this newsletter.', 0
+ try:
+ _, sub = eligible(delivery)
+ content = store.version(delivery)
+ runtime = get_runtime_settings()
+ if not content or content['playback_url'] != playback_url(runtime) or content['source'] != source_key(runtime.jellyfin_base_url):
+ raise mail.DeliveryCancelled()
+ content = await asyncio.wait_for(catalog.for_recipient(runtime, content, sub['identity_id']), timeout=120)
+ if content.get('recipient_disabled') or (not content['titles'] and not content['intro'].strip()):
+ state, detail = 'skipped', 'No selected titles are available to this account.'
+ else:
+ images = await asyncio.wait_for(catalog.posters(runtime, content), timeout=90)
+ unsubscribe = delivery['public_url'] + '/newsletter-subscription#' + urlencode({'action': 'unsubscribe', 'token': sub['unsubscribe_token']})
+ rendered = template.render(content, images, delivery['public_url'], content['playback_url'], unsubscribe, test=delivery['kind'] == 'test')
+
+ def before_data():
+ eligible(delivery)
+ if not store.begin_sending(delivery, time.time()):
+ raise mail.DeliveryCancelled()
+
+ await asyncio.to_thread(mail.send_email, delivery['email'], rendered, mail.message_id(delivery['id'], delivery['public_url']), before_data)
+ state, detail = 'sent', 'Accepted by the mail server.'
+ except mail.DeliveryCancelled:
+ state, detail = 'cancelled', 'Subscription, account, edition or email settings changed.'
+ except (catalog.CatalogError, TimeoutError):
+ state, detail = 'retry', 'Jellyfin content or library access could not be checked.'
+ except mail.DeliveryError as exc:
+ state, detail = exc.state, exc.detail
+ except Exception as exc:
+ logger.error('newsletter delivery error id=%s type=%s', delivery['id'], type(exc).__name__)
+ current = store.read_one('SELECT state FROM newsletter_deliveries WHERE id=?', (delivery['id'],))
+ if current and current['state'] == 'sending':
+ state, detail = 'unknown', 'Delivery outcome is unknown; check the mail server.'
+ if state == 'retry':
+ if delivery['attempts'] >= 3:
+ state, detail = 'failed', detail + ' Stopped after three attempts.'
+ else:
+ delay = 300 if delivery['attempts'] == 1 else 1800
+ store.finish(delivery, state, detail, time.time(), delay)
+
+
+async def run_once():
+ if delivery_ready()[0]:
+ config = store.claim_weekly(datetime.now(timezone.utc))
+ if config:
+ try:
+ content = await collect(config['due'] - timedelta(days=7), config['due'], config['limit_titles'])
+ store.complete_weekly(config, content, datetime.now(timezone.utc))
+ except (catalog.CatalogError, TimeoutError):
+ store.complete_weekly(config, None, datetime.now(timezone.utc), 'Could not collect a complete weekly edition from Jellyfin. No newsletter was queued.')
+ store.enqueue_due(time.time())
+ for _ in range(10):
+ delivery = store.claim_delivery(time.time())
+ if not delivery:
+ break
+ await process_delivery(delivery)
+ store.finish_editions(time.time())
+
+
+async def run_newsletter_loop():
+ while True:
+ try:
+ await run_once()
+ except Exception as exc:
+ logger.error('newsletter worker failed type=%s', type(exc).__name__)
+ await asyncio.sleep(30)
diff --git a/backend/app/services/recap_email.py b/backend/app/services/recap_email.py
index d2af859..d94af35 100644
--- a/backend/app/services/recap_email.py
+++ b/backend/app/services/recap_email.py
@@ -40,13 +40,13 @@ def number(value: float) -> str:
return f"{value:,.0f}"
-def document(*, title: str, intro: str, content: str, action: str, url: str, footer: str) -> str:
+def document(*, title: str, intro: str, content: str, action: str, url: str, footer: str, kicker: str = 'YOUR MONTH IN VIEWING') -> str:
esc = html.escape
return f'''{esc(title)}