chore: wire WrapUpStatsQuery panic stub into binaries

crates/application/src/context.rs

@@ -6,7 +6,7 @@ use domain::ports::{
     MovieRepository, PasswordHasher, PersonCommand, PersonQuery, PosterFetcherClient,
     RemoteWatchlistRepository, ReviewRepository, SearchCommand, SearchPort, SocialQueryPort,
     StatsRepository, UserProfileFieldsRepository, UserRepository, WatchEventRepository,
-    WatchlistRepository, WebhookTokenRepository,
+    WatchlistRepository, WrapUpStatsQuery, WebhookTokenRepository,
 };
 
 use crate::config::AppConfig;
@@ -31,6 +31,7 @@ pub struct Repositories {
     pub profile_fields: Arc<dyn UserProfileFieldsRepository>,
     pub remote_watchlist: Arc<dyn RemoteWatchlistRepository>,
     pub social_query: Arc<dyn SocialQueryPort>,
+    pub wrapup_stats: Arc<dyn WrapUpStatsQuery>,
 }
 
 #[derive(Clone)]

crates/application/src/lib.rs

@@ -13,6 +13,7 @@ pub mod person;
 pub mod search;
 pub mod users;
 pub mod watchlist;
+pub mod wrapup;
 
 #[cfg(test)]
 pub mod test_helpers;

