Add Grizzlyflix newsletters with curated editions and weekly delivery

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()

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)

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'})

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<?""", (now, now))
+    conn.execute(f"""UPDATE {table} SET state=CASE WHEN attempts>=3 THEN 'failed' ELSE 'retry' END,
+        next_attempt_at=?, updated_at=?, detail='Email preparation interrupted.'
+        WHERE state='preparing' AND lease_until<?""", (now, now, now))
+    row = conn.execute(f"""SELECT * FROM {table} WHERE state IN ('queued', 'retry') AND next_attempt_at<=?
+        ORDER BY created_at, id LIMIT 1""", (now,)).fetchone()
+    if not row:
+        return None
+    claim_id = uuid.uuid4().hex
+    conn.execute(f"""UPDATE {table} SET state='preparing', claim=?, lease_until=?,
+        attempts=attempts+1, updated_at=? WHERE id=?""", (claim_id, now + 1800, now, row["id"]))
+    return dict(conn.execute(f"SELECT * FROM {table} WHERE id=?", (row["id"],)).fetchone())
+
+
+def finish(conn, table: str, delivery: dict, state: str, detail: str, now: float, delay: int = 0):
+    table = queue_table(table)
+    conn.execute(f"""UPDATE {table} SET state=?, detail=?, updated_at=?, next_attempt_at=?, lease_until=NULL
+        WHERE id=? AND claim=? AND state IN ('preparing', 'sending')""",
+        (state, detail, now, now + delay, delivery["id"], delivery["claim"]))

backend/app/services/newsletter_catalog.py

@@ -0,0 +1,202 @@
+"""Bounded Jellyfin arrival snapshots, recipient access checks and email-safe posters."""
+
+import asyncio
+import hashlib
+import io
+import time
+from collections import OrderedDict
+from datetime import datetime, timezone
+
+import httpx
+from PIL import Image
+
+from .insights_artwork import item_id
+from .jellyfin_identity import source_key
+
+MAX_ITEMS = 5000
+PAGE_SIZE = 200
+MAX_TITLES = 60
+_posters = OrderedDict()
+_poster_lock = asyncio.Semaphore(4)
+
+
+class CatalogError(Exception):
+    pass
+
+
+def date(value) -> 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}

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='<p style="color:#bdb6c3;font-size:14px;line-height:1.7">A weekly look at new movies and TV updates, with posters and links to watch.</p>',
+            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'<p style="font-size:15px;line-height:1.8;color:#e5e1e4;overflow-wrap:anywhere">{esc(intro).replace(chr(10), "<br>")}</p>')
+        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'<h2 style="font-size:20px;margin:28px 0 8px;color:#e5e1e4">{heading}</h2>')
+        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'<img src="{source}" width="80" alt="{esc(entry["title"], quote=True)}" style="display:block;width:80px;height:auto;border-radius:7px;border:0">'
+                if not preview:
+                    attachments.append({'cid': cid, 'data': image_data})
+            else:
+                poster = f'<div style="width:80px;height:112px;line-height:112px;background:#353039;color:#c7bdff;text-align:center;border-radius:7px;font-size:11px">{"TV" if entry["type"] == "series" else "MOVIE"}</div>'
+            details = description(entry)
+            overview = str(entry.get('overview') or '')[:180]
+            copy = f'<p style="margin:8px 0;font-size:12px;line-height:1.6;color:#bdb6c3">{esc(overview)}</p>' if overview and entry['featured'] else ''
+            body.append(f'''<table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="table-layout:fixed;border-bottom:1px solid #363338"><tr>
+<td width="92" valign="top" style="padding:18px 12px 18px 0">{poster}</td><td valign="top" style="padding:18px 0;overflow-wrap:anywhere">
+<h3 style="margin:0 0 8px;font-size:16px;line-height:1.4;color:#eee8f2">{esc(entry['title'])}</h3><p style="font-size:12px;line-height:1.6;color:#a69fac;margin:0 0 10px">{esc(details)}</p>{copy}
+<a href="{esc(watch, quote=True)}" style="display:inline-block;padding:8px 0;color:#c7bdff;text-decoration:none;font-size:13px;font-weight:bold">Watch on Grizzlyflix &#8599;</a></td></tr></table>''')
+            lines += [entry['title'], details, watch, '']
+    if not titles:
+        body.append('<p style="font-size:14px;line-height:1.7;color:#bdb6c3">Your next discovery is waiting in Grizzlyflix.</p>')
+    period = f"{content['period_start'][:10]} to {content['period_end'][:10]} · UTC"
+    footer = f'You subscribed to the Grizzlyflix newsletter.<br>Arrivals recorded by Jellyfin · {esc(period)}<br><a href="{esc(unsubscribe_url, quote=True)}" style="color:#c7bdff">Unsubscribe from newsletters</a> · <a href="{esc(public_url + "/profile#newsletters", quote=True)}" style="color:#c7bdff">Email preferences</a>'
+    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}

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}

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)

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'''<!doctype html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1"><meta name="color-scheme" content="dark"><title>{esc(title)}</title><style>@media(max-width:280px){{.email-metrics td{{display:block!important;width:auto!important;padding:16px 0!important}}.email-metrics tr{{display:block!important}}}}</style></head>
 <body style="margin:0;padding:0;background:#131315;color:#e5e1e4;font-family:Arial,Helvetica,sans-serif">
 <table role="presentation" width="100%" cellpadding="0" cellspacing="0" style="background:#131315"><tr><td align="center" style="padding:24px 12px">
 <table role="presentation" width="600" cellpadding="0" cellspacing="0" style="width:100%;max-width:600px;table-layout:fixed;background:#1c1b1d;border:1px solid #363338;border-radius:16px">
-<tr><td style="padding:32px 24px 8px;color:#c7bdff;font-size:12px;letter-spacing:2px;font-weight:bold">MAGENT <span style="color:#918b98;letter-spacing:0">/ YOUR MONTH IN VIEWING</span></td></tr>
+<tr><td style="padding:32px 24px 8px;color:#c7bdff;font-size:12px;letter-spacing:2px;font-weight:bold">MAGENT <span style="color:#918b98;letter-spacing:0">/ {esc(kicker)}</span></td></tr>
 <tr><td style="padding:12px 24px"><h1 style="margin:0 0 16px;font-size:32px;line-height:1.2;color:#f3eef6">{esc(title)}</h1><p style="margin:0;color:#bdb6c3;font-size:15px;line-height:1.7;overflow-wrap:anywhere">{esc(intro)}</p></td></tr>
 <tr><td style="padding:12px 24px">{content}</td></tr>
 <tr><td style="padding:20px 24px 32px"><a href="{esc(url, quote=True)}" style="display:inline-block;padding:15px 22px;border-radius:8px;background:#c7bdff;color:#211b30;text-decoration:none;font-size:14px;font-weight:bold">{esc(action)} &#8599;</a></td></tr>
@@ -122,6 +122,11 @@ def send_email(recipient: str, rendered: dict, message_id: str, before_data=lamb
     message["Auto-Submitted"], message["X-Auto-Response-Suppress"] = "auto-generated", "All"
     message.set_content(rendered["body_text"])
     message.add_alternative(rendered["body_html"], subtype="html")
+    html_part = message.get_payload()[-1]
+    for attachment in rendered.get('inline_images', []):
+        html_part.add_related(
+            attachment['data'], maintype='image', subtype='jpeg', cid=f"<{attachment['cid']}>",
+            filename=attachment['cid'].split('@')[0] + '.jpg', disposition='inline')
     payload = message.as_bytes()
     smtp, stage = None, "connect"
     try:

backend/app/services/recap_store.py

@@ -9,6 +9,7 @@ from datetime import datetime
 
 from .. import db
 from .monthly_reports import shift_month
+from . import email_queue
 
 
 def init_schema(conn: sqlite3.Connection) -> None:
@@ -186,20 +187,7 @@ def enqueue_due(now: datetime) -> int:
 
 def claim_delivery(now: float) -> dict | None:
     with transaction() as conn:
-        # A crashed worker could already have handed DATA to SMTP. Do not resend it automatically.
-        conn.execute("""UPDATE email_recap_deliveries SET state='unknown', detail='Delivery interrupted after sending began; check the mail server.', updated_at=?
-            WHERE state='sending' AND lease_until<?""", (now, now))
-        conn.execute("""UPDATE email_recap_deliveries SET state=CASE WHEN attempts>=3 THEN 'failed' ELSE 'retry' END,
-            next_attempt_at=?, updated_at=?, detail='Report preparation interrupted.'
-            WHERE state='preparing' AND lease_until<?""", (now, now, now))
-        row = conn.execute("""SELECT * FROM email_recap_deliveries WHERE state IN ('queued', 'retry') AND next_attempt_at<=?
-            ORDER BY created_at, id LIMIT 1""", (now,)).fetchone()
-        if not row:
-            return None
-        claim = uuid.uuid4().hex
-        conn.execute("""UPDATE email_recap_deliveries SET state='preparing', claim=?, lease_until=?,
-            attempts=attempts+1, updated_at=? WHERE id=?""", (claim, now + 1800, now, row["id"]))
-        return dict(conn.execute("SELECT * FROM email_recap_deliveries WHERE id=?", (row["id"],)).fetchone())
+        return email_queue.claim(conn, "email_recap_deliveries", now)
 
 
 def begin_sending(delivery: dict, now: float) -> bool:
@@ -219,9 +207,7 @@ def begin_sending(delivery: dict, now: float) -> bool:
 
 def finish(delivery: dict, state: str, detail: str, now: float, delay: int = 0) -> None:
     with transaction() as conn:
-        conn.execute("""UPDATE email_recap_deliveries SET state=?, detail=?, updated_at=?, next_attempt_at=?, lease_until=NULL
-            WHERE id=? AND claim=? AND state IN ('preparing', 'sending')""",
-            (state, detail, now, now + delay, delivery["id"], delivery["claim"]))
+        email_queue.finish(conn, "email_recap_deliveries", delivery, state, detail, now, delay)
 
 
 def history(limit: int = 50, offset: int = 0) -> dict: