docs: update project description to reflect current scope

- add federate_reviews + federate_watchlist to UserSettings (default true)
- new UserFederationSettingsQuery port with FederationFlags struct
- remove get_user_federate_goals from LocalApContentQuery
- gate ReviewLogged, ReviewUpdated, WatchlistEntryAdded, on_poster_synced on flags
- goals gating migrated to UserFederationSettingsQuery
- ReviewDeleted and WatchlistEntryRemoved ungated (tombstones always fire)
- sqlite + postgres migrations and adapter impls
- settings API and SPA toggles

Cargo.lock

@@ -314,6 +314,7 @@ name = "application"
 version = "0.1.0"
 dependencies = [
  "async-trait",
+ "bytes",
  "chrono",
  "domain",
  "futures",
@@ -567,6 +568,28 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "async-stream"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
+
 [[package]]
 name = "async-task"
 version = "4.7.1"
@@ -1822,9 +1845,12 @@ dependencies = [
 name = "export"
 version = "0.1.0"
 dependencies = [
+ "async-stream",
  "async-trait",
+ "bytes",
  "chrono",
  "domain",
+ "futures",
  "serde_json",
  "tokio",
  "uuid",
@@ -3848,9 +3874,12 @@ name = "postgres"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-stream",
  "async-trait",
+ "bytes",
  "chrono",
  "domain",
+ "futures",
  "serde",
  "serde_json",
  "sqlx",
@@ -5124,9 +5153,12 @@ name = "sqlite"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "async-stream",
  "async-trait",
+ "bytes",
  "chrono",
  "domain",
+ "futures",
  "serde",
  "serde_json",
  "sqlx",

Cargo.toml

@@ -38,6 +38,7 @@ resolver = "2"
 tokio = { version = "1.0", features = ["macros", "net", "rt", "rt-multi-thread", "sync", "time"] }
 bytes = "1"
 futures = "0.3"
+async-stream = "0.3"
 dotenvy = "0.15"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

README.md

@@ -1,6 +1,49 @@
 # Movies Diary
 
-A self-hosted, server-side rendered movie logging system with a full REST API. Built in Rust — no JavaScript in the HTML interface, just HTML forms and an RSS feed. Designed to run as a lightweight widget embedded on a personal site or as a backend for third-party clients.
+A self-hosted movie diary built in Rust. Ships a classic server-rendered HTML interface (no JavaScript) alongside a full React SPA, both backed by the same REST API. Federates over ActivityPub so reviews reach the Fediverse. Supports Jellyfin and Plex auto-import, full-text search, annual wrap-ups, goals, and bulk import from Letterboxd, IMDb, and other sources. Runs on SQLite or PostgreSQL.
+
+[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
+[![Built with Rust](https://img.shields.io/badge/built_with-Rust-orange.svg?logo=rust)](https://www.rust-lang.org/)
+[![Docker](https://img.shields.io/badge/docker-ready-2496ED?logo=docker&logoColor=white)](https://hub.docker.com/)
+[![ActivityPub](https://img.shields.io/badge/ActivityPub-federated-5b5ea6)](https://activitypub.rocks/)
+[![SQLite](https://img.shields.io/badge/database-SQLite%20%7C%20PostgreSQL-003B57?logo=sqlite&logoColor=white)](https://www.sqlite.org/)
+
+---
+
+## Table of Contents
+
+- [Quick Start](#quick-start)
+- [Features](#features)
+- [Screenshots](#screenshots)
+- [Architecture](#architecture)
+- [Prerequisites](#prerequisites)
+- [Configuration](#configuration)
+- [Run](#run)
+- [API](#api)
+- [SPA](#spa)
+- [Development](#development)
+- [Test](#test)
+- [Docker](#docker)
+- [Media Server Integration](#media-server-integration)
+- [Annual Wrap-Up](#annual-wrap-up)
+- [Contributing](#contributing)
+- [License](#license)
+
+---
+
+## Quick Start
+
+The fastest way to run Movies Diary is via Docker Compose:
+
+```bash
+cp .env.example .env
+# Set JWT_SECRET and OMDB_API_KEY (or TMDB_API_KEY) in .env
+docker compose up -d
+```
+
+Open `http://localhost:3000`. The HTTP server and background worker start together; data is persisted in a Docker volume.
+
+---
 
 ## Features
 
@@ -28,6 +71,18 @@ A self-hosted, server-side rendered movie logging system with a full REST API. B
 - Single-page app at `/app/` — React + TanStack Router + shadcn/ui, built with Vite, served from the backend with client-side routing fallback
 - Terminal UI client (`crates/tui`, deprecated) for logging reviews, bulk CSV import, and diary browsing
 
+## Screenshots
+
+> SPA at `/app/` — React + TanStack Router + shadcn/ui
+
+| Feed | Movie | Person |
+|------|-------|--------|
+| ![Feed](screenshots/feed.jpeg) | ![Movie detail](screenshots/movie.jpeg) | ![Person detail](screenshots/person.jpeg) |
+
+| Profile | Wrap-Up | Wrap-Up card |
+|---------|---------|--------------|
+| ![Profile](screenshots/profile.jpeg) | ![Wrap-Up stats](screenshots/wrapup-stats.jpeg) | ![Wrap-Up shareable card](screenshots/wrapup-card.jpeg) |
+
 ## Architecture
 
 Hexagonal (Ports & Adapters) with Domain-Driven Design:
@@ -75,59 +130,35 @@ spa/                — React SPA (TanStack Router + shadcn/ui + Vite); served a
 - Poster storage: local filesystem (zero deps) or an S3-compatible object store (e.g. MinIO)
 - An [OMDb API key](https://www.omdbapi.com/apikey.aspx)
 
-## Environment Variables
+## Configuration
 
-A `.env.example` file is provided at the repo root — copy it to `.env` and fill in your values.
+Copy `.env.example` to `.env` and set the values below. Required fields must be set before the server will start.
 
-```env
-# Database
-DATABASE_URL=sqlite://movies.db
-
-# Authentication
-JWT_SECRET=change-me
-
-# OMDb metadata
-OMDB_API_KEY=your-key
-
-# TMDb metadata + enrichment (optional — enables full cast/crew/genre data)
-# TMDB_API_KEY=your-key
-
-# Public base URL (used for ActivityPub actor URLs and canonical links)
-BASE_URL=https://yourdomain.example.com
-
-# Image storage — pick one backend:
-
-# Option A: local filesystem (zero deps)
-IMAGE_STORAGE_BACKEND=local
-IMAGE_STORAGE_PATH=./images
-
-# Option B: S3-compatible (MinIO, AWS S3, etc.)
-# IMAGE_STORAGE_BACKEND=s3
-# MINIO_ENDPOINT=http://localhost:9000
-# MINIO_BUCKET=posters
-# MINIO_REGION=minio
-# MINIO_ACCESS_KEY_ID=minioadmin
-# MINIO_SECRET_ACCESS_KEY=minioadmin
-
-# Image conversion (optional — converts stored images to AVIF or WebP to save space)
-# IMAGE_CONVERSION_ENABLED=false
-# IMAGE_CONVERSION_FORMAT=avif   # avif or webp
-
-# Optional
-HOST=0.0.0.0
-PORT=3000
-RATE_LIMIT=60           # requests per minute per IP (default: 60)
-ALLOW_REGISTRATION=true # set to false to disable new sign-ups
-SECURE_COOKIES=true     # set when serving over HTTPS
-RUST_LOG=presentation=info,tower_http=info,worker=info,application=info
-
-# CORS — comma-separated origins for SPA dev (omit or "*" for any)
-# CORS_ORIGINS=http://localhost:5173
-
-# Event bus — "db" (default, uses same database) or "nats"
-EVENT_BUS_BACKEND=db
-# NATS_URL=nats://localhost:4222   # required when EVENT_BUS_BACKEND=nats
-```
+| Variable | Default | Required | Description |
+|---|---|---|---|
+| `DATABASE_URL` | `sqlite://movies.db` | Yes | SQLite or PostgreSQL connection string |
+| `JWT_SECRET` | — | Yes | Secret for JWT signing — use a long random string |
+| `OMDB_API_KEY` | — | Yes | [OMDb](https://www.omdbapi.com/apikey.aspx) key for movie metadata |
+| `TMDB_API_KEY` | — | No | [TMDb](https://www.themoviedb.org/settings/api) key — enables cast, crew, genres, enrichment |
+| `BASE_URL` | — | Yes | Public URL of your instance (used for ActivityPub actor URLs) |
+| `IMAGE_STORAGE_BACKEND` | `local` | No | `local` or `s3` |
+| `IMAGE_STORAGE_PATH` | `./images` | No | Path for local image storage |
+| `MINIO_ENDPOINT` | — | S3 only | S3-compatible endpoint (e.g. `http://localhost:9000`) |
+| `MINIO_BUCKET` | — | S3 only | Bucket name |
+| `MINIO_REGION` | — | S3 only | Region (e.g. `minio`) |
+| `MINIO_ACCESS_KEY_ID` | — | S3 only | Access key ID |
+| `MINIO_SECRET_ACCESS_KEY` | — | S3 only | Secret access key |
+| `IMAGE_CONVERSION_ENABLED` | `false` | No | Convert stored images to AVIF or WebP |
+| `IMAGE_CONVERSION_FORMAT` | `avif` | No | `avif` or `webp` |
+| `HOST` | `0.0.0.0` | No | Bind address |
+| `PORT` | `3000` | No | HTTP port |
+| `RATE_LIMIT` | `60` | No | Requests per minute per IP |
+| `ALLOW_REGISTRATION` | `true` | No | Set `false` to disable new sign-ups |
+| `SECURE_COOKIES` | `true` | No | Must be `true` when serving over HTTPS |
+| `RUST_LOG` | — | No | Log verbosity (e.g. `presentation=info,worker=info`) |
+| `CORS_ORIGINS` | `*` | No | Comma-separated allowed origins for SPA dev |
+| `EVENT_BUS_BACKEND` | `db` | No | `db` (default) or `nats` |
+| `NATS_URL` | — | NATS only | NATS connection URL (e.g. `nats://localhost:4222`) |
 
 The `worker` binary must run alongside `presentation` to process events:
 
@@ -153,6 +184,8 @@ Interactive API documentation is available at runtime:
 - **Swagger UI** — `http://localhost:3000/docs`
 - **Scalar** — `http://localhost:3000/scalar`
 
+An [Insomnia](https://insomnia.rest/) collection covering all endpoints is included at [`movies-diary.insomnia.json`](movies-diary.insomnia.json). Import it via **File → Import**, set `base_url` and `token` in the environment, and you're ready to go.
+
 ## SPA
 
 The single-page app lives in `spa/` and is served at `/app/` by the backend. For local development:

crates/adapters/activitypub/src/event_handler.rs

@@ -4,7 +4,7 @@ use domain::ports::EventHandler;
 use domain::{
     errors::DomainError,
     events::DomainEvent,
-    ports::LocalApContentQuery,
+    ports::{LocalApContentQuery, UserFederationSettingsQuery},
     value_objects::{MovieId, ReviewId, UserId},
 };
 use std::sync::Arc;
@@ -17,6 +17,7 @@ use crate::urls::{actor_url, goal_url, review_url};
 pub struct ActivityPubEventHandler {
     ap_service: Arc<ActivityPubService>,
     content_query: Arc<dyn LocalApContentQuery>,
+    federation_settings: Arc<dyn UserFederationSettingsQuery>,
     base_url: String,
 }
 
@@ -24,11 +25,13 @@ impl ActivityPubEventHandler {
     pub fn new(
         ap_service: Arc<ActivityPubService>,
         content_query: Arc<dyn LocalApContentQuery>,
+        federation_settings: Arc<dyn UserFederationSettingsQuery>,
         base_url: String,
     ) -> Self {
         Self {
             ap_service,
             content_query,
+            federation_settings,
             base_url,
         }
     }
@@ -131,6 +134,19 @@ impl EventHandler for ActivityPubEventHandler {
 
 impl ActivityPubEventHandler {
     async fn on_review_logged(&self, user_id: &UserId, review_id: &ReviewId) -> anyhow::Result<()> {
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
+            .await
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.reviews {
+            return Ok(());
+        }
+
         let review = match self.content_query.get_review_by_id(review_id).await? {
             Some(r) => r,
             None => return Ok(()),
@@ -184,6 +200,19 @@ impl ActivityPubEventHandler {
         user_id: &UserId,
         review_id: &ReviewId,
     ) -> anyhow::Result<()> {
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
+            .await
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.reviews {
+            return Ok(());
+        }
+
         let review = match self.content_query.get_review_by_id(review_id).await? {
             Some(r) => r,
             None => return Ok(()),
@@ -250,6 +279,19 @@ impl ActivityPubEventHandler {
         external_metadata_id: &Option<String>,
         added_at: &chrono::NaiveDateTime,
     ) -> anyhow::Result<()> {
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
+            .await
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.watchlist {
+            return Ok(());
+        }
+
         use crate::urls::watchlist_entry_url;
         let ap_id = watchlist_entry_url(&self.base_url, user_id.value(), movie_id.value());
         let actor = actor_url(&self.base_url, user_id.value());
@@ -316,6 +358,20 @@ impl ActivityPubEventHandler {
         for entry in entries {
             let review = entry.review();
             let user_id = review.user_id();
+
+            let flags = self
+                .federation_settings
+                .get_federation_flags(user_id)
+                .await
+                .unwrap_or(domain::ports::FederationFlags {
+                    goals: true,
+                    reviews: true,
+                    watchlist: true,
+                });
+            if !flags.reviews {
+                continue;
+            }
+
             let ap_id = review_url(&self.base_url, review.id());
             let actor = actor_url(&self.base_url, user_id.value());
 
@@ -343,12 +399,16 @@ impl ActivityPubEventHandler {
         user_id: &UserId,
         year: u16,
     ) -> anyhow::Result<()> {
-        if !self
-            .content_query
-            .get_user_federate_goals(user_id)
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
             .await
-            .unwrap_or(false)
-        {
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.goals {
             return Ok(());
         }
         let Some((goal, current)) = self
@@ -384,12 +444,16 @@ impl ActivityPubEventHandler {
         target_count: u32,
         is_create: bool,
     ) -> anyhow::Result<()> {
-        if !self
-            .content_query
-            .get_user_federate_goals(user_id)
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
             .await
-            .unwrap_or(false)
-        {
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.goals {
             return Ok(());
         }
         let current = self
@@ -418,12 +482,16 @@ impl ActivityPubEventHandler {
     }
 
     async fn on_goal_deleted(&self, user_id: &UserId, year: u16) -> anyhow::Result<()> {
-        if !self
-            .content_query
-            .get_user_federate_goals(user_id)
+        let flags = self
+            .federation_settings
+            .get_federation_flags(user_id)
             .await
-            .unwrap_or(false)
-        {
+            .unwrap_or(domain::ports::FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            });
+        if !flags.goals {
             return Ok(());
         }
         let ap_id = goal_url(&self.base_url, user_id.value(), year);

crates/adapters/activitypub/src/lib.rs

@@ -52,6 +52,7 @@ pub struct ActivityPubDeps {
     pub remote_goal_repo: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
     pub local_ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
     pub user_repo: std::sync::Arc<dyn domain::ports::UserRepository>,
+    pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub base_url: String,
     pub allow_registration: bool,
     pub event_publisher: std::sync::Arc<dyn domain::ports::EventPublisher>,
@@ -68,6 +69,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
         remote_goal_repo,
         local_ap_content,
         user_repo,
+        federation_settings,
         base_url,
         allow_registration,
         event_publisher,
@@ -129,6 +131,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
     let event_handler = std::sync::Arc::new(ActivityPubEventHandler::new(
         std::sync::Arc::clone(&concrete),
         local_ap_content,
+        federation_settings,
         base_url,
     )) as std::sync::Arc<dyn domain::ports::EventHandler>;
 

crates/adapters/event-payload/src/lib.rs

@@ -2,7 +2,7 @@ use chrono::NaiveDateTime;
 use domain::{
     errors::DomainError,
     events::DomainEvent,
-    models::PersonId,
+    models::{ExternalPersonId, PersonId},
     value_objects::{
         ExternalMetadataId, GoalId, MovieId, PosterPath, Rating, ReviewId, UserId, WrapUpId,
     },
@@ -210,7 +210,7 @@ impl From<&DomainEvent> for EventPayload {
                 external_metadata_id,
             } => EventPayload::MovieEnrichmentRequested {
                 movie_id: movie_id.value().to_string(),
-                external_metadata_id: external_metadata_id.clone(),
+                external_metadata_id: external_metadata_id.value().to_string(),
             },
             DomainEvent::ImageStored { key } => EventPayload::ImageStored { key: key.clone() },
             DomainEvent::WatchlistEntryAdded {
@@ -322,7 +322,7 @@ impl From<&DomainEvent> for EventPayload {
                 external_person_id,
             } => EventPayload::PersonEnrichmentRequested {
                 person_id: person_id.value().to_string(),
-                external_person_id: external_person_id.clone(),
+                external_person_id: external_person_id.value().to_string(),
             },
         }
     }
@@ -391,7 +391,8 @@ impl TryFrom<EventPayload> for DomainEvent {
                 external_metadata_id,
             } => Ok(DomainEvent::MovieEnrichmentRequested {
                 movie_id: MovieId::from_uuid(parse_uuid(&movie_id, "movie_id")?),
-                external_metadata_id,
+                external_metadata_id: ExternalMetadataId::new(external_metadata_id)
+                    .map_err(|e| DomainError::InfrastructureError(e.to_string()))?,
             }),
             EventPayload::ImageStored { key } => Ok(DomainEvent::ImageStored { key }),
             EventPayload::WatchlistEntryAdded {
@@ -514,7 +515,7 @@ impl TryFrom<EventPayload> for DomainEvent {
                 external_person_id,
             } => Ok(DomainEvent::PersonEnrichmentRequested {
                 person_id: PersonId::from_uuid(parse_uuid(&person_id, "person_id")?),
-                external_person_id,
+                external_person_id: ExternalPersonId::new(external_person_id),
             }),
         }
     }

crates/adapters/export/Cargo.toml

@@ -8,6 +8,9 @@ domain = { workspace = true }
 async-trait = { workspace = true }
 serde_json = { workspace = true }
 chrono = { workspace = true }
+futures = { workspace = true }
+bytes = { workspace = true }
+async-stream = { workspace = true }
 
 [dev-dependencies]
 uuid = { workspace = true }

crates/adapters/export/src/lib.rs

@@ -1,51 +1,89 @@
-use async_trait::async_trait;
+use bytes::Bytes;
 use domain::{
     errors::DomainError,
     models::{DiaryEntry, ExportFormat},
     ports::DiaryExporter,
 };
+use futures::stream::BoxStream;
 
 pub struct ExportAdapter;
 
-#[async_trait]
 impl DiaryExporter for ExportAdapter {
-    async fn serialize_entries(
+    fn stream_entries(
         &self,
-        entries: &[DiaryEntry],
+        stream: BoxStream<'static, Result<DiaryEntry, DomainError>>,
         format: ExportFormat,
-    ) -> Result<Vec<u8>, DomainError> {
+    ) -> BoxStream<'static, Result<Bytes, DomainError>> {
         match format {
-            ExportFormat::Csv => serialize_csv(entries),
-            ExportFormat::Json => serialize_json(entries),
+            ExportFormat::Csv => stream_csv(stream),
+            ExportFormat::Json => stream_json(stream),
         }
     }
 }
 
-fn serialize_csv(entries: &[DiaryEntry]) -> Result<Vec<u8>, DomainError> {
-    let mut out =
-        String::from("title,year,director,rating,comment,watched_at,external_metadata_id\n");
-    for e in entries {
-        let title = csv_escape(e.movie().title().value());
-        let year = e.movie().release_year().value();
-        let director = e.movie().director().map(csv_escape).unwrap_or_default();
-        let rating = e.review().rating().value();
-        let comment = e
-            .review()
-            .comment()
-            .map(|c| csv_escape(c.value()))
-            .unwrap_or_default();
-        let watched_at = e.review().watched_at().format("%Y-%m-%d");
-        let ext_id = e
-            .movie()
-            .external_metadata_id()
-            .map(|id| id.value().to_string())
-            .unwrap_or_default();
-        out.push_str(&format!(
-            "{},{},{},{},{},{},{}\n",
-            title, year, director, rating, comment, watched_at, ext_id
-        ));
-    }
-    Ok(out.into_bytes())
+fn stream_csv(
+    entries: BoxStream<'static, Result<DiaryEntry, DomainError>>,
+) -> BoxStream<'static, Result<Bytes, DomainError>> {
+    use futures::StreamExt;
+    let header = futures::stream::once(async {
+        Ok(Bytes::from_static(
+            b"title,year,director,rating,comment,watched_at,external_metadata_id\n",
+        ))
+    });
+    let rows = entries.map(|r| r.map(|e| Bytes::from(csv_row(&e))));
+    Box::pin(header.chain(rows))
+}
+
+fn stream_json(
+    stream: BoxStream<'static, Result<DiaryEntry, DomainError>>,
+) -> BoxStream<'static, Result<Bytes, DomainError>> {
+    Box::pin(async_stream::stream! {
+        futures::pin_mut!(stream);
+        let mut is_first = true;
+        while let Some(r) = futures::StreamExt::next(&mut stream).await {
+            match r {
+                Err(e) => { yield Err(e); return; }
+                Ok(entry) => {
+                    let json = serde_json::to_string(&entry_to_json(&entry))
+                        .map_err(|e| DomainError::InfrastructureError(e.to_string()));
+                    let json = match json {
+                        Ok(s) => s,
+                        Err(e) => { yield Err(e); return; }
+                    };
+                    let prefix = if is_first { "[" } else { "," };
+                    is_first = false;
+                    yield Ok(Bytes::from(format!("{}{}", prefix, json)));
+                }
+            }
+        }
+        if is_first {
+            yield Ok(Bytes::from_static(b"[]"));
+        } else {
+            yield Ok(Bytes::from_static(b"]"));
+        }
+    })
+}
+
+fn csv_row(e: &DiaryEntry) -> String {
+    let title = csv_escape(e.movie().title().value());
+    let year = e.movie().release_year().value();
+    let director = e.movie().director().map(csv_escape).unwrap_or_default();
+    let rating = e.review().rating().value();
+    let comment = e
+        .review()
+        .comment()
+        .map(|c| csv_escape(c.value()))
+        .unwrap_or_default();
+    let watched_at = e.review().watched_at().format("%Y-%m-%d");
+    let ext_id = e
+        .movie()
+        .external_metadata_id()
+        .map(|id| id.value().to_string())
+        .unwrap_or_default();
+    format!(
+        "{},{},{},{},{},{},{}\n",
+        title, year, director, rating, comment, watched_at, ext_id
+    )
 }
 
 fn csv_escape(s: &str) -> String {
@@ -56,22 +94,16 @@ fn csv_escape(s: &str) -> String {
     }
 }
 
-fn serialize_json(entries: &[DiaryEntry]) -> Result<Vec<u8>, DomainError> {
-    let arr: Vec<serde_json::Value> = entries
-        .iter()
-        .map(|e| {
-            serde_json::json!({
-                "title": e.movie().title().value(),
-                "year": e.movie().release_year().value(),
-                "director": e.movie().director(),
-                "rating": e.review().rating().value(),
-                "comment": e.review().comment().map(|c| c.value()),
-                "watched_at": e.review().watched_at().format("%Y-%m-%d").to_string(),
-                "external_metadata_id": e.movie().external_metadata_id().map(|id| id.value()),
-            })
-        })
-        .collect();
-    serde_json::to_vec_pretty(&arr).map_err(|e| DomainError::InfrastructureError(e.to_string()))
+fn entry_to_json(e: &DiaryEntry) -> serde_json::Value {
+    serde_json::json!({
+        "title": e.movie().title().value(),
+        "year": e.movie().release_year().value(),
+        "director": e.movie().director(),
+        "rating": e.review().rating().value(),
+        "comment": e.review().comment().map(|c| c.value().to_string()),
+        "watched_at": e.review().watched_at().format("%Y-%m-%d").to_string(),
+        "external_metadata_id": e.movie().external_metadata_id().map(|id| id.value().to_string()),
+    })
 }
 
 #[cfg(test)]

crates/adapters/export/src/tests/lib.rs

@@ -5,6 +5,27 @@ use domain::{
     value_objects::{ExternalMetadataId, MovieTitle, Rating, ReleaseYear},
 };
 
+async fn collect_stream(
+    stream: futures::stream::BoxStream<'static, Result<bytes::Bytes, domain::errors::DomainError>>,
+) -> Vec<u8> {
+    use futures::StreamExt;
+    let mut out = Vec::new();
+    futures::pin_mut!(stream);
+    while let Some(chunk) = stream.next().await {
+        out.extend_from_slice(&chunk.unwrap());
+    }
+    out
+}
+
+fn entry_stream(
+    entries: Vec<domain::models::DiaryEntry>,
+) -> futures::stream::BoxStream<
+    'static,
+    Result<domain::models::DiaryEntry, domain::errors::DomainError>,
+> {
+    Box::pin(futures::stream::iter(entries.into_iter().map(Ok)))
+}
+
 fn make_entry(
     title: &str,
     year: u16,
@@ -55,10 +76,8 @@ async fn csv_has_header_and_one_row() {
         5,
         Some("great"),
     );
-    let bytes = adapter
-        .serialize_entries(&[entry], ExportFormat::Csv)
-        .await
-        .unwrap();
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![entry]), ExportFormat::Csv)).await;
     let text = String::from_utf8(bytes).unwrap();
     assert!(
         text.starts_with("title,year,director,rating,comment,watched_at,external_metadata_id\n")
@@ -75,10 +94,8 @@ async fn csv_has_header_and_one_row() {
 async fn csv_escapes_commas_in_title() {
     let adapter = ExportAdapter;
     let entry = make_entry("Tár, A Film", 2022, None, 4, None);
-    let bytes = adapter
-        .serialize_entries(&[entry], ExportFormat::Csv)
-        .await
-        .unwrap();
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![entry]), ExportFormat::Csv)).await;
     let text = String::from_utf8(bytes).unwrap();
     assert!(text.contains("\"Tár, A Film\""));
 }
@@ -87,10 +104,8 @@ async fn csv_escapes_commas_in_title() {
 async fn json_is_valid_array() {
     let adapter = ExportAdapter;
     let entry = make_entry("Dune", 2021, Some("Denis Villeneuve"), 5, None);
-    let bytes = adapter
-        .serialize_entries(&[entry], ExportFormat::Json)
-        .await
-        .unwrap();
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![entry]), ExportFormat::Json)).await;
     let arr: Vec<serde_json::Value> = serde_json::from_slice(&bytes).unwrap();
     assert_eq!(arr.len(), 1);
     assert_eq!(arr[0]["title"], "Dune");
@@ -104,27 +119,23 @@ async fn json_is_valid_array() {
 async fn external_metadata_id_included_when_present() {
     let adapter = ExportAdapter;
     let entry = make_entry_full("Alien", 1979, None, 5, None, Some("tt0078748"));
-    let bytes = adapter
-        .serialize_entries(&[entry], ExportFormat::Json)
-        .await
-        .unwrap();
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![entry]), ExportFormat::Json)).await;
     let arr: Vec<serde_json::Value> = serde_json::from_slice(&bytes).unwrap();
     assert_eq!(arr[0]["external_metadata_id"], "tt0078748");
 
-    let bytes = adapter
-        .serialize_entries(
-            &[make_entry_full(
-                "Alien",
-                1979,
-                None,
-                5,
-                None,
-                Some("tt0078748"),
-            )],
-            ExportFormat::Csv,
-        )
-        .await
-        .unwrap();
+    let bytes = collect_stream(adapter.stream_entries(
+        entry_stream(vec![make_entry_full(
+            "Alien",
+            1979,
+            None,
+            5,
+            None,
+            Some("tt0078748"),
+        )]),
+        ExportFormat::Csv,
+    ))
+    .await;
     let text = String::from_utf8(bytes).unwrap();
     assert!(text.contains("tt0078748"));
 }
@@ -132,13 +143,20 @@ async fn external_metadata_id_included_when_present() {
 #[tokio::test]
 async fn empty_entries_returns_csv_header_only() {
     let adapter = ExportAdapter;
-    let bytes = adapter
-        .serialize_entries(&[], ExportFormat::Csv)
-        .await
-        .unwrap();
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![]), ExportFormat::Csv)).await;
     let text = String::from_utf8(bytes).unwrap();
     assert_eq!(
         text,
         "title,year,director,rating,comment,watched_at,external_metadata_id\n"
     );
 }
+
+#[tokio::test]
+async fn empty_json_is_valid_empty_array() {
+    let adapter = ExportAdapter;
+    let bytes =
+        collect_stream(adapter.stream_entries(entry_stream(vec![]), ExportFormat::Json)).await;
+    let arr: Vec<serde_json::Value> = serde_json::from_slice(&bytes).unwrap();
+    assert!(arr.is_empty());
+}

crates/adapters/poster-sync/src/lib.rs

@@ -123,7 +123,7 @@ impl EventHandler for PosterSyncHandler {
                 if already_has_poster {
                     return Ok(());
                 }
-                (movie_id.value(), external_metadata_id.clone())
+                (movie_id.value(), external_metadata_id.value().to_owned())
             }
             _ => return Ok(()),
         };

crates/adapters/postgres/Cargo.toml

@@ -20,3 +20,6 @@ async-trait = { workspace = true }
 tokio       = { workspace = true }
 serde       = { workspace = true, features = ["derive"] }
 serde_json  = { workspace = true }
+futures = { workspace = true }
+bytes = { workspace = true }
+async-stream = { workspace = true }

crates/adapters/postgres/migrations/0028_federation_privacy.sql

@@ -0,0 +1,2 @@
+ALTER TABLE user_settings ADD COLUMN federate_reviews BOOLEAN NOT NULL DEFAULT TRUE;
+ALTER TABLE user_settings ADD COLUMN federate_watchlist BOOLEAN NOT NULL DEFAULT TRUE;

crates/adapters/postgres/src/ap_content.rs

@@ -229,23 +229,6 @@ impl LocalApContentQuery for PostgresApContentQuery {
         rows.into_iter().map(DiaryRow::into_domain).collect()
     }
 
-    async fn get_user_federate_goals(&self, user_id: &UserId) -> Result<bool, DomainError> {
-        let uid = user_id.value().to_string();
-        let row = sqlx::query("SELECT federate_goals FROM user_settings WHERE user_id = $1")
-            .bind(&uid)
-            .fetch_optional(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
-        match row {
-            Some(r) => {
-                let val: i64 = r.try_get("federate_goals").unwrap_or(0);
-                Ok(val != 0)
-            }
-            None => Ok(false),
-        }
-    }
-
     async fn get_goal_with_progress(
         &self,
         user_id: &UserId,

crates/adapters/postgres/src/diary.rs

@@ -8,6 +8,7 @@ use domain::{
     ports::DiaryRepository,
     value_objects::{MovieId, UserId},
 };
+use futures::stream::BoxStream;
 use sqlx::PgPool;
 
 use crate::models::{DiaryRow, FeedRow, MovieRow, MovieStatsRow, ReviewRow};
@@ -427,6 +428,35 @@ impl DiaryRepository for PostgresDiaryRepository {
         rows.into_iter().map(DiaryRow::into_domain).collect()
     }
 
+    fn stream_user_history(
+        &self,
+        user_id: UserId,
+    ) -> BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        let pool = self.pool.clone();
+        let uid = user_id.value().to_string();
+        Box::pin(async_stream::stream! {
+            let mut rows = sqlx::query_as::<_, DiaryRow>(
+                "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                        r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment,
+                        to_char(r.watched_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS watched_at,
+                        to_char(r.created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS created_at,
+                        r.remote_actor_url
+                 FROM reviews r
+                 INNER JOIN movies m ON m.id = r.movie_id
+                 WHERE r.user_id = $1
+                 ORDER BY r.watched_at DESC",
+            )
+            .bind(&uid)
+            .fetch(&pool);
+            while let Some(row) = futures::StreamExt::next(&mut rows).await {
+                yield match row {
+                    Ok(r) => r.into_domain(),
+                    Err(e) => Err(Self::map_err(e)),
+                };
+            }
+        })
+    }
+
     async fn get_movie_stats(&self, movie_id: &MovieId) -> Result<MovieStats, DomainError> {
         let id_str = movie_id.value().to_string();
         sqlx::query_as::<_, MovieStatsRow>(

crates/adapters/postgres/src/import_session.rs

@@ -5,6 +5,7 @@ use domain::{
     models::{
         AnnotatedRow, FieldMapping, ImportSession, ParsedFile,
         import::{DomainField, ImportRow, RowResult, Transform},
+        import_session::PersistedImportSession,
     },
     ports::ImportSessionRepository,
     value_objects::{ImportSessionId, UserId},
@@ -266,7 +267,7 @@ impl PostgresImportSessionRepository {
                 Ok(js.into_iter().map(annotated_from_json).collect())
             })
             .transpose()?;
-        Ok(ImportSession {
+        Ok(ImportSession::from_persistence(PersistedImportSession {
             id: ImportSessionId::from_uuid(
                 id.parse::<uuid::Uuid>()
                     .map_err(|e| DomainError::InfrastructureError(e.to_string()))?,
@@ -281,7 +282,7 @@ impl PostgresImportSessionRepository {
             row_results,
             created_at,
             expires_at,
-        })
+        }))
     }
 }
 

crates/adapters/postgres/src/lib.rs

@@ -93,6 +93,7 @@ pub struct PostgresWireOutput {
     pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
     pub goal: std::sync::Arc<dyn domain::ports::GoalRepository>,
     pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
+    pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
 }
 
@@ -108,6 +109,10 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
         .map_err(|e| anyhow::anyhow!("{e}"))
         .context("Database migration failed")?;
 
+    let user_settings_repo = std::sync::Arc::new(
+        user_settings::PostgresUserSettingsRepository::new(pool.clone()),
+    );
+
     Ok(PostgresWireOutput {
         pool: pool.clone(),
         movie: std::sync::Arc::new(PostgresMovieRepository::new(pool.clone())) as _,
@@ -125,9 +130,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
         wrapup_repo: std::sync::Arc::new(PostgresWrapUpRepository::new(pool.clone())) as _,
         wrapup_stats: std::sync::Arc::new(PostgresWrapUpStatsQuery::new(pool.clone())) as _,
         goal: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
-        user_settings: std::sync::Arc::new(user_settings::PostgresUserSettingsRepository::new(
-            pool.clone(),
-        )) as _,
+        user_settings: std::sync::Arc::clone(&user_settings_repo) as _,
+        federation_settings: user_settings_repo as _,
         remote_goal: std::sync::Arc::new(remote_goals::PostgresRemoteGoalRepository::new(pool))
             as _,
     })

crates/adapters/postgres/src/user_settings.rs

@@ -1,6 +1,9 @@
 use async_trait::async_trait;
 use domain::{
-    errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
+    errors::DomainError,
+    models::UserSettings,
+    ports::{FederationFlags, UserFederationSettingsQuery, UserSettingsRepository},
+    value_objects::UserId,
 };
 use sqlx::{PgPool, Row};
 
@@ -23,20 +26,25 @@ impl PostgresUserSettingsRepository {
 impl UserSettingsRepository for PostgresUserSettingsRepository {
     async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError> {
         let uid = user_id.value().to_string();
-
-        let row =
-            sqlx::query("SELECT user_id, federate_goals FROM user_settings WHERE user_id = $1")
-                .bind(&uid)
-                .fetch_optional(&self.pool)
-                .await
-                .map_err(Self::map_err)?;
+        let row = sqlx::query(
+            "SELECT federate_goals, federate_reviews, federate_watchlist \
+             FROM user_settings WHERE user_id = $1",
+        )
+        .bind(&uid)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(Self::map_err)?;
 
         match row {
             Some(r) => {
-                let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
+                let goals: bool = r.try_get("federate_goals").unwrap_or(true);
+                let reviews: bool = r.try_get("federate_reviews").unwrap_or(true);
+                let watchlist: bool = r.try_get("federate_watchlist").unwrap_or(true);
                 Ok(UserSettings::from_persistence(
                     user_id.clone(),
-                    federate != 0,
+                    goals,
+                    reviews,
+                    watchlist,
                 ))
             }
             None => Ok(UserSettings::new(user_id.clone())),
@@ -45,18 +53,52 @@ impl UserSettingsRepository for PostgresUserSettingsRepository {
 
     async fn save(&self, settings: &UserSettings) -> Result<(), DomainError> {
         let uid = settings.user_id().value().to_string();
-        let federate = if settings.federate_goals() { 1i64 } else { 0 };
-
         sqlx::query(
-            "INSERT INTO user_settings (user_id, federate_goals) VALUES ($1, $2) \
-             ON CONFLICT (user_id) DO UPDATE SET federate_goals = $2",
+            "INSERT INTO user_settings (user_id, federate_goals, federate_reviews, federate_watchlist) \
+             VALUES ($1, $2, $3, $4) \
+             ON CONFLICT (user_id) DO UPDATE \
+             SET federate_goals = $2, federate_reviews = $3, federate_watchlist = $4",
         )
         .bind(&uid)
-        .bind(federate)
+        .bind(settings.federate_goals())
+        .bind(settings.federate_reviews())
+        .bind(settings.federate_watchlist())
         .execute(&self.pool)
         .await
         .map_err(Self::map_err)?;
-
         Ok(())
     }
 }
+
+#[async_trait]
+impl UserFederationSettingsQuery for PostgresUserSettingsRepository {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
+        let uid = user_id.value().to_string();
+        let row = sqlx::query(
+            "SELECT federate_goals, federate_reviews, federate_watchlist \
+             FROM user_settings WHERE user_id = $1",
+        )
+        .bind(&uid)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(Self::map_err)?;
+
+        match row {
+            Some(r) => {
+                let goals: bool = r.try_get("federate_goals").unwrap_or(true);
+                let reviews: bool = r.try_get("federate_reviews").unwrap_or(true);
+                let watchlist: bool = r.try_get("federate_watchlist").unwrap_or(true);
+                Ok(FederationFlags {
+                    goals,
+                    reviews,
+                    watchlist,
+                })
+            }
+            None => Ok(FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            }),
+        }
+    }
+}

crates/adapters/sqlite/Cargo.toml

@@ -20,3 +20,6 @@ chrono = { workspace = true }
 tracing = { workspace = true }
 async-trait = { workspace = true }
 tokio = { workspace = true }
+futures = { workspace = true }
+bytes = { workspace = true }
+async-stream = { workspace = true }

crates/adapters/sqlite/migrations/0028_federation_privacy.sql

@@ -0,0 +1,2 @@
+ALTER TABLE user_settings ADD COLUMN federate_reviews BOOLEAN NOT NULL DEFAULT TRUE;
+ALTER TABLE user_settings ADD COLUMN federate_watchlist BOOLEAN NOT NULL DEFAULT TRUE;

crates/adapters/sqlite/src/ap_content.rs

@@ -5,7 +5,7 @@ use domain::{
     ports::LocalApContentQuery,
     value_objects::{MovieId, ReviewId, UserId},
 };
-use sqlx::{Row, SqlitePool};
+use sqlx::SqlitePool;
 
 use crate::models::{DiaryRow, MovieRow, ReviewRow, WatchlistRow};
 
@@ -169,23 +169,6 @@ impl LocalApContentQuery for SqliteApContentQuery {
         rows.into_iter().map(DiaryRow::into_domain).collect()
     }
 
-    async fn get_user_federate_goals(&self, user_id: &UserId) -> Result<bool, DomainError> {
-        let uid = user_id.value().to_string();
-        let row = sqlx::query("SELECT federate_goals FROM user_settings WHERE user_id = ?")
-            .bind(&uid)
-            .fetch_optional(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
-        match row {
-            Some(r) => {
-                let val: i64 = r.try_get("federate_goals").unwrap_or(0);
-                Ok(val != 0)
-            }
-            None => Ok(false),
-        }
-    }
-
     async fn get_goal_with_progress(
         &self,
         user_id: &UserId,

crates/adapters/sqlite/src/diary.rs

@@ -8,6 +8,7 @@ use domain::{
     ports::DiaryRepository,
     value_objects::{MovieId, UserId},
 };
+use futures::stream::BoxStream;
 use sqlx::SqlitePool;
 
 use crate::models::{DiaryRow, FeedRow, MovieRow, MovieStatsRow, ReviewRow};
@@ -389,6 +390,32 @@ impl DiaryRepository for SqliteDiaryRepository {
         rows.into_iter().map(DiaryRow::into_domain).collect()
     }
 
+    fn stream_user_history(
+        &self,
+        user_id: UserId,
+    ) -> BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        let pool = self.pool.clone();
+        let uid = user_id.value().to_string();
+        Box::pin(async_stream::stream! {
+            let mut rows = sqlx::query_as::<_, DiaryRow>(
+                "SELECT m.id, m.external_metadata_id, m.title, m.release_year, m.director, m.poster_path,
+                        r.id AS review_id, r.movie_id, r.user_id, r.rating, r.comment, r.watched_at, r.created_at, r.remote_actor_url
+                 FROM reviews r
+                 INNER JOIN movies m ON m.id = r.movie_id
+                 WHERE r.user_id = ?
+                 ORDER BY r.watched_at DESC",
+            )
+            .bind(&uid)
+            .fetch(&pool);
+            while let Some(row) = futures::StreamExt::next(&mut rows).await {
+                yield match row {
+                    Ok(r) => r.into_domain(),
+                    Err(e) => Err(Self::map_err(e)),
+                };
+            }
+        })
+    }
+
     async fn get_movie_stats(&self, movie_id: &MovieId) -> Result<MovieStats, DomainError> {
         let id_str = movie_id.value().to_string();
         sqlx::query_as::<_, MovieStatsRow>(

crates/adapters/sqlite/src/import_session.rs

@@ -5,6 +5,7 @@ use domain::{
     models::{
         AnnotatedRow, FieldMapping, ImportSession, ParsedFile,
         import::{DomainField, ImportRow, RowResult, Transform},
+        import_session::PersistedImportSession,
     },
     ports::ImportSessionRepository,
     value_objects::{ImportSessionId, UserId},
@@ -275,7 +276,7 @@ impl SqliteImportSessionRepository {
             })
             .transpose()?;
 
-        Ok(ImportSession {
+        Ok(ImportSession::from_persistence(PersistedImportSession {
             id: ImportSessionId::from_uuid(
                 id.parse::<uuid::Uuid>()
                     .map_err(|e| DomainError::InfrastructureError(e.to_string()))?,
@@ -290,7 +291,7 @@ impl SqliteImportSessionRepository {
             row_results,
             created_at: Self::parse_dt(created_at)?,
             expires_at: Self::parse_dt(expires_at)?,
-        })
+        }))
     }
 }
 

crates/adapters/sqlite/src/lib.rs

@@ -89,6 +89,7 @@ pub struct SqliteWireOutput {
     pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
     pub goal: std::sync::Arc<dyn domain::ports::GoalRepository>,
     pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
+    pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub remote_goal: std::sync::Arc<dyn domain::ports::RemoteGoalRepository>,
 }
 
@@ -113,6 +114,10 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
         .map_err(|e| anyhow::anyhow!("{e}"))
         .context("Database migration failed")?;
 
+    let user_settings_repo = std::sync::Arc::new(user_settings::SqliteUserSettingsRepository::new(
+        pool.clone(),
+    ));
+
     Ok(SqliteWireOutput {
         pool: pool.clone(),
         movie: std::sync::Arc::new(SqliteMovieRepository::new(pool.clone())) as _,
@@ -128,9 +133,8 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
         wrapup_repo: std::sync::Arc::new(SqliteWrapUpRepository::new(pool.clone())) as _,
         wrapup_stats: std::sync::Arc::new(SqliteWrapUpStatsQuery::new(pool.clone())) as _,
         goal: std::sync::Arc::new(goals::SqliteGoalRepository::new(pool.clone())) as _,
-        user_settings: std::sync::Arc::new(user_settings::SqliteUserSettingsRepository::new(
-            pool.clone(),
-        )) as _,
+        user_settings: std::sync::Arc::clone(&user_settings_repo) as _,
+        federation_settings: user_settings_repo as _,
         remote_goal: std::sync::Arc::new(remote_goals::SqliteRemoteGoalRepository::new(pool)) as _,
     })
 }

crates/adapters/sqlite/src/user_settings.rs

@@ -1,6 +1,9 @@
 use async_trait::async_trait;
 use domain::{
-    errors::DomainError, models::UserSettings, ports::UserSettingsRepository, value_objects::UserId,
+    errors::DomainError,
+    models::UserSettings,
+    ports::{FederationFlags, UserFederationSettingsQuery, UserSettingsRepository},
+    value_objects::UserId,
 };
 use sqlx::{Row, SqlitePool};
 
@@ -23,20 +26,25 @@ impl SqliteUserSettingsRepository {
 impl UserSettingsRepository for SqliteUserSettingsRepository {
     async fn get(&self, user_id: &UserId) -> Result<UserSettings, DomainError> {
         let uid = user_id.value().to_string();
-
-        let row =
-            sqlx::query("SELECT user_id, federate_goals FROM user_settings WHERE user_id = ?")
-                .bind(&uid)
-                .fetch_optional(&self.pool)
-                .await
-                .map_err(Self::map_err)?;
+        let row = sqlx::query(
+            "SELECT federate_goals, federate_reviews, federate_watchlist \
+             FROM user_settings WHERE user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(Self::map_err)?;
 
         match row {
             Some(r) => {
-                let federate: i64 = r.try_get("federate_goals").unwrap_or(0);
+                let goals: i64 = r.try_get("federate_goals").unwrap_or(1);
+                let reviews: i64 = r.try_get("federate_reviews").unwrap_or(1);
+                let watchlist: i64 = r.try_get("federate_watchlist").unwrap_or(1);
                 Ok(UserSettings::from_persistence(
                     user_id.clone(),
-                    federate != 0,
+                    goals != 0,
+                    reviews != 0,
+                    watchlist != 0,
                 ))
             }
             None => Ok(UserSettings::new(user_id.clone())),
@@ -45,15 +53,55 @@ impl UserSettingsRepository for SqliteUserSettingsRepository {
 
     async fn save(&self, settings: &UserSettings) -> Result<(), DomainError> {
         let uid = settings.user_id().value().to_string();
-        let federate = if settings.federate_goals() { 1i64 } else { 0 };
-
-        sqlx::query("INSERT OR REPLACE INTO user_settings (user_id, federate_goals) VALUES (?, ?)")
-            .bind(&uid)
-            .bind(federate)
-            .execute(&self.pool)
-            .await
-            .map_err(Self::map_err)?;
-
+        sqlx::query(
+            "INSERT OR REPLACE INTO user_settings \
+             (user_id, federate_goals, federate_reviews, federate_watchlist) \
+             VALUES (?, ?, ?, ?)",
+        )
+        .bind(&uid)
+        .bind(if settings.federate_goals() { 1i64 } else { 0 })
+        .bind(if settings.federate_reviews() { 1i64 } else { 0 })
+        .bind(if settings.federate_watchlist() {
+            1i64
+        } else {
+            0
+        })
+        .execute(&self.pool)
+        .await
+        .map_err(Self::map_err)?;
         Ok(())
     }
 }
+
+#[async_trait]
+impl UserFederationSettingsQuery for SqliteUserSettingsRepository {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
+        let uid = user_id.value().to_string();
+        let row = sqlx::query(
+            "SELECT federate_goals, federate_reviews, federate_watchlist \
+             FROM user_settings WHERE user_id = ?",
+        )
+        .bind(&uid)
+        .fetch_optional(&self.pool)
+        .await
+        .map_err(Self::map_err)?;
+
+        match row {
+            Some(r) => {
+                let goals: i64 = r.try_get("federate_goals").unwrap_or(1);
+                let reviews: i64 = r.try_get("federate_reviews").unwrap_or(1);
+                let watchlist: i64 = r.try_get("federate_watchlist").unwrap_or(1);
+                Ok(FederationFlags {
+                    goals: goals != 0,
+                    reviews: reviews != 0,
+                    watchlist: watchlist != 0,
+                })
+            }
+            None => Ok(FederationFlags {
+                goals: true,
+                reviews: true,
+                watchlist: true,
+            }),
+        }
+    }
+}

crates/adapters/tmdb-enrichment/src/movie_handler.rs

@@ -83,7 +83,7 @@ impl EventHandler for MovieEnrichmentHandler {
             self.enrichment_client.as_ref(),
             &self.profile_repo,
             movie_id.clone(),
-            &external_metadata_id,
+            external_metadata_id.value(),
         )
         .await?
         else {

crates/adapters/tmdb-enrichment/src/person_handler.rs

@@ -40,6 +40,7 @@ impl EventHandler for PersonEnrichmentHandler {
             _ => return Ok(()),
         };
 
-        application::person::enrich::execute(&self.deps, person_id, &external_person_id).await
+        application::person::enrich::execute(&self.deps, person_id, external_person_id.value())
+            .await
     }
 }

crates/api-types/src/goals.rs

@@ -29,9 +29,13 @@ pub struct UpdateGoalRequest {
 #[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
 pub struct UserSettingsDto {
     pub federate_goals: bool,
+    pub federate_reviews: bool,
+    pub federate_watchlist: bool,
 }
 
 #[derive(Debug, Clone, Deserialize, utoipa::ToSchema)]
 pub struct UpdateUserSettingsRequest {
     pub federate_goals: bool,
+    pub federate_reviews: bool,
+    pub federate_watchlist: bool,
 }

crates/application/Cargo.toml

@@ -15,6 +15,7 @@ sha2 = { workspace = true }
 rand = { workspace = true }
 hex = { workspace = true }
 serde_json = { workspace = true }
+bytes = { workspace = true }
 
 [features]
 xlsx = []

crates/application/src/diary/export_diary.rs

@@ -1,22 +1,21 @@
 use std::sync::Arc;
 
+use bytes::Bytes;
 use domain::{
     errors::DomainError,
     ports::{DiaryExporter, DiaryRepository},
     value_objects::UserId,
 };
+use futures::stream::BoxStream;
 
 use crate::diary::queries::ExportQuery;
 
-pub async fn execute(
+pub fn execute(
     diary: &Arc<dyn DiaryRepository>,
     diary_exporter: &Arc<dyn DiaryExporter>,
     query: ExportQuery,
-) -> Result<Vec<u8>, DomainError> {
-    let entries = diary
-        .get_user_history(&UserId::from_uuid(query.user_id))
-        .await?;
-    diary_exporter
-        .serialize_entries(&entries, query.format)
-        .await
+) -> BoxStream<'static, Result<Bytes, DomainError>> {
+    let user_id = UserId::from_uuid(query.user_id);
+    let entry_stream = diary.stream_user_history(user_id);
+    diary_exporter.stream_entries(entry_stream, query.format)
 }

crates/application/src/diary/review_logger.rs

@@ -112,7 +112,7 @@ async fn publish_events(
         publisher
             .publish(&DomainEvent::MovieEnrichmentRequested {
                 movie_id: movie.id().clone(),
-                external_metadata_id: ext_id.value().to_string(),
+                external_metadata_id: ext_id.clone(),
             })
             .await?;
     }

crates/application/src/import/create_session.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use chrono::Utc;
 use domain::{
     errors::DomainError,
     models::ImportSession,
@@ -31,8 +30,7 @@ pub async fn execute(
     let sample_rows = parsed.rows.iter().take(5).cloned().collect();
     let columns = parsed.columns.clone();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(ImportSessionId::generate(), user_id, now);
+    let mut session = ImportSession::new(user_id);
     let session_id = session.id.clone();
     session.parsed_file = Some(parsed);
 

crates/application/src/import/execute.rs

@@ -15,6 +15,8 @@ use crate::{
     ports::ReviewLogger,
 };
 
+const CONCURRENCY_LIMIT: usize = 10;
+
 pub struct ImportSummary {
     pub imported: usize,
     pub skipped_duplicates: usize,
@@ -41,22 +43,38 @@ pub async fn execute(
     let mut skipped_duplicates = 0;
     let mut failed = Vec::new();
 
+    let semaphore = Arc::new(tokio::sync::Semaphore::new(CONCURRENCY_LIMIT));
+    let mut tasks: tokio::task::JoinSet<(usize, Result<(), String>)> = tokio::task::JoinSet::new();
+
     for (idx, annotated) in row_results.into_iter().enumerate() {
         if !confirmed_set.contains(&idx) {
             skipped_duplicates += 1;
             continue;
         }
         match annotated.result {
-            RowResult::Valid(row) => match row_to_command(&row, user_id.value()) {
-                Ok(cmd) => match review_logger.log_review(cmd).await {
-                    Ok(_) => imported += 1,
-                    Err(e) => failed.push((idx, e.to_string())),
-                },
-                Err(e) => failed.push((idx, e)),
-            },
             RowResult::Invalid { errors, .. } => {
                 failed.push((idx, errors.join("; ")));
             }
+            RowResult::Valid(row) => match row_to_command(&row, user_id.value()) {
+                Err(e) => failed.push((idx, e)),
+                Ok(log_cmd) => {
+                    let permit = Arc::clone(&semaphore).acquire_owned().await.unwrap();
+                    let logger = Arc::clone(&review_logger);
+                    tasks.spawn(async move {
+                        let result = logger.log_review(log_cmd).await.map_err(|e| e.to_string());
+                        drop(permit);
+                        (idx, result)
+                    });
+                }
+            },
+        }
+    }
+
+    while let Some(res) = tasks.join_next().await {
+        let (idx, outcome) = res.expect("import task panicked");
+        match outcome {
+            Ok(()) => imported += 1,
+            Err(e) => failed.push((idx, e)),
         }
     }
 

crates/application/src/import/tests/apply_profile.rs

@@ -78,11 +78,7 @@ async fn applies_profile_mappings_to_session() {
     let profile_id = profile.id.clone();
     profiles.save(&profile).await.unwrap();
 
-    let session = domain::models::ImportSession::new(
-        domain::value_objects::ImportSessionId::generate(),
-        UserId::from_uuid(user_id),
-        Utc::now().naive_utc(),
-    );
+    let session = domain::models::ImportSession::new(UserId::from_uuid(user_id));
     let session_id = session.id.clone();
     sessions.create(&session).await.unwrap();
 

crates/application/src/import/tests/execute.rs

@@ -1,19 +1,17 @@
 use std::sync::Arc;
 
-use chrono::Utc;
 use domain::models::{AnnotatedRow, ImportSession, import::RowResult};
 use domain::ports::ImportSessionRepository;
 use domain::testing::InMemoryImportSessionRepository;
-use domain::value_objects::{ImportSessionId, UserId};
+use domain::value_objects::UserId;
 use uuid::Uuid;
 
 use crate::import::commands::ExecuteImportCommand;
 use crate::import::execute;
 use crate::test_helpers::NoopReviewLogger;
 
-fn make_session_with_rows(user_id: UserId, session_id: ImportSessionId) -> ImportSession {
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(session_id, user_id, now);
+fn make_session_with_rows(user_id: UserId) -> ImportSession {
+    let mut session = ImportSession::new(user_id);
     session.row_results = Some(vec![
         AnnotatedRow {
             result: RowResult::Valid(domain::models::ImportRow {
@@ -47,9 +45,9 @@ fn make_session_with_rows(user_id: UserId, session_id: ImportSessionId) -> Impor
 async fn imports_confirmed_rows() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let session = make_session_with_rows(UserId::from_uuid(uid), sid.clone());
+    let session = make_session_with_rows(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     sessions.create(&session).await.unwrap();
 
     let result = execute::execute(
@@ -73,9 +71,9 @@ async fn imports_confirmed_rows() {
 async fn skips_unconfirmed_rows() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let session = make_session_with_rows(UserId::from_uuid(uid), sid.clone());
+    let session = make_session_with_rows(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     sessions.create(&session).await.unwrap();
 
     let result = execute::execute(
@@ -116,10 +114,9 @@ async fn fails_when_session_not_found() {
 async fn handles_datetime_format() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("DateTime Movie".into()),
@@ -154,10 +151,9 @@ async fn handles_datetime_format() {
 async fn fails_on_invalid_rating() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("Bad Rating Movie".into()),
@@ -192,10 +188,9 @@ async fn fails_on_invalid_rating() {
 async fn fails_on_missing_watched_at() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("No Date Movie".into()),
@@ -230,10 +225,9 @@ async fn fails_on_missing_watched_at() {
 async fn imports_row_with_external_metadata_id() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("TMDB Movie".into()),
@@ -268,10 +262,9 @@ async fn imports_row_with_external_metadata_id() {
 async fn imports_row_with_director_and_comment() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("Directed Movie".into()),
@@ -306,10 +299,9 @@ async fn imports_row_with_director_and_comment() {
 async fn handles_space_separated_datetime_format() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("Space DateTime".into()),
@@ -344,10 +336,9 @@ async fn handles_space_separated_datetime_format() {
 async fn reports_invalid_row_result_errors() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Invalid {
             errors: vec!["missing title".into(), "bad year".into()],
@@ -379,10 +370,9 @@ async fn reports_invalid_row_result_errors() {
 async fn fails_on_missing_rating() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("No Rating Movie".into()),
@@ -418,10 +408,9 @@ async fn fails_on_missing_rating() {
 async fn fails_on_unparseable_date() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("Bad Date Movie".into()),
@@ -457,10 +446,9 @@ async fn fails_on_unparseable_date() {
 async fn imports_row_without_release_year() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let now = Utc::now().naive_utc();
-    let mut session = ImportSession::new(sid.clone(), UserId::from_uuid(uid), now);
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     session.row_results = Some(vec![AnnotatedRow {
         result: RowResult::Valid(domain::models::ImportRow {
             title: Some("No Year Movie".into()),
@@ -495,9 +483,9 @@ async fn imports_row_without_release_year() {
 async fn deletes_session_after_import() {
     let sessions = InMemoryImportSessionRepository::new();
     let uid = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let session = make_session_with_rows(UserId::from_uuid(uid), sid.clone());
+    let session = make_session_with_rows(UserId::from_uuid(uid));
+    let sid = session.id.clone();
     sessions.create(&session).await.unwrap();
     assert_eq!(sessions.count(), 1);
 
@@ -519,3 +507,46 @@ async fn deletes_session_after_import() {
         "session should be deleted after import"
     );
 }
+
+#[tokio::test]
+async fn imports_more_rows_than_concurrency_limit() {
+    let sessions = InMemoryImportSessionRepository::new();
+    let uid = Uuid::new_v4();
+
+    let rows: Vec<_> = (0..15)
+        .map(|i| AnnotatedRow {
+            result: RowResult::Valid(domain::models::ImportRow {
+                title: Some(format!("Movie {i}")),
+                release_year: Some("2024".into()),
+                rating: Some("4".into()),
+                watched_at: Some("2024-06-01".into()),
+                external_metadata_id: None,
+                director: None,
+                comment: None,
+            }),
+            is_duplicate: false,
+        })
+        .collect();
+
+    let mut session = ImportSession::new(UserId::from_uuid(uid));
+    let sid = session.id.clone();
+    session.row_results = Some(rows);
+    sessions.create(&session).await.unwrap();
+
+    let confirmed_indices: Vec<usize> = (0..15).collect();
+    let result = execute::execute(
+        Arc::clone(&sessions) as _,
+        Arc::new(NoopReviewLogger),
+        ExecuteImportCommand {
+            user_id: uid,
+            session_id: sid.value(),
+            confirmed_indices,
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(result.imported, 15);
+    assert_eq!(result.skipped_duplicates, 0);
+    assert!(result.failed.is_empty());
+}

crates/application/src/import/tests/save_profile.rs

@@ -1,10 +1,9 @@
 use std::sync::Arc;
 
-use chrono::Utc;
 use domain::models::ImportSession;
 use domain::ports::ImportSessionRepository;
 use domain::testing::{InMemoryImportProfileRepository, InMemoryImportSessionRepository};
-use domain::value_objects::{ImportSessionId, UserId};
+use domain::value_objects::UserId;
 use uuid::Uuid;
 
 use crate::import::{commands::SaveImportProfileCommand, save_profile};
@@ -33,13 +32,9 @@ async fn saves_profile_from_session() {
     let sessions = InMemoryImportSessionRepository::new();
     let profiles = InMemoryImportProfileRepository::new();
     let user_id = Uuid::new_v4();
-    let sid = ImportSessionId::generate();
 
-    let mut session = ImportSession::new(
-        sid.clone(),
-        UserId::from_uuid(user_id),
-        Utc::now().naive_utc(),
-    );
+    let mut session = ImportSession::new(UserId::from_uuid(user_id));
+    let sid = session.id.clone();
     session.field_mappings = Some(vec![]);
     sessions.create(&session).await.unwrap();
 

crates/application/src/jobs/enrichment_staleness.rs

@@ -6,6 +6,7 @@ use domain::{
     errors::DomainError,
     events::DomainEvent,
     ports::{EventPublisher, MovieProfileRepository, PeriodicJob},
+    value_objects::ExternalMetadataId,
 };
 
 pub struct EnrichmentStalenessJob {
@@ -38,9 +39,16 @@ impl PeriodicJob for EnrichmentStalenessJob {
         }
         tracing::info!("enrichment scan: {} stale movies", stale.len());
         for (movie_id, external_metadata_id) in stale {
+            let ext_id = match ExternalMetadataId::new(external_metadata_id) {
+                Ok(id) => id,
+                Err(e) => {
+                    tracing::warn!("skipping stale movie with malformed external_metadata_id: {e}");
+                    continue;
+                }
+            };
             let event = DomainEvent::MovieEnrichmentRequested {
                 movie_id,
-                external_metadata_id,
+                external_metadata_id: ext_id,
             };
             self.event_publisher.publish(&event).await?;
         }

crates/application/src/person/get.rs

@@ -18,7 +18,7 @@ pub async fn execute(deps: &GetPersonDeps, id: PersonId) -> Result<Option<Person
             .event_publisher
             .publish(&DomainEvent::PersonEnrichmentRequested {
                 person_id: id,
-                external_person_id: p.external_id().value().to_string(),
+                external_person_id: p.external_id().clone(),
             })
             .await;
     }

crates/application/src/person/get_credits.rs

@@ -16,7 +16,7 @@ pub async fn execute(deps: &GetPersonDeps, id: PersonId) -> Result<PersonCredits
             .event_publisher
             .publish(&DomainEvent::PersonEnrichmentRequested {
                 person_id: id,
-                external_person_id: credits.person.external_id().value().to_string(),
+                external_person_id: credits.person.external_id().clone(),
             })
             .await;
     }

crates/application/src/tests/worker.rs

@@ -20,6 +20,23 @@ impl AckHandle for NoopAck {
     }
 }
 
+struct TrackingAck {
+    acked: Arc<Mutex<bool>>,
+    nacked: Arc<Mutex<bool>>,
+}
+
+#[async_trait]
+impl AckHandle for TrackingAck {
+    async fn ack(&self) -> Result<(), DomainError> {
+        *self.acked.lock().unwrap() = true;
+        Ok(())
+    }
+    async fn nack(&self) -> Result<(), DomainError> {
+        *self.nacked.lock().unwrap() = true;
+        Ok(())
+    }
+}
+
 struct VecConsumer {
     events: Vec<DomainEvent>,
 }
@@ -97,85 +114,15 @@ async fn dispatches_to_all_handlers() {
     assert_eq!(*calls.lock().unwrap(), vec!["movie_discovered"]);
 }
 
-#[tokio::test]
-async fn acks_even_when_handler_fails() {
-    let ack_called = Arc::new(Mutex::new(false));
-
-    struct TrackingAck {
-        ack_called: Arc<Mutex<bool>>,
-    }
-
-    #[async_trait]
-    impl AckHandle for TrackingAck {
-        async fn ack(&self) -> Result<(), DomainError> {
-            *self.ack_called.lock().unwrap() = true;
-            Ok(())
-        }
-        async fn nack(&self) -> Result<(), DomainError> {
-            Ok(())
-        }
-    }
-
-    struct TrackingConsumer {
-        event: DomainEvent,
-        ack_called: Arc<Mutex<bool>>,
-    }
-
-    impl EventConsumer for TrackingConsumer {
-        fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
-            let envelope = EventEnvelope::new(
-                self.event.clone(),
-                Box::new(TrackingAck {
-                    ack_called: Arc::clone(&self.ack_called),
-                }),
-            );
-            Box::pin(stream::iter(vec![Ok(envelope)]))
-        }
-    }
-
-    struct FailingHandler;
-
-    #[async_trait]
-    impl EventHandler for FailingHandler {
-        async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
-            Err(DomainError::InfrastructureError("boom".into()))
-        }
-    }
-
-    let consumer = TrackingConsumer {
-        event: movie_discovered(),
-        ack_called: Arc::clone(&ack_called),
-    };
-
-    WorkerService::new(Arc::new(consumer), vec![Arc::new(FailingHandler)])
-        .run(tokio::sync::watch::channel(false).1)
-        .await;
-
-    assert!(*ack_called.lock().unwrap());
-}
-
 #[tokio::test]
 async fn acks_when_all_handlers_succeed() {
-    let ack_called = Arc::new(Mutex::new(false));
-
-    struct TrackingAck {
-        ack_called: Arc<Mutex<bool>>,
-    }
-
-    #[async_trait]
-    impl AckHandle for TrackingAck {
-        async fn ack(&self) -> Result<(), DomainError> {
-            *self.ack_called.lock().unwrap() = true;
-            Ok(())
-        }
-        async fn nack(&self) -> Result<(), DomainError> {
-            Ok(())
-        }
-    }
+    let acked = Arc::new(Mutex::new(false));
+    let nacked = Arc::new(Mutex::new(false));
 
     struct TrackingConsumer {
         event: DomainEvent,
-        ack_called: Arc<Mutex<bool>>,
+        acked: Arc<Mutex<bool>>,
+        nacked: Arc<Mutex<bool>>,
     }
 
     impl EventConsumer for TrackingConsumer {
@@ -183,7 +130,8 @@ async fn acks_when_all_handlers_succeed() {
             let envelope = EventEnvelope::new(
                 self.event.clone(),
                 Box::new(TrackingAck {
-                    ack_called: Arc::clone(&self.ack_called),
+                    acked: Arc::clone(&self.acked),
+                    nacked: Arc::clone(&self.nacked),
                 }),
             );
             Box::pin(stream::iter(vec![Ok(envelope)]))
@@ -192,12 +140,165 @@ async fn acks_when_all_handlers_succeed() {
 
     let consumer = TrackingConsumer {
         event: movie_discovered(),
-        ack_called: Arc::clone(&ack_called),
+        acked: Arc::clone(&acked),
+        nacked: Arc::clone(&nacked),
     };
 
     WorkerService::new(Arc::new(consumer), vec![])
         .run(tokio::sync::watch::channel(false).1)
         .await;
 
-    assert!(*ack_called.lock().unwrap());
+    assert!(*acked.lock().unwrap());
+    assert!(!*nacked.lock().unwrap());
+}
+
+#[tokio::test]
+async fn nacks_on_transient_handler_failure() {
+    let acked = Arc::new(Mutex::new(false));
+    let nacked = Arc::new(Mutex::new(false));
+
+    struct TrackingConsumer {
+        event: DomainEvent,
+        acked: Arc<Mutex<bool>>,
+        nacked: Arc<Mutex<bool>>,
+    }
+    impl EventConsumer for TrackingConsumer {
+        fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+            let envelope = EventEnvelope::new(
+                self.event.clone(),
+                Box::new(TrackingAck {
+                    acked: Arc::clone(&self.acked),
+                    nacked: Arc::clone(&self.nacked),
+                }),
+            );
+            Box::pin(stream::iter(vec![Ok(envelope)]))
+        }
+    }
+    struct TransientHandler;
+    #[async_trait]
+    impl EventHandler for TransientHandler {
+        async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
+            Err(DomainError::InfrastructureError("network timeout".into()))
+        }
+    }
+
+    WorkerService::new(
+        Arc::new(TrackingConsumer {
+            event: movie_discovered(),
+            acked: Arc::clone(&acked),
+            nacked: Arc::clone(&nacked),
+        }),
+        vec![Arc::new(TransientHandler)],
+    )
+    .run(tokio::sync::watch::channel(false).1)
+    .await;
+
+    assert!(!*acked.lock().unwrap(), "should not ack on transient error");
+    assert!(*nacked.lock().unwrap(), "should nack on transient error");
+}
+
+#[tokio::test]
+async fn acks_on_permanent_handler_failure() {
+    let acked = Arc::new(Mutex::new(false));
+    let nacked = Arc::new(Mutex::new(false));
+
+    struct TrackingConsumer {
+        event: DomainEvent,
+        acked: Arc<Mutex<bool>>,
+        nacked: Arc<Mutex<bool>>,
+    }
+    impl EventConsumer for TrackingConsumer {
+        fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+            let envelope = EventEnvelope::new(
+                self.event.clone(),
+                Box::new(TrackingAck {
+                    acked: Arc::clone(&self.acked),
+                    nacked: Arc::clone(&self.nacked),
+                }),
+            );
+            Box::pin(stream::iter(vec![Ok(envelope)]))
+        }
+    }
+    struct PermanentHandler;
+    #[async_trait]
+    impl EventHandler for PermanentHandler {
+        async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
+            Err(DomainError::NotFound("movie not on tmdb".into()))
+        }
+    }
+
+    WorkerService::new(
+        Arc::new(TrackingConsumer {
+            event: movie_discovered(),
+            acked: Arc::clone(&acked),
+            nacked: Arc::clone(&nacked),
+        }),
+        vec![Arc::new(PermanentHandler)],
+    )
+    .run(tokio::sync::watch::channel(false).1)
+    .await;
+
+    assert!(*acked.lock().unwrap(), "should ack on permanent error");
+    assert!(
+        !*nacked.lock().unwrap(),
+        "should not nack on permanent error"
+    );
+}
+
+#[tokio::test]
+async fn nacks_if_any_handler_is_transient() {
+    let acked = Arc::new(Mutex::new(false));
+    let nacked = Arc::new(Mutex::new(false));
+
+    struct TrackingConsumer {
+        event: DomainEvent,
+        acked: Arc<Mutex<bool>>,
+        nacked: Arc<Mutex<bool>>,
+    }
+    impl EventConsumer for TrackingConsumer {
+        fn consume(&self) -> BoxStream<'_, Result<EventEnvelope, DomainError>> {
+            let envelope = EventEnvelope::new(
+                self.event.clone(),
+                Box::new(TrackingAck {
+                    acked: Arc::clone(&self.acked),
+                    nacked: Arc::clone(&self.nacked),
+                }),
+            );
+            Box::pin(stream::iter(vec![Ok(envelope)]))
+        }
+    }
+    struct OkHandler;
+    #[async_trait]
+    impl EventHandler for OkHandler {
+        async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
+            Ok(())
+        }
+    }
+    struct TransientHandler;
+    #[async_trait]
+    impl EventHandler for TransientHandler {
+        async fn handle(&self, _: &DomainEvent) -> Result<(), DomainError> {
+            Err(DomainError::InfrastructureError("db gone".into()))
+        }
+    }
+
+    WorkerService::new(
+        Arc::new(TrackingConsumer {
+            event: movie_discovered(),
+            acked: Arc::clone(&acked),
+            nacked: Arc::clone(&nacked),
+        }),
+        vec![Arc::new(OkHandler), Arc::new(TransientHandler)],
+    )
+    .run(tokio::sync::watch::channel(false).1)
+    .await;
+
+    assert!(
+        !*acked.lock().unwrap(),
+        "should not ack when any handler is transient"
+    );
+    assert!(
+        *nacked.lock().unwrap(),
+        "should nack when any handler is transient"
+    );
 }

crates/application/src/users/tests/get_settings.rs

@@ -11,5 +11,7 @@ async fn returns_default_settings() {
         .await
         .unwrap();
 
-    assert!(!settings.federate_goals());
+    assert!(settings.federate_goals());
+    assert!(settings.federate_reviews());
+    assert!(settings.federate_watchlist());
 }

crates/application/src/users/tests/update_settings.rs

@@ -13,7 +13,31 @@ async fn updates_federate_goals() {
     let settings_repo = InMemoryUserSettingsRepository::new();
     let b = TestContextBuilder::new().with_user_settings(Arc::clone(&settings_repo) as _);
     let user_settings = b.user_settings_repo.clone();
+    let uid = Uuid::nil();
 
+    crate::users::update_settings::execute(
+        user_settings.clone(),
+        UpdateUserSettingsCommand {
+            user_id: uid,
+            federate_goals: false,
+            federate_reviews: true,
+            federate_watchlist: true,
+        },
+    )
+    .await
+    .unwrap();
+
+    let settings = get_settings::execute(user_settings, uid).await.unwrap();
+    assert!(!settings.federate_goals());
+    assert!(settings.federate_reviews());
+    assert!(settings.federate_watchlist());
+}
+
+#[tokio::test]
+async fn updates_federate_reviews() {
+    let settings_repo = InMemoryUserSettingsRepository::new();
+    let b = TestContextBuilder::new().with_user_settings(Arc::clone(&settings_repo) as _);
+    let user_settings = b.user_settings_repo.clone();
     let uid = Uuid::nil();
 
     crate::users::update_settings::execute(
@@ -21,6 +45,8 @@ async fn updates_federate_goals() {
         UpdateUserSettingsCommand {
             user_id: uid,
             federate_goals: true,
+            federate_reviews: false,
+            federate_watchlist: true,
         },
     )
     .await
@@ -28,4 +54,31 @@ async fn updates_federate_goals() {
 
     let settings = get_settings::execute(user_settings, uid).await.unwrap();
     assert!(settings.federate_goals());
+    assert!(!settings.federate_reviews());
+    assert!(settings.federate_watchlist());
+}
+
+#[tokio::test]
+async fn updates_federate_watchlist() {
+    let settings_repo = InMemoryUserSettingsRepository::new();
+    let b = TestContextBuilder::new().with_user_settings(Arc::clone(&settings_repo) as _);
+    let user_settings = b.user_settings_repo.clone();
+    let uid = Uuid::nil();
+
+    crate::users::update_settings::execute(
+        user_settings.clone(),
+        UpdateUserSettingsCommand {
+            user_id: uid,
+            federate_goals: true,
+            federate_reviews: true,
+            federate_watchlist: false,
+        },
+    )
+    .await
+    .unwrap();
+
+    let settings = get_settings::execute(user_settings, uid).await.unwrap();
+    assert!(settings.federate_goals());
+    assert!(settings.federate_reviews());
+    assert!(!settings.federate_watchlist());
 }

crates/application/src/users/update_settings.rs

@@ -5,6 +5,8 @@ use domain::{errors::DomainError, ports::UserSettingsRepository, value_objects::
 pub struct UpdateUserSettingsCommand {
     pub user_id: uuid::Uuid,
     pub federate_goals: bool,
+    pub federate_reviews: bool,
+    pub federate_watchlist: bool,
 }
 
 pub async fn execute(
@@ -14,6 +16,8 @@ pub async fn execute(
     let uid = UserId::from_uuid(cmd.user_id);
     let mut settings = user_settings.get(&uid).await?;
     settings.set_federate_goals(cmd.federate_goals);
+    settings.set_federate_reviews(cmd.federate_reviews);
+    settings.set_federate_watchlist(cmd.federate_watchlist);
     user_settings.save(&settings).await
 }
 

crates/application/src/worker.rs

@@ -65,12 +65,22 @@ impl WorkerService {
 }
 
 async fn dispatch(handlers: Arc<Vec<Arc<dyn EventHandler>>>, envelope: EventEnvelope) {
+    let mut any_transient = false;
     for handler in handlers.iter() {
         if let Err(e) = handler.handle(&envelope.event).await {
-            tracing::warn!("event handler error (non-fatal): {e}");
+            if e.is_transient() {
+                tracing::warn!("transient handler error, will retry: {e}");
+                any_transient = true;
+            } else {
+                tracing::warn!("permanent handler error (not retrying): {e}");
+            }
         }
     }
-    if let Err(e) = envelope.ack().await {
+    if any_transient {
+        if let Err(e) = envelope.nack().await {
+            tracing::error!("nack failed: {e}");
+        }
+    } else if let Err(e) = envelope.ack().await {
         tracing::error!("ack failed: {e}");
     }
 }

crates/domain/src/errors.rs

@@ -20,3 +20,44 @@ pub enum DomainError {
     #[error("Forbidden: {0}")]
     Forbidden(String),
 }
+
+impl DomainError {
+    pub fn is_transient(&self) -> bool {
+        matches!(self, DomainError::InfrastructureError(_))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn infrastructure_error_is_transient() {
+        assert!(DomainError::InfrastructureError("network timeout".into()).is_transient());
+    }
+
+    #[test]
+    fn not_found_is_not_transient() {
+        assert!(!DomainError::NotFound("thing".into()).is_transient());
+    }
+
+    #[test]
+    fn validation_error_is_not_transient() {
+        assert!(!DomainError::ValidationError("bad input".into()).is_transient());
+    }
+
+    #[test]
+    fn unauthorized_is_not_transient() {
+        assert!(!DomainError::Unauthorized("token expired".into()).is_transient());
+    }
+
+    #[test]
+    fn forbidden_is_not_transient() {
+        assert!(!DomainError::Forbidden("no access".into()).is_transient());
+    }
+
+    #[test]
+    fn invalid_rating_is_not_transient() {
+        assert!(!DomainError::InvalidRating { max: 5, given: 9 }.is_transient());
+    }
+}

crates/domain/src/events.rs

@@ -3,7 +3,7 @@ use chrono::NaiveDateTime;
 
 use crate::{
     errors::DomainError,
-    models::PersonId,
+    models::{ExternalPersonId, PersonId},
     value_objects::{
         ExternalMetadataId, GoalId, MovieId, PosterPath, Rating, ReviewId, UserId, WrapUpId,
     },
@@ -42,11 +42,11 @@ pub enum DomainEvent {
     },
     MovieEnrichmentRequested {
         movie_id: MovieId,
-        external_metadata_id: String,
+        external_metadata_id: ExternalMetadataId,
     },
     PersonEnrichmentRequested {
         person_id: PersonId,
-        external_person_id: String,
+        external_person_id: ExternalPersonId,
     },
     ImageStored {
         key: String,

crates/domain/src/models/import_session.rs

@@ -15,11 +15,22 @@ pub struct ImportSession {
     pub expires_at: NaiveDateTime,
 }
 
+pub struct PersistedImportSession {
+    pub id: ImportSessionId,
+    pub user_id: UserId,
+    pub parsed_file: Option<ParsedFile>,
+    pub field_mappings: Option<Vec<FieldMapping>>,
+    pub row_results: Option<Vec<AnnotatedRow>>,
+    pub created_at: NaiveDateTime,
+    pub expires_at: NaiveDateTime,
+}
+
 impl ImportSession {
-    pub fn new(id: ImportSessionId, user_id: UserId, created_at: NaiveDateTime) -> Self {
+    pub fn new(user_id: UserId) -> Self {
+        let created_at = chrono::Utc::now().naive_utc();
         let expires_at = created_at + chrono::Duration::hours(24);
         Self {
-            id,
+            id: ImportSessionId::generate(),
             user_id,
             parsed_file: None,
             field_mappings: None,
@@ -28,4 +39,16 @@ impl ImportSession {
             expires_at,
         }
     }
+
+    pub fn from_persistence(p: PersistedImportSession) -> Self {
+        Self {
+            id: p.id,
+            user_id: p.user_id,
+            parsed_file: p.parsed_file,
+            field_mappings: p.field_mappings,
+            row_results: p.row_results,
+            created_at: p.created_at,
+            expires_at: p.expires_at,
+        }
+    }
 }

crates/domain/src/models/user_settings.rs

@@ -4,27 +4,34 @@ use crate::value_objects::UserId;
 pub struct UserSettings {
     user_id: UserId,
     federate_goals: bool,
+    federate_reviews: bool,
+    federate_watchlist: bool,
 }
 
 impl UserSettings {
     pub fn new(user_id: UserId) -> Self {
         Self {
             user_id,
-            federate_goals: false,
+            federate_goals: true,
+            federate_reviews: true,
+            federate_watchlist: true,
         }
     }
 
-    pub fn from_persistence(user_id: UserId, federate_goals: bool) -> Self {
+    pub fn from_persistence(
+        user_id: UserId,
+        federate_goals: bool,
+        federate_reviews: bool,
+        federate_watchlist: bool,
+    ) -> Self {
         Self {
             user_id,
             federate_goals,
+            federate_reviews,
+            federate_watchlist,
         }
     }
 
-    pub fn set_federate_goals(&mut self, value: bool) {
-        self.federate_goals = value;
-    }
-
     pub fn user_id(&self) -> &UserId {
         &self.user_id
     }
@@ -32,4 +39,24 @@ impl UserSettings {
     pub fn federate_goals(&self) -> bool {
         self.federate_goals
     }
+
+    pub fn set_federate_goals(&mut self, value: bool) {
+        self.federate_goals = value;
+    }
+
+    pub fn federate_reviews(&self) -> bool {
+        self.federate_reviews
+    }
+
+    pub fn set_federate_reviews(&mut self, value: bool) {
+        self.federate_reviews = value;
+    }
+
+    pub fn federate_watchlist(&self) -> bool {
+        self.federate_watchlist
+    }
+
+    pub fn set_federate_watchlist(&mut self, value: bool) {
+        self.federate_watchlist = value;
+    }
 }

crates/domain/src/ports.rs

@@ -144,6 +144,10 @@ pub trait DiaryRepository: Send + Sync {
     ) -> Result<Paginated<FeedEntry>, DomainError>;
     async fn get_review_history(&self, movie_id: &MovieId) -> Result<ReviewHistory, DomainError>;
     async fn get_user_history(&self, user_id: &UserId) -> Result<Vec<DiaryEntry>, DomainError>;
+    fn stream_user_history(
+        &self,
+        user_id: UserId,
+    ) -> futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>>;
     async fn get_movie_stats(&self, movie_id: &MovieId) -> Result<MovieStats, DomainError>;
     async fn get_movie_social_feed(
         &self,
@@ -253,13 +257,12 @@ pub trait PasswordHasher: Send + Sync {
     async fn verify(&self, plain_password: &str, hash: &PasswordHash) -> Result<bool, DomainError>;
 }
 
-#[async_trait]
 pub trait DiaryExporter: Send + Sync {
-    async fn serialize_entries(
+    fn stream_entries(
         &self,
-        entries: &[DiaryEntry],
+        stream: futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>>,
         format: ExportFormat,
-    ) -> Result<Vec<u8>, DomainError>;
+    ) -> futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>;
 }
 
 #[async_trait]
@@ -453,6 +456,17 @@ pub trait UserSettingsRepository: Send + Sync {
     async fn save(&self, settings: &UserSettings) -> Result<(), DomainError>;
 }
 
+pub struct FederationFlags {
+    pub goals: bool,
+    pub reviews: bool,
+    pub watchlist: bool,
+}
+
+#[async_trait]
+pub trait UserFederationSettingsQuery: Send + Sync {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError>;
+}
+
 #[async_trait]
 pub trait RemoteGoalRepository: Send + Sync {
     async fn save(&self, entry: RemoteGoalEntry) -> Result<(), DomainError>;
@@ -499,8 +513,6 @@ pub trait LocalApContentQuery: Send + Sync {
         limit: usize,
     ) -> Result<Vec<DiaryEntry>, DomainError>;
 
-    async fn get_user_federate_goals(&self, user_id: &UserId) -> Result<bool, DomainError>;
-
     async fn get_goal_with_progress(
         &self,
         user_id: &UserId,

crates/domain/src/testing/fakes.rs

@@ -154,6 +154,13 @@ impl DiaryRepository for FakeDiaryRepository {
         Ok(vec![])
     }
 
+    fn stream_user_history(
+        &self,
+        _user_id: UserId,
+    ) -> futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        Box::pin(futures::stream::empty())
+    }
+
     async fn get_movie_stats(&self, _movie_id: &MovieId) -> Result<MovieStats, DomainError> {
         Ok(MovieStats {
             total_count: 0,

crates/domain/src/testing/in_memory.rs

@@ -18,10 +18,10 @@ use crate::{
         collections::{PageParams, Paginated},
     },
     ports::{
-        GoalRepository, ImportProfileRepository, ImportSessionRepository, MovieProfileRepository,
-        MovieRepository, RefreshSessionRepository, ReviewRepository, UserProfileFieldsRepository,
-        UserRepository, UserSettingsRepository, WatchEventRepository, WatchlistRepository,
-        WebhookTokenRepository,
+        FederationFlags, GoalRepository, ImportProfileRepository, ImportSessionRepository,
+        MovieProfileRepository, MovieRepository, RefreshSessionRepository, ReviewRepository,
+        UserFederationSettingsQuery, UserProfileFieldsRepository, UserRepository,
+        UserSettingsRepository, WatchEventRepository, WatchlistRepository, WebhookTokenRepository,
     },
     value_objects::{
         Email, ExternalMetadataId, GoalId, ImportProfileId, ImportSessionId, MovieId, MovieTitle,
@@ -441,6 +441,22 @@ impl UserSettingsRepository for InMemoryUserSettingsRepository {
     }
 }
 
+#[async_trait]
+impl UserFederationSettingsQuery for InMemoryUserSettingsRepository {
+    async fn get_federation_flags(&self, user_id: &UserId) -> Result<FederationFlags, DomainError> {
+        let store = self.store.lock().unwrap();
+        let settings = store
+            .get(&user_id.value())
+            .cloned()
+            .unwrap_or_else(|| UserSettings::new(user_id.clone()));
+        Ok(FederationFlags {
+            goals: settings.federate_goals(),
+            reviews: settings.federate_reviews(),
+            watchlist: settings.federate_watchlist(),
+        })
+    }
+}
+
 // ── InMemoryWebhookTokenRepository ──────────────────────────────────────────
 
 pub struct InMemoryWebhookTokenRepository {

crates/domain/src/testing/panics.rs

@@ -49,6 +49,12 @@ impl DiaryRepository for PanicDiaryRepository {
     async fn get_user_history(&self, _: &UserId) -> Result<Vec<DiaryEntry>, DomainError> {
         panic!("PanicDiaryRepository called")
     }
+    fn stream_user_history(
+        &self,
+        _: UserId,
+    ) -> futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        panic!("PanicDiaryRepository called")
+    }
     async fn get_movie_stats(&self, _: &MovieId) -> Result<MovieStats, DomainError> {
         panic!("PanicDiaryRepository called")
     }
@@ -250,13 +256,12 @@ impl PosterFetcherClient for PanicPosterFetcher {
 
 pub struct PanicDiaryExporter;
 
-#[async_trait]
 impl DiaryExporter for PanicDiaryExporter {
-    async fn serialize_entries(
+    fn stream_entries(
         &self,
-        _: &[DiaryEntry],
-        _: ExportFormat,
-    ) -> Result<Vec<u8>, DomainError> {
+        _stream: futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>>,
+        _format: ExportFormat,
+    ) -> futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>> {
         panic!("PanicDiaryExporter called")
     }
 }

crates/presentation/Cargo.toml

@@ -42,6 +42,7 @@ dotenvy = { workspace = true }
 uuid = { workspace = true }
 chrono = { workspace = true }
 async-trait = { workspace = true }
+futures = { workspace = true }
 
 api-types = { workspace = true }
 domain = { workspace = true, features = ["test-helpers"] }

crates/presentation/src/factory.rs

@@ -36,6 +36,7 @@ pub struct DatabaseOutput {
     pub wrapup_repo: Arc<dyn domain::ports::WrapUpRepository>,
     pub goal: Arc<dyn domain::ports::GoalRepository>,
     pub user_settings: Arc<dyn domain::ports::UserSettingsRepository>,
+    pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub remote_goal: Arc<dyn domain::ports::RemoteGoalRepository>,
     pub refresh_session: Arc<dyn RefreshSessionRepository>,
     pub db_pool: DbPool,
@@ -78,6 +79,7 @@ pub async fn build_database_adapters(backend: &str, url: &str) -> anyhow::Result
                 wrapup_repo: w.wrapup_repo,
                 goal: w.goal,
                 user_settings: w.user_settings,
+                federation_settings: w.federation_settings,
                 remote_goal: w.remote_goal,
                 refresh_session: Arc::new(postgres::PostgresRefreshSessionAdapter::new(
                     w.pool.clone(),
@@ -119,6 +121,7 @@ pub async fn build_database_adapters(backend: &str, url: &str) -> anyhow::Result
                 wrapup_repo: w.wrapup_repo,
                 goal: w.goal,
                 user_settings: w.user_settings,
+                federation_settings: w.federation_settings,
                 remote_goal: w.remote_goal,
                 refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone()))
                     as _,

crates/presentation/src/handlers/diary.rs

@@ -1,9 +1,11 @@
 use axum::{
     Form, Json,
+    body::Body,
     extract::{Extension, Path, Query, State},
     http::StatusCode,
     response::{IntoResponse, Redirect},
 };
+use futures::StreamExt;
 use uuid::Uuid;
 
 use application::diary::{
@@ -147,30 +149,29 @@ pub async fn export_diary(
         user_id: user.0.value(),
         format,
     };
-    match export_diary_uc::execute(
+    let stream = export_diary_uc::execute(
         &state.app_ctx.repos.diary,
         &state.app_ctx.services.diary_exporter,
         query,
-    )
-    .await
-    {
-        Ok(bytes) => (
-            StatusCode::OK,
-            [
-                (axum::http::header::CONTENT_TYPE, content_type.to_string()),
-                (
-                    axum::http::header::CONTENT_DISPOSITION,
-                    format!("attachment; filename=\"{}\"", filename),
-                ),
-            ],
-            bytes,
-        )
-            .into_response(),
-        Err(e) => {
-            tracing::error!("export error: {:?}", e);
-            StatusCode::INTERNAL_SERVER_ERROR.into_response()
+    );
+    let stream = stream.map(|r| {
+        if let Err(ref e) = r {
+            tracing::error!("diary export stream error: {e}");
         }
-    }
+        r
+    });
+    (
+        StatusCode::OK,
+        [
+            (axum::http::header::CONTENT_TYPE, content_type.to_string()),
+            (
+                axum::http::header::CONTENT_DISPOSITION,
+                format!("attachment; filename=\"{}\"", filename),
+            ),
+        ],
+        Body::from_stream(stream),
+    )
+        .into_response()
 }
 
 #[utoipa::path(
@@ -314,27 +315,29 @@ pub async fn get_export_html(
         user_id: user_id.value(),
         format,
     };
-    match export_diary_uc::execute(
+    let stream = export_diary_uc::execute(
         &state.app_ctx.repos.diary,
         &state.app_ctx.services.diary_exporter,
         query,
+    );
+    let stream = stream.map(|r| {
+        if let Err(ref e) = r {
+            tracing::error!("diary export stream error: {e}");
+        }
+        r
+    });
+    (
+        StatusCode::OK,
+        [
+            (axum::http::header::CONTENT_TYPE, content_type.to_string()),
+            (
+                axum::http::header::CONTENT_DISPOSITION,
+                format!("attachment; filename=\"{}\"", filename),
+            ),
+        ],
+        Body::from_stream(stream),
     )
-    .await
-    {
-        Ok(bytes) => (
-            StatusCode::OK,
-            [
-                (axum::http::header::CONTENT_TYPE, content_type.to_string()),
-                (
-                    axum::http::header::CONTENT_DISPOSITION,
-                    format!("attachment; filename=\"{}\"", filename),
-                ),
-            ],
-            bytes,
-        )
-            .into_response(),
-        Err(e) => crate::errors::domain_error_response(e),
-    }
+        .into_response()
 }
 
 pub async fn get_activity_feed_html(

crates/presentation/src/handlers/goals.rs

@@ -176,6 +176,8 @@ pub async fn get_settings(
     .await?;
     Ok(Json(UserSettingsDto {
         federate_goals: settings.federate_goals(),
+        federate_reviews: settings.federate_reviews(),
+        federate_watchlist: settings.federate_watchlist(),
     }))
 }
 
@@ -198,6 +200,8 @@ pub async fn update_settings(
         application::users::update_settings::UpdateUserSettingsCommand {
             user_id: user.0.value(),
             federate_goals: req.federate_goals,
+            federate_reviews: req.federate_reviews,
+            federate_watchlist: req.federate_watchlist,
         },
     )
     .await?;

crates/presentation/src/main.rs

@@ -119,6 +119,7 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
             remote_goal_repo: Arc::clone(&db.remote_goal),
             local_ap_content: Arc::clone(&ap_content_repo),
             user_repo: Arc::clone(&db.user),
+            federation_settings: std::sync::Arc::clone(&db.federation_settings),
             base_url: app_config.base_url.clone(),
             allow_registration: app_config.allow_registration,
             event_publisher: Arc::clone(&ep),

crates/presentation/src/tests/extractors.rs

@@ -120,6 +120,12 @@ impl DiaryRepository for Panic {
     async fn get_user_history(&self, _: &UserId) -> Result<Vec<DiaryEntry>, DomainError> {
         panic!()
     }
+    fn stream_user_history(
+        &self,
+        _: UserId,
+    ) -> futures::stream::BoxStream<'static, Result<DiaryEntry, DomainError>> {
+        panic!()
+    }
     async fn get_movie_stats(
         &self,
         _: &MovieId,
@@ -379,14 +385,17 @@ impl domain::ports::MovieProfileRepository for Panic {
         Ok(vec![])
     }
 }
-#[async_trait::async_trait]
 impl domain::ports::DiaryExporter for Panic {
-    async fn serialize_entries(
+    fn stream_entries(
         &self,
-        _: &[domain::models::DiaryEntry],
-        _: domain::models::ExportFormat,
-    ) -> Result<Vec<u8>, domain::errors::DomainError> {
-        panic!()
+        _stream: futures::stream::BoxStream<
+            'static,
+            Result<domain::models::DiaryEntry, domain::errors::DomainError>,
+        >,
+        _format: domain::models::ExportFormat,
+    ) -> futures::stream::BoxStream<'static, Result<bytes::Bytes, domain::errors::DomainError>>
+    {
+        panic!("Panic DiaryExporter called")
     }
 }
 

crates/presentation/tests/api_test.rs

@@ -165,14 +165,16 @@ impl domain::ports::UserProfileFieldsRepository for PanicProfileFields {
 }
 
 struct PanicExporter;
-#[async_trait]
 impl domain::ports::DiaryExporter for PanicExporter {
-    async fn serialize_entries(
+    fn stream_entries(
         &self,
-        _: &[domain::models::DiaryEntry],
-        _: domain::models::ExportFormat,
-    ) -> Result<Vec<u8>, DomainError> {
-        panic!()
+        _stream: futures::stream::BoxStream<
+            'static,
+            Result<domain::models::DiaryEntry, DomainError>,
+        >,
+        _format: domain::models::ExportFormat,
+    ) -> futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>> {
+        panic!("PanicExporter::stream_entries")
     }
 }
 

crates/worker/src/db.rs

@@ -30,6 +30,7 @@ pub struct WorkerDbOutput {
     pub wrapup_repo: Arc<dyn domain::ports::WrapUpRepository>,
     pub remote_goal: Arc<dyn domain::ports::RemoteGoalRepository>,
     pub refresh_session: Arc<dyn domain::ports::RefreshSessionRepository>,
+    pub federation_settings: Arc<dyn domain::ports::UserFederationSettingsQuery>,
     pub db_pool: DbPool,
 }
 
@@ -64,6 +65,7 @@ pub async fn connect(database_url: &str, backend: &str) -> anyhow::Result<Worker
                 refresh_session: Arc::new(postgres::PostgresRefreshSessionAdapter::new(
                     w.pool.clone(),
                 )) as _,
+                federation_settings: w.federation_settings,
                 db_pool: DbPool::Postgres(w.pool),
             })
         }
@@ -95,6 +97,7 @@ pub async fn connect(database_url: &str, backend: &str) -> anyhow::Result<Worker
                 remote_goal: w.remote_goal,
                 refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone()))
                     as _,
+                federation_settings: w.federation_settings,
                 db_pool: DbPool::Sqlite(w.pool),
             })
         }

crates/worker/src/main.rs

@@ -241,6 +241,7 @@ async fn main() -> anyhow::Result<()> {
                 base_url,
                 allow_registration,
                 event_publisher: Arc::clone(&event_publisher),
+                federation_settings: std::sync::Arc::clone(&db.federation_settings),
             })
             .await?;
 

spa/src/lib/api/goals.ts

@@ -18,11 +18,15 @@ export type UpdateGoalRequest = {
 
 export const userSettingsDtoSchema = z.object({
   federate_goals: z.boolean(),
+  federate_reviews: z.boolean(),
+  federate_watchlist: z.boolean(),
 })
 export type UserSettingsDto = z.infer<typeof userSettingsDtoSchema>
 
 export type UpdateUserSettingsRequest = {
   federate_goals: boolean
+  federate_reviews: boolean
+  federate_watchlist: boolean
 }
 
 export function getGoals() {

spa/src/locales/en.json

@@ -178,6 +178,10 @@
     "privacy": "Privacy",
     "federateGoals": "Share goals on Fediverse",
     "federateGoalsDesc": "Broadcast goal progress to followers",
+    "federateReviews": "Share reviews on Fediverse",
+    "federateReviewsDesc": "Broadcast diary entries to followers",
+    "federateWatchlist": "Share watchlist on Fediverse",
+    "federateWatchlistDesc": "Broadcast watchlist additions to followers",
     "export": "Export",
     "exportDesc": "Download your diary",
     "exportCsv": "CSV",

spa/src/routes/_app/settings/index.tsx

@@ -3,9 +3,11 @@ import { useTranslation } from "react-i18next"
 import { useMutation } from "@tanstack/react-query"
 import {
   ArrowLeft,
+  BookOpen,
   ChevronRight,
   Download,
   Key,
+  List,
   LogOut,
   RefreshCw,
   ShieldBan,
@@ -19,6 +21,7 @@ import { Switch } from "@/components/ui/switch"
 import { useAuth, useIsAdmin } from "@/components/auth-provider"
 import { reindexSearch } from "@/lib/api/users"
 import { useSettings, useUpdateSettings } from "@/hooks/use-goals"
+import type { UpdateUserSettingsRequest } from "@/lib/api/goals"
 import { useDocumentTitle } from "@/hooks/use-document-title"
 
 export const Route = createFileRoute("/_app/settings/")({
@@ -128,6 +131,18 @@ function PrivacySection() {
   const { data: settings } = useSettings()
   const updateMutation = useUpdateSettings()
 
+  const disabled = updateMutation.isPending
+
+  const toggle = (patch: Partial<UpdateUserSettingsRequest>) => {
+    if (!settings) return
+    updateMutation.mutate({
+      federate_goals: settings.federate_goals,
+      federate_reviews: settings.federate_reviews,
+      federate_watchlist: settings.federate_watchlist,
+      ...patch,
+    })
+  }
+
   return (
     <div>
       <p className="mb-1.5 px-1 text-xs font-medium text-muted-foreground">
@@ -145,11 +160,41 @@ function PrivacySection() {
             </p>
           </div>
           <Switch
-            checked={settings?.federate_goals ?? false}
-            onCheckedChange={(checked) =>
-              updateMutation.mutate({ federate_goals: checked })
-            }
-            disabled={updateMutation.isPending}
+            checked={settings?.federate_goals ?? true}
+            onCheckedChange={(checked) => toggle({ federate_goals: checked })}
+            disabled={disabled}
+          />
+        </div>
+        <div className="flex items-center gap-3 p-3">
+          <span className="text-muted-foreground">
+            <BookOpen className="size-4" />
+          </span>
+          <div className="flex-1">
+            <p className="text-sm font-medium">{t("settings.federateReviews")}</p>
+            <p className="text-xs text-muted-foreground">
+              {t("settings.federateReviewsDesc")}
+            </p>
+          </div>
+          <Switch
+            checked={settings?.federate_reviews ?? true}
+            onCheckedChange={(checked) => toggle({ federate_reviews: checked })}
+            disabled={disabled}
+          />
+        </div>
+        <div className="flex items-center gap-3 p-3">
+          <span className="text-muted-foreground">
+            <List className="size-4" />
+          </span>
+          <div className="flex-1">
+            <p className="text-sm font-medium">{t("settings.federateWatchlist")}</p>
+            <p className="text-xs text-muted-foreground">
+              {t("settings.federateWatchlistDesc")}
+            </p>
+          </div>
+          <Switch
+            checked={settings?.federate_watchlist ?? true}
+            onCheckedChange={(checked) => toggle({ federate_watchlist: checked })}
+            disabled={disabled}
           />
         </div>
       </div>