crates/application/src/test_helpers.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use domain::testing::{NoopRemoteWatchlistRepository, NoopSocialQueryPort};
+use domain::testing::{InMemoryWrapUpStatsQuery, NoopRemoteWatchlistRepository, NoopSocialQueryPort};
 use domain::{
     ports::{
         AuthService, DiaryExporter, DiaryRepository, DocumentParser, EventPublisher, ImageStorage,
@@ -8,6 +8,7 @@ use domain::{
         MovieRepository, PasswordHasher, PersonCommand, PersonQuery, PosterFetcherClient,
         ReviewRepository, SearchCommand, SearchPort, StatsRepository, UserProfileFieldsRepository,
         UserRepository, WatchEventRepository, WatchlistRepository, WebhookTokenRepository,
+        WrapUpStatsQuery,
     },
     testing::{
         FakeAuthService, FakeMetadataClient, FakePasswordHasher, InMemoryMovieRepository,
@@ -50,6 +51,7 @@ pub struct TestContextBuilder {
     pub person_query: Arc<dyn PersonQuery>,
     pub search_port: Arc<dyn SearchPort>,
     pub search_command: Arc<dyn SearchCommand>,
+    pub wrapup_stats: Arc<dyn WrapUpStatsQuery>,
     pub config: AppConfig,
 }
 
@@ -80,6 +82,7 @@ impl TestContextBuilder {
             person_query: Arc::new(PanicPersonQuery),
             search_port: Arc::new(PanicSearchPort),
             search_command: Arc::new(PanicSearchCommand),
+            wrapup_stats: InMemoryWrapUpStatsQuery::new(),
             config: AppConfig {
                 allow_registration: true,
                 base_url: "http://localhost:3000".into(),
@@ -118,6 +121,11 @@ impl TestContextBuilder {
         self
     }
 
+    pub fn wrapup_stats(mut self, r: Arc<dyn WrapUpStatsQuery>) -> Self {
+        self.wrapup_stats = r;
+        self
+    }
+
     pub fn with_config(mut self, config: AppConfig) -> Self {
         self.config = config;
         self
@@ -144,6 +152,7 @@ impl TestContextBuilder {
                 search_command: self.search_command,
                 remote_watchlist: Arc::new(NoopRemoteWatchlistRepository),
                 social_query: Arc::new(NoopSocialQueryPort),
+                wrapup_stats: self.wrapup_stats,
             },
             services: Services {
                 auth: self.auth_service,

crates/application/src/wrapup/compute.rs

@@ -0,0 +1,457 @@
+use std::collections::HashMap;
+
+use chrono::Datelike;
+use uuid::Uuid;
+
+use crate::context::AppContext;
+use crate::wrapup::queries::ComputeWrapUpQuery;
+use domain::errors::DomainError;
+use domain::models::wrapup::*;
+use domain::ports::WrapUpMovieRow;
+
+pub async fn execute(
+    ctx: &AppContext,
+    query: ComputeWrapUpQuery,
+) -> Result<WrapUpReport, DomainError> {
+    let rows = ctx
+        .repos
+        .wrapup_stats
+        .get_reviews_with_profiles(&query.scope, &query.date_range)
+        .await?;
+
+    Ok(build_report(query.scope, query.date_range, &rows))
+}
+
+fn build_report(scope: WrapUpScope, date_range: DateRange, rows: &[WrapUpMovieRow]) -> WrapUpReport {
+    let total_movies = rows.len() as u32;
+
+    let total_watch_time_minutes: u32 = rows.iter().filter_map(|r| r.runtime_minutes).sum();
+
+    let avg_rating = if rows.is_empty() {
+        None
+    } else {
+        Some(rows.iter().map(|r| r.rating as f64).sum::<f64>() / rows.len() as f64)
+    };
+
+    let rating_distribution = {
+        let mut dist = [0u32; 5];
+        for r in rows {
+            let idx = (r.rating as usize).saturating_sub(1).min(4);
+            dist[idx] += 1;
+        }
+        dist
+    };
+
+    let movies_per_month = compute_movies_per_month(rows);
+    let busiest_month = movies_per_month
+        .iter()
+        .max_by_key(|m| m.count)
+        .map(|m| m.label.clone());
+
+    let busiest_day_of_week = compute_busiest_day(rows);
+
+    let (longest_movie, shortest_movie) = compute_runtime_extremes(rows);
+    let (highest_rated_movie, lowest_rated_movie) = compute_rating_extremes(rows);
+    let (first_movie_of_period, last_movie_of_period) = compute_chronological_extremes(rows);
+    let (oldest_movie, newest_movie) = compute_year_extremes(rows);
+
+    let (top_directors, director_diversity) = compute_director_stats(rows);
+    let (top_actors, actor_diversity, top_cast_profile_paths) = compute_actor_stats(rows);
+
+    let (top_genres, genre_diversity, highest_rated_genre, lowest_rated_genre) =
+        compute_genre_stats(rows);
+    let top_keywords = compute_keyword_stats(rows);
+
+    let (total_budget_watched, avg_budget) = compute_budget_stats(rows);
+    let language_distribution = compute_language_stats(rows);
+
+    let (total_rewatches, most_rewatched_movie, avg_rating_change_on_rewatch) =
+        compute_rewatch_stats(rows);
+
+    let (most_active_user, most_watched_movie_global, total_users_active) =
+        if matches!(scope, WrapUpScope::Global) {
+            compute_global_stats(rows)
+        } else {
+            (None, None, None)
+        };
+
+    let poster_paths: Vec<String> = rows.iter().filter_map(|r| r.poster_path.clone()).collect();
+
+    WrapUpReport {
+        scope,
+        date_range,
+        generated_at: chrono::Utc::now(),
+        total_movies,
+        total_watch_time_minutes,
+        movies_per_month,
+        busiest_month,
+        busiest_day_of_week,
+        avg_rating,
+        rating_distribution,
+        longest_movie,
+        shortest_movie,
+        top_directors,
+        top_actors,
+        director_diversity,
+        actor_diversity,
+        top_genres,
+        genre_diversity,
+        highest_rated_genre,
+        lowest_rated_genre,
+        top_keywords,
+        total_budget_watched,
+        avg_budget,
+        language_distribution,
+        oldest_movie,
+        newest_movie,
+        total_rewatches,
+        most_rewatched_movie,
+        avg_rating_change_on_rewatch,
+        highest_rated_movie,
+        lowest_rated_movie,
+        first_movie_of_period,
+        last_movie_of_period,
+        most_active_user,
+        most_watched_movie_global,
+        total_users_active,
+        poster_paths,
+        top_cast_profile_paths,
+    }
+}
+
+fn movie_ref(r: &WrapUpMovieRow) -> MovieRef {
+    MovieRef {
+        title: r.title.clone(),
+        year: r.release_year,
+        runtime_minutes: r.runtime_minutes,
+        poster_path: r.poster_path.clone(),
+    }
+}
+
+fn compute_movies_per_month(rows: &[WrapUpMovieRow]) -> Vec<MonthCount> {
+    let mut counts: HashMap<String, u32> = HashMap::new();
+    for r in rows {
+        let ym = r.watched_at.format("%Y-%m").to_string();
+        *counts.entry(ym).or_default() += 1;
+    }
+    let mut result: Vec<MonthCount> = counts
+        .into_iter()
+        .map(|(ym, count)| {
+            let label = chrono::NaiveDate::parse_from_str(&format!("{ym}-01"), "%Y-%m-%d")
+                .map(|d| d.format("%B %Y").to_string())
+                .unwrap_or_else(|_| ym.clone());
+            MonthCount {
+                year_month: ym,
+                label,
+                count,
+            }
+        })
+        .collect();
+    result.sort_by(|a, b| a.year_month.cmp(&b.year_month));
+    result
+}
+
+fn compute_busiest_day(rows: &[WrapUpMovieRow]) -> Option<String> {
+    if rows.is_empty() {
+        return None;
+    }
+    let mut day_counts = [0u32; 7];
+    for r in rows {
+        let weekday = r.watched_at.date().weekday().num_days_from_monday() as usize;
+        day_counts[weekday] += 1;
+    }
+    let max_idx = day_counts
+        .iter()
+        .enumerate()
+        .max_by_key(|(_, c)| *c)
+        .map(|(i, _)| i)
+        .unwrap_or(0);
+    let names = [
+        "Monday",
+        "Tuesday",
+        "Wednesday",
+        "Thursday",
+        "Friday",
+        "Saturday",
+        "Sunday",
+    ];
+    Some(names[max_idx].to_string())
+}
+
+fn compute_runtime_extremes(rows: &[WrapUpMovieRow]) -> (Option<MovieRef>, Option<MovieRef>) {
+    let with_runtime: Vec<_> = rows.iter().filter(|r| r.runtime_minutes.is_some()).collect();
+    let longest = with_runtime
+        .iter()
+        .max_by_key(|r| r.runtime_minutes.unwrap_or(0))
+        .map(|r| movie_ref(r));
+    let shortest = with_runtime
+        .iter()
+        .min_by_key(|r| r.runtime_minutes.unwrap_or(u32::MAX))
+        .map(|r| movie_ref(r));
+    (longest, shortest)
+}
+
+fn compute_rating_extremes(rows: &[WrapUpMovieRow]) -> (Option<MovieRef>, Option<MovieRef>) {
+    let highest = rows.iter().max_by_key(|r| r.rating).map(|r| movie_ref(r));
+    let lowest = rows.iter().min_by_key(|r| r.rating).map(|r| movie_ref(r));
+    (highest, lowest)
+}
+
+fn compute_chronological_extremes(
+    rows: &[WrapUpMovieRow],
+) -> (Option<MovieRef>, Option<MovieRef>) {
+    let first = rows
+        .iter()
+        .min_by_key(|r| r.watched_at)
+        .map(|r| movie_ref(r));
+    let last = rows
+        .iter()
+        .max_by_key(|r| r.watched_at)
+        .map(|r| movie_ref(r));
+    (first, last)
+}
+
+fn compute_year_extremes(rows: &[WrapUpMovieRow]) -> (Option<MovieRef>, Option<MovieRef>) {
+    let oldest = rows
+        .iter()
+        .min_by_key(|r| r.release_year)
+        .map(|r| movie_ref(r));
+    let newest = rows
+        .iter()
+        .max_by_key(|r| r.release_year)
+        .map(|r| movie_ref(r));
+    (oldest, newest)
+}
+
+fn compute_director_stats(rows: &[WrapUpMovieRow]) -> (Vec<PersonStat>, u32) {
+    let mut director_movies: HashMap<String, Vec<u8>> = HashMap::new();
+    for r in rows {
+        if let Some(ref dir) = r.director {
+            director_movies
+                .entry(dir.clone())
+                .or_default()
+                .push(r.rating);
+        }
+    }
+    let diversity = director_movies.len() as u32;
+    let mut stats: Vec<PersonStat> = director_movies
+        .into_iter()
+        .map(|(name, ratings)| {
+            let count = ratings.len() as u32;
+            let avg = ratings.iter().map(|&r| r as f64).sum::<f64>() / ratings.len() as f64;
+            PersonStat {
+                name,
+                count,
+                avg_rating: avg,
+            }
+        })
+        .collect();
+    stats.sort_by(|a, b| b.count.cmp(&a.count).then(b.avg_rating.total_cmp(&a.avg_rating)));
+    stats.truncate(5);
+    (stats, diversity)
+}
+
+fn compute_actor_stats(rows: &[WrapUpMovieRow]) -> (Vec<PersonStat>, u32, Vec<String>) {
+    let mut actor_movies: HashMap<String, Vec<u8>> = HashMap::new();
+    let mut actor_profiles: HashMap<String, Option<String>> = HashMap::new();
+    for r in rows {
+        for (i, (name, billing)) in r.cast_names.iter().enumerate() {
+            if *billing <= 3 {
+                actor_movies
+                    .entry(name.clone())
+                    .or_default()
+                    .push(r.rating);
+                if let Some(path) = r.cast_profile_paths.get(i) {
+                    actor_profiles
+                        .entry(name.clone())
+                        .or_insert_with(|| path.clone());
+                }
+            }
+        }
+    }
+    let diversity = actor_movies.len() as u32;
+    let mut stats: Vec<PersonStat> = actor_movies
+        .into_iter()
+        .map(|(name, ratings)| {
+            let count = ratings.len() as u32;
+            let avg = ratings.iter().map(|&r| r as f64).sum::<f64>() / ratings.len() as f64;
+            PersonStat {
+                name,
+                count,
+                avg_rating: avg,
+            }
+        })
+        .collect();
+    stats.sort_by(|a, b| b.count.cmp(&a.count).then(b.avg_rating.total_cmp(&a.avg_rating)));
+    stats.truncate(5);
+    let profile_paths: Vec<String> = stats
+        .iter()
+        .filter_map(|s| actor_profiles.get(&s.name)?.clone())
+        .collect();
+    (stats, diversity, profile_paths)
+}
+
+fn compute_genre_stats(
+    rows: &[WrapUpMovieRow],
+) -> (Vec<GenreStat>, u32, Option<String>, Option<String>) {
+    let mut genre_ratings: HashMap<String, Vec<u8>> = HashMap::new();
+    for r in rows {
+        for genre in &r.genres {
+            genre_ratings
+                .entry(genre.clone())
+                .or_default()
+                .push(r.rating);
+        }
+    }
+    let diversity = genre_ratings.len() as u32;
+    let mut stats: Vec<GenreStat> = genre_ratings
+        .into_iter()
+        .map(|(genre, ratings)| {
+            let count = ratings.len() as u32;
+            let avg = ratings.iter().map(|&r| r as f64).sum::<f64>() / ratings.len() as f64;
+            GenreStat {
+                genre,
+                count,
+                avg_rating: avg,
+            }
+        })
+        .collect();
+    stats.sort_by(|a, b| b.count.cmp(&a.count));
+    let highest = stats
+        .iter()
+        .max_by(|a, b| a.avg_rating.total_cmp(&b.avg_rating))
+        .map(|g| g.genre.clone());
+    let lowest = stats
+        .iter()
+        .filter(|g| g.count >= 3)
+        .min_by(|a, b| a.avg_rating.total_cmp(&b.avg_rating))
+        .map(|g| g.genre.clone());
+    stats.truncate(5);
+    (stats, diversity, highest, lowest)
+}
+
+fn compute_keyword_stats(rows: &[WrapUpMovieRow]) -> Vec<KeywordStat> {
+    let mut kw_counts: HashMap<String, u32> = HashMap::new();
+    for r in rows {
+        for kw in &r.keywords {
+            *kw_counts.entry(kw.clone()).or_default() += 1;
+        }
+    }
+    let mut stats: Vec<KeywordStat> = kw_counts
+        .into_iter()
+        .map(|(keyword, count)| KeywordStat { keyword, count })
+        .collect();
+    stats.sort_by(|a, b| b.count.cmp(&a.count));
+    stats.truncate(20);
+    stats
+}
+
+fn compute_budget_stats(rows: &[WrapUpMovieRow]) -> (Option<i64>, Option<i64>) {
+    let budgets: Vec<i64> = rows
+        .iter()
+        .filter_map(|r| r.budget_usd)
+        .filter(|&b| b > 0)
+        .collect();
+    if budgets.is_empty() {
+        return (None, None);
+    }
+    let total: i64 = budgets.iter().sum();
+    let avg = total / budgets.len() as i64;
+    (Some(total), Some(avg))
+}
+
+fn compute_language_stats(rows: &[WrapUpMovieRow]) -> Vec<LangStat> {
+    let mut lang_counts: HashMap<String, u32> = HashMap::new();
+    for r in rows {
+        if let Some(ref lang) = r.original_language {
+            *lang_counts.entry(lang.clone()).or_default() += 1;
+        }
+    }
+    let mut stats: Vec<LangStat> = lang_counts
+        .into_iter()
+        .map(|(language, count)| LangStat { language, count })
+        .collect();
+    stats.sort_by(|a, b| b.count.cmp(&a.count));
+    stats
+}
+
+fn compute_rewatch_stats(rows: &[WrapUpMovieRow]) -> (u32, Option<MovieRef>, Option<f64>) {
+    let mut movie_reviews: HashMap<Uuid, Vec<&WrapUpMovieRow>> = HashMap::new();
+    for r in rows {
+        movie_reviews.entry(r.movie_id).or_default().push(r);
+    }
+
+    let rewatched: Vec<_> = movie_reviews
+        .iter()
+        .filter(|(_, reviews)| reviews.len() > 1)
+        .collect();
+
+    let total_rewatches = rewatched.iter().map(|(_, rs)| rs.len() as u32 - 1).sum();
+
+    let most_rewatched = rewatched
+        .iter()
+        .max_by_key(|(_, rs)| rs.len())
+        .map(|(_, rs)| movie_ref(rs[0]));
+
+    let avg_change = if rewatched.is_empty() {
+        None
+    } else {
+        let changes: Vec<f64> = rewatched
+            .iter()
+            .filter_map(|(_, rs)| {
+                let mut sorted: Vec<_> = rs.to_vec();
+                sorted.sort_by_key(|r| r.watched_at);
+                let first = sorted.first()?.rating as f64;
+                let last = sorted.last()?.rating as f64;
+                Some(last - first)
+            })
+            .collect();
+        if changes.is_empty() {
+            None
+        } else {
+            Some(changes.iter().sum::<f64>() / changes.len() as f64)
+        }
+    };
+
+    (total_rewatches, most_rewatched, avg_change)
+}
+
+fn compute_global_stats(
+    rows: &[WrapUpMovieRow],
+) -> (Option<UserRef>, Option<MovieRef>, Option<u32>) {
+    if rows.is_empty() {
+        return (None, None, None);
+    }
+
+    let mut user_counts: HashMap<Uuid, u32> = HashMap::new();
+    for r in rows {
+        *user_counts.entry(r.user_id).or_default() += 1;
+    }
+    let total_users_active = Some(user_counts.len() as u32);
+
+    let most_active_user = user_counts
+        .iter()
+        .max_by_key(|(_, c)| *c)
+        .map(|(&uid, _)| UserRef {
+            user_id: uid,
+            display_name: String::new(),
+        });
+
+    let mut movie_counts: HashMap<Uuid, (u32, &WrapUpMovieRow)> = HashMap::new();
+    for r in rows {
+        movie_counts
+            .entry(r.movie_id)
+            .and_modify(|(c, _)| *c += 1)
+            .or_insert((1, r));
+    }
+    let most_watched = movie_counts
+        .into_iter()
+        .max_by_key(|(_, (c, _))| *c)
+        .map(|(_, (_, r))| movie_ref(r));
+
+    (most_active_user, most_watched, total_users_active)
+}
+
+#[cfg(test)]
+#[path = "tests/compute.rs"]
+mod tests;

crates/application/src/wrapup/mod.rs

@@ -0,0 +1,2 @@
+pub mod compute;
+pub mod queries;

crates/application/src/wrapup/queries.rs

@@ -0,0 +1,6 @@
+use domain::models::wrapup::{DateRange, WrapUpScope};
+
+pub struct ComputeWrapUpQuery {
+    pub scope: WrapUpScope,
+    pub date_range: DateRange,
+}

crates/application/src/wrapup/tests/compute.rs

@@ -0,0 +1,158 @@
+use chrono::NaiveDate;
+use domain::models::wrapup::{DateRange, WrapUpScope};
+use domain::ports::WrapUpMovieRow;
+use domain::testing::InMemoryWrapUpStatsQuery;
+use uuid::Uuid;
+
+use crate::test_helpers::TestContextBuilder;
+use crate::wrapup::queries::ComputeWrapUpQuery;
+
+fn make_row(title: &str, rating: u8, watched_at: &str) -> WrapUpMovieRow {
+    WrapUpMovieRow {
+        movie_id: Uuid::new_v4(),
+        title: title.to_string(),
+        release_year: 2024,
+        director: Some("Director".to_string()),
+        poster_path: None,
+        rating,
+        watched_at: chrono::NaiveDateTime::parse_from_str(
+            &format!("{watched_at} 20:00:00"),
+            "%Y-%m-%d %H:%M:%S",
+        )
+        .unwrap(),
+        user_id: Uuid::new_v4(),
+        runtime_minutes: Some(120),
+        budget_usd: Some(50_000_000),
+        original_language: Some("en".to_string()),
+        genres: vec!["Action".to_string()],
+        keywords: vec!["heist".to_string()],
+        cast_names: vec![("Actor A".to_string(), 1)],
+        cast_profile_paths: vec![None],
+    }
+}
+
+fn year_2024_range() -> DateRange {
+    DateRange {
+        start: NaiveDate::from_ymd_opt(2024, 1, 1).unwrap(),
+        end: NaiveDate::from_ymd_opt(2025, 1, 1).unwrap(),
+    }
+}
+
+#[tokio::test]
+async fn empty_report() {
+    let stats = InMemoryWrapUpStatsQuery::new();
+    let ctx = TestContextBuilder::new().wrapup_stats(stats).build();
+    let user_id = Uuid::new_v4();
+
+    let report = super::execute(
+        &ctx,
+        ComputeWrapUpQuery {
+            scope: WrapUpScope::User(user_id),
+            date_range: year_2024_range(),
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(report.total_movies, 0);
+    assert!(report.avg_rating.is_none());
+    assert_eq!(report.rating_distribution, [0; 5]);
+}
+
+#[tokio::test]
+async fn basic_stats() {
+    let user_id = Uuid::new_v4();
+    let mut r1 = make_row("Film A", 4, "2024-03-10");
+    r1.user_id = user_id;
+    r1.runtime_minutes = Some(120);
+    r1.genres = vec!["Action".to_string()];
+
+    let mut r2 = make_row("Film B", 2, "2024-03-20");
+    r2.user_id = user_id;
+    r2.runtime_minutes = Some(90);
+    r2.genres = vec!["Comedy".to_string()];
+
+    let stats = InMemoryWrapUpStatsQuery::with_rows(vec![r1, r2]);
+    let ctx = TestContextBuilder::new().wrapup_stats(stats).build();
+
+    let report = super::execute(
+        &ctx,
+        ComputeWrapUpQuery {
+            scope: WrapUpScope::User(user_id),
+            date_range: year_2024_range(),
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(report.total_movies, 2);
+    assert_eq!(report.total_watch_time_minutes, 210);
+    assert!((report.avg_rating.unwrap() - 3.0).abs() < f64::EPSILON);
+    assert_eq!(report.rating_distribution, [0, 1, 0, 1, 0]);
+    assert_eq!(report.busiest_month.as_deref(), Some("March 2024"));
+    assert_eq!(report.director_diversity, 1);
+    assert_eq!(report.genre_diversity, 2);
+}
+
+#[tokio::test]
+async fn rewatch_detection() {
+    let user_id = Uuid::new_v4();
+    let movie_id = Uuid::new_v4();
+
+    let mut r1 = make_row("Film A", 3, "2024-02-01");
+    r1.user_id = user_id;
+    r1.movie_id = movie_id;
+
+    let mut r2 = make_row("Film A", 5, "2024-06-01");
+    r2.user_id = user_id;
+    r2.movie_id = movie_id;
+
+    let stats = InMemoryWrapUpStatsQuery::with_rows(vec![r1, r2]);
+    let ctx = TestContextBuilder::new().wrapup_stats(stats).build();
+
+    let report = super::execute(
+        &ctx,
+        ComputeWrapUpQuery {
+            scope: WrapUpScope::User(user_id),
+            date_range: year_2024_range(),
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(report.total_rewatches, 1);
+    assert_eq!(
+        report.most_rewatched_movie.as_ref().unwrap().title,
+        "Film A"
+    );
+    assert!((report.avg_rating_change_on_rewatch.unwrap() - 2.0).abs() < f64::EPSILON);
+}
+
+#[tokio::test]
+async fn global_scope() {
+    let user_a = Uuid::new_v4();
+    let user_b = Uuid::new_v4();
+
+    let mut r1 = make_row("Film X", 4, "2024-05-01");
+    r1.user_id = user_a;
+
+    let mut r2 = make_row("Film Y", 3, "2024-07-01");
+    r2.user_id = user_b;
+
+    let stats = InMemoryWrapUpStatsQuery::with_rows(vec![r1, r2]);
+    let ctx = TestContextBuilder::new().wrapup_stats(stats).build();
+
+    let report = super::execute(
+        &ctx,
+        ComputeWrapUpQuery {
+            scope: WrapUpScope::Global,
+            date_range: year_2024_range(),
+        },
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(report.total_movies, 2);
+    assert_eq!(report.total_users_active, Some(2));
+    assert!(report.most_active_user.is_some());
+}

crates/domain/src/models/mod.rs

@@ -19,6 +19,8 @@ pub use watchlist::{WatchlistEntry, WatchlistWithMovie};
 pub mod remote_watchlist;
 pub use remote_watchlist::RemoteWatchlistEntry;
 pub mod watch_event;
+pub mod wrapup;
+pub use wrapup::*;
 pub use watch_event::{
     ParsedPlaybackEvent, PersistedWatchEvent, WatchEvent, WatchEventSource, WatchEventStatus,
     WebhookToken,

crates/domain/src/models/wrapup.rs

@@ -0,0 +1,119 @@
+use chrono::NaiveDate;
+use uuid::Uuid;
+
+#[derive(Clone, Debug)]
+pub struct DateRange {
+    pub start: NaiveDate,
+    pub end: NaiveDate,
+}
+
+#[derive(Clone, Debug)]
+pub struct MovieRef {
+    pub title: String,
+    pub year: u16,
+    pub runtime_minutes: Option<u32>,
+    pub poster_path: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+pub struct UserRef {
+    pub user_id: Uuid,
+    pub display_name: String,
+}
+
+#[derive(Clone, Debug)]
+pub struct PersonStat {
+    pub name: String,
+    pub count: u32,
+    pub avg_rating: f64,
+}
+
+#[derive(Clone, Debug)]
+pub struct GenreStat {
+    pub genre: String,
+    pub count: u32,
+    pub avg_rating: f64,
+}
+
+#[derive(Clone, Debug)]
+pub struct KeywordStat {
+    pub keyword: String,
+    pub count: u32,
+}
+
+#[derive(Clone, Debug)]
+pub struct LangStat {
+    pub language: String,
+    pub count: u32,
+}
+
+#[derive(Clone, Debug)]
+pub struct MonthCount {
+    pub year_month: String,
+    pub label: String,
+    pub count: u32,
+}
+
+#[derive(Clone, Debug)]
+pub enum WrapUpScope {
+    User(Uuid),
+    Global,
+}
+
+#[derive(Clone, Debug)]
+pub struct WrapUpReport {
+    pub scope: WrapUpScope,
+    pub date_range: DateRange,
+    pub generated_at: chrono::DateTime<chrono::Utc>,
+
+    // Core viewing
+    pub total_movies: u32,
+    pub total_watch_time_minutes: u32,
+    pub movies_per_month: Vec<MonthCount>,
+    pub busiest_month: Option<String>,
+    pub busiest_day_of_week: Option<String>,
+    pub avg_rating: Option<f64>,
+    pub rating_distribution: [u32; 5],
+    pub longest_movie: Option<MovieRef>,
+    pub shortest_movie: Option<MovieRef>,
+
+    // People insights
+    pub top_directors: Vec<PersonStat>,
+    pub top_actors: Vec<PersonStat>,
+    pub director_diversity: u32,
+    pub actor_diversity: u32,
+
+    // Genre & taste
+    pub top_genres: Vec<GenreStat>,
+    pub genre_diversity: u32,
+    pub highest_rated_genre: Option<String>,
+    pub lowest_rated_genre: Option<String>,
+    pub top_keywords: Vec<KeywordStat>,
+
+    // Financial/production (None when data unavailable)
+    pub total_budget_watched: Option<i64>,
+    pub avg_budget: Option<i64>,
+    pub language_distribution: Vec<LangStat>,
+    pub oldest_movie: Option<MovieRef>,
+    pub newest_movie: Option<MovieRef>,
+
+    // Rewatch stats
+    pub total_rewatches: u32,
+    pub most_rewatched_movie: Option<MovieRef>,
+    pub avg_rating_change_on_rewatch: Option<f64>,
+
+    // Highlights
+    pub highest_rated_movie: Option<MovieRef>,
+    pub lowest_rated_movie: Option<MovieRef>,
+    pub first_movie_of_period: Option<MovieRef>,
+    pub last_movie_of_period: Option<MovieRef>,
+
+    // Global-only (None for per-user)
+    pub most_active_user: Option<UserRef>,
+    pub most_watched_movie_global: Option<MovieRef>,
+    pub total_users_active: Option<u32>,
+
+    // Asset references for renderers
+    pub poster_paths: Vec<String>,
+    pub top_cast_profile_paths: Vec<String>,
+}

crates/domain/src/ports.rs

@@ -1,5 +1,6 @@
 use async_trait::async_trait;
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, NaiveDateTime, Utc};
+use uuid::Uuid;
 
 use crate::{
     errors::DomainError,
@@ -12,6 +13,7 @@ use crate::{
         ReviewHistory, SearchQuery, SearchResults, User, UserStats, UserSummary, UserTrends,
         WatchEvent, WatchEventStatus, WatchlistEntry, WatchlistWithMovie, WebhookToken,
         collections::{self, PageParams, Paginated},
+        wrapup::{DateRange, WrapUpScope},
     },
     value_objects::{
         Email, ExternalMetadataId, ImportProfileId, ImportSessionId, MovieId, MovieTitle,
@@ -468,3 +470,33 @@ pub trait WebhookTokenRepository: Send + Sync {
     async fn delete(&self, id: &WebhookTokenId, user_id: &UserId) -> Result<(), DomainError>;
     async fn touch_last_used(&self, id: &WebhookTokenId) -> Result<(), DomainError>;
 }
+
+// ── Wrap-up / Year-in-Review ─────────────────────────────────────────────────
+
+#[derive(Clone, Debug)]
+pub struct WrapUpMovieRow {
+    pub movie_id: Uuid,
+    pub title: String,
+    pub release_year: u16,
+    pub director: Option<String>,
+    pub poster_path: Option<String>,
+    pub rating: u8,
+    pub watched_at: NaiveDateTime,
+    pub user_id: Uuid,
+    pub runtime_minutes: Option<u32>,
+    pub budget_usd: Option<i64>,
+    pub original_language: Option<String>,
+    pub genres: Vec<String>,
+    pub keywords: Vec<String>,
+    pub cast_names: Vec<(String, u32)>,
+    pub cast_profile_paths: Vec<Option<String>>,
+}
+
+#[async_trait]
+pub trait WrapUpStatsQuery: Send + Sync {
+    async fn get_reviews_with_profiles(
+        &self,
+        scope: &WrapUpScope,
+        range: &DateRange,
+    ) -> Result<Vec<WrapUpMovieRow>, DomainError>;
+}

crates/domain/src/testing.rs

@@ -991,3 +991,62 @@ impl crate::ports::WebhookTokenRepository for PanicWebhookTokenRepository {
         panic!("PanicWebhookTokenRepository called")
     }
 }
+
+// ── PanicWrapUpStatsQuery ───────────────────────────────────────────────────
+
+pub struct PanicWrapUpStatsQuery;
+
+#[async_trait]
+impl crate::ports::WrapUpStatsQuery for PanicWrapUpStatsQuery {
+    async fn get_reviews_with_profiles(
+        &self,
+        _scope: &crate::models::wrapup::WrapUpScope,
+        _range: &crate::models::wrapup::DateRange,
+    ) -> Result<Vec<crate::ports::WrapUpMovieRow>, DomainError> {
+        unimplemented!("WrapUpStatsQuery not wired")
+    }
+}
+
+// ── InMemoryWrapUpStatsQuery ────────────────────────────────────────────────
+
+pub struct InMemoryWrapUpStatsQuery {
+    pub rows: Mutex<Vec<crate::ports::WrapUpMovieRow>>,
+}
+
+impl InMemoryWrapUpStatsQuery {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self {
+            rows: Mutex::new(Vec::new()),
+        })
+    }
+
+    pub fn with_rows(rows: Vec<crate::ports::WrapUpMovieRow>) -> Arc<Self> {
+        Arc::new(Self {
+            rows: Mutex::new(rows),
+        })
+    }
+}
+
+#[async_trait]
+impl crate::ports::WrapUpStatsQuery for InMemoryWrapUpStatsQuery {
+    async fn get_reviews_with_profiles(
+        &self,
+        scope: &crate::models::wrapup::WrapUpScope,
+        range: &crate::models::wrapup::DateRange,
+    ) -> Result<Vec<crate::ports::WrapUpMovieRow>, DomainError> {
+        let rows = self.rows.lock().unwrap();
+        let filtered: Vec<_> = rows
+            .iter()
+            .filter(|r| {
+                let date = r.watched_at.date();
+                date >= range.start && date < range.end
+            })
+            .filter(|r| match scope {
+                crate::models::wrapup::WrapUpScope::User(uid) => r.user_id == *uid,
+                crate::models::wrapup::WrapUpScope::Global => true,
+            })
+            .cloned()
+            .collect();
+        Ok(filtered)
+    }
+}

crates/presentation/Cargo.toml

@@ -42,7 +42,7 @@ chrono = { workspace = true }
 async-trait = { workspace = true }
 
 api-types = { workspace = true }
-domain = { workspace = true }
+domain = { workspace = true, features = ["test-helpers"] }
 application = { workspace = true }
 auth = { workspace = true }
 metadata = { workspace = true }

crates/presentation/src/main.rs

@@ -193,6 +193,7 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
             social_query: social_query.clone(),
             #[cfg(not(feature = "federation"))]
             social_query: Arc::new(domain::testing::NoopSocialQueryPort),
+            wrapup_stats: Arc::new(domain::testing::PanicWrapUpStatsQuery) as Arc<dyn domain::ports::WrapUpStatsQuery>,
         },
         services: Services {
             auth: auth_service,

crates/worker/Cargo.toml

@@ -13,7 +13,7 @@ sqlite-federation = ["sqlite", "dep:sqlite-federation", "dep:activitypub", "fede
 postgres-federation = ["postgres", "dep:postgres-federation", "dep:activitypub", "federation"]
 
 [dependencies]
-domain = { workspace = true }
+domain = { workspace = true, features = ["test-helpers"] }
 application = { workspace = true }
 tokio = { workspace = true }
 anyhow = { workspace = true }

crates/worker/src/main.rs

@@ -92,6 +92,7 @@ async fn main() -> anyhow::Result<()> {
             social_query: fed_social_query,
             #[cfg(not(feature = "federation"))]
             social_query: Arc::new(domain::testing::NoopSocialQueryPort),
+            wrapup_stats: Arc::new(domain::testing::PanicWrapUpStatsQuery) as Arc<dyn domain::ports::WrapUpStatsQuery>,
         },
         services: Services {
             auth: auth_service,