refactor/tighten-architecture #26

Merged
GKaszewski merged 104 commits from refactor/tighten-architecture into master 2026-10-26 09:36:27 +00:00
11 changed files with 650 additions and 2086 deletions
Showing only changes of commit 1e6e9ddf6b - Show all commits

13
Cargo.lock generated
View File

@@ -4929,25 +4929,12 @@ name = "server"
version = "0.1.0"
dependencies = [
"anyhow",
"application",
"async-trait",
"axum",
"bootstrap",
"bytes",
"config",
"domain",
"dotenvy",
"futures",
"http-axum",
"http-body-util",
"rss 0.1.0",
"serde_json",
"sqlite",
"sqlx",
"tokio",
"tower",
"tracing",
"uuid",
]
[[package]]

View File

@@ -43,7 +43,6 @@ sqlite-federation = { workspace = true, optional = true }
[dev-dependencies]
chrono = { workspace = true }
domain = { workspace = true, features = ["test-helpers"] }
uuid = { workspace = true }
serde_json = { workspace = true }
tower = { version = "0.5", features = ["util"] }

View File

@@ -21,8 +21,4 @@ pub use event_bus::EventBusBackend;
pub use repositories::Repositories;
pub use server::{ServerApp, build_server, build_server_on};
pub use tracing_setup::init_tracing;
pub use worker::{WorkerApp, build_worker};
#[cfg(test)]
#[path = "tests/build.rs"]
mod build_tests;
pub use worker::{WorkerApp, build_worker, build_worker_on};

File diff suppressed because it is too large Load Diff

View File

@@ -1,13 +1,13 @@
use std::sync::Arc;
use application::config::AppConfig as AppSettings;
use application::{
MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, worker::WorkerService,
};
use domain::ports::{EventHandler, MovieEnrichmentClient, PeriodicJob, PersonEnrichmentClient};
use domain::value_objects::InstanceIdentity;
use crate::{database, event_bus};
use crate::database::{self, DatabaseOutput};
use crate::event_bus;
pub struct WorkerApp {
pub service: WorkerService,
@@ -17,9 +17,27 @@ pub struct WorkerApp {
/// spawned) and the consumer. Run it with `service.run(shutdown)`.
pub async fn build_worker(config: &config::AppConfig) -> anyhow::Result<WorkerApp> {
database::warn_if_database_backend_set();
let app_config = AppSettings::from(config);
let instance = InstanceIdentity::new(app_config.base_url.clone());
let instance = InstanceIdentity::new(config.server.base_url.clone());
let db = database::build_database_adapters(&config.database.url, &instance).await?;
assemble(config, db, instance).await
}
/// Builds the same worker as [`build_worker`] on a caller-provided pool.
/// Migrations run on it exactly as they do for the normal path.
pub async fn build_worker_on(
config: &config::AppConfig,
pool: sqlx::SqlitePool,
) -> anyhow::Result<WorkerApp> {
let instance = InstanceIdentity::new(config.server.base_url.clone());
let db = database::build_database_adapters_on(pool, &instance).await?;
assemble(config, db, instance).await
}
async fn assemble(
config: &config::AppConfig,
db: DatabaseOutput,
#[cfg_attr(not(feature = "federation"), allow(unused_variables))] instance: InstanceIdentity,
) -> anyhow::Result<WorkerApp> {
let metadata_client = metadata::create(&config.metadata)?;
let poster_fetcher = poster_fetcher::create(&config.metadata)?;
let object_storage = object_storage::create(&config.storage)?;
@@ -40,7 +58,6 @@ pub async fn build_worker(config: &config::AppConfig) -> anyhow::Result<WorkerAp
.clone()
.map(|c| c as Arc<dyn PersonEnrichmentClient>);
let db = database::build_database_adapters(&config.database.url, &instance).await?;
let (event_publisher_arc, consumer_arc) =
event_bus::create_channel(&db.db_pool, config).await?;

View File

@@ -0,0 +1,390 @@
//! The HTTP surface of the production app, built through `bootstrap` on an
//! in-memory database with real adapters.
mod support;
use axum::{
Router,
body::Body,
http::{Request, StatusCode},
};
use sqlx::SqlitePool;
use tower::ServiceExt;
use uuid::Uuid;
async fn test_app() -> Router {
support::server().await.0
}
/// An app with one registered user; returns its pool, id and bearer token.
async fn test_app_with_user() -> (Router, SqlitePool, Uuid, String) {
let (app, pool) = support::server().await;
let session = support::register(&app, "insights", "insights@example.com").await;
let user_id = session.user_id.parse().unwrap();
(app, pool, user_id, session.token)
}
/// Inject a fake peer IP so the GovernorLayer can extract ConnectInfo.
fn with_ip(req: Request<Body>) -> Request<Body> {
let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap();
let mut req = req;
req.extensions_mut()
.insert(axum::extract::ConnectInfo::<std::net::SocketAddr>(addr));
req
}
#[tokio::test]
async fn get_api_diary_returns_empty_list() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/diary")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(json["total_count"], 0);
assert_eq!(json["items"], serde_json::json!([]));
assert_eq!(json["limit"], 5);
assert_eq!(json["offset"], 0);
}
#[tokio::test]
async fn post_api_reviews_without_auth_returns_401() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.method("POST")
.uri("/api/v1/reviews")
.header("content-type", "application/json")
.body(Body::from(
r#"{"rating":4,"watched_at":"2026-01-01T20:00:00","manual_title":"Dune","manual_release_year":2021}"#,
))
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn post_api_auth_login_unknown_user_returns_401() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.method("POST")
.uri("/api/v1/auth/login")
.header("content-type", "application/json")
.body(Body::from(r#"{"email":"a@b.com","password":"x"}"#))
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn get_api_movie_detail_returns_404_for_unknown_id() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/movies/00000000-0000-0000-0000-000000000000")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn tags_moviesdiary_redirects_to_home() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/tags/moviesdiary")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::TEMPORARY_REDIRECT);
assert_eq!(response.headers().get("location").unwrap(), "/");
}
#[tokio::test]
async fn tags_other_redirects_to_search() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/tags/batman")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::TEMPORARY_REDIRECT);
assert_eq!(
response.headers().get("location").unwrap(),
"/?search=batman"
);
}
#[tokio::test]
async fn get_movie_detail_html_returns_404_for_unknown_id() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/movies/00000000-0000-0000-0000-000000000000")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
// ── Insights ─────────────────────────────────────────────────────────────────
async fn seed_insights_diary(pool: &SqlitePool, user_id: Uuid) {
// 12 thrillers rated 5 late at night, 12 comedies rated 2 — enough for a
// rating lift once genres come from movie_genres.
for i in 0..12u32 {
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Thriller {i}"))
.bind(2020)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(44400 + i as i64)
.bind("Thriller")
.execute(pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at) \
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(5)
.bind("2024-01-05 22:30:00")
.bind("2024-01-05 22:30:00")
.execute(pool)
.await
.unwrap();
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Comedy {i}"))
.bind(2015)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(55300 + i as i64)
.bind("Comedy")
.execute(pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at) \
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(2)
.bind("2024-03-10 13:00:00")
.bind("2024-03-10 13:00:00")
.execute(pool)
.await
.unwrap();
}
}
#[tokio::test]
async fn get_api_insights_returns_rating_lift_for_own_user() {
let (app, pool, user_id, token) = test_app_with_user().await;
seed_insights_diary(&pool, user_id).await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/insights")
.header("authorization", format!("Bearer {token}"))
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
let (status, bytes) = (
response.status(),
axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap(),
);
assert_eq!(
status,
StatusCode::OK,
"body: {}",
String::from_utf8(bytes.to_vec()).unwrap()
);
let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
let rating_lifts: Vec<&serde_json::Value> = json["insights"]
.as_array()
.unwrap()
.iter()
.filter(|i| i["kind"] == "rating_lift")
.collect();
assert_eq!(rating_lifts.len(), 2, "thriller up, comedy down: {json}");
let thriller = rating_lifts
.iter()
.find(|i| i["genre"] == "Thriller")
.unwrap();
assert_eq!(thriller["delta"], 1.5);
assert_eq!(thriller["sample_count"], 12);
// Timestamps are real → rhythm fires. Hours split evenly between 22
// (thrillers) and 13 (comedies); every window covering either bucket ties,
// so the earliest one, 11:00–14:00, wins. Days tie too (Fri/Sun), sorted
// ascending.
let rhythm = json["insights"]
.as_array()
.unwrap()
.iter()
.find(|i| i["kind"] == "rhythm")
.expect("rhythm expected for real timestamps");
assert_eq!(rhythm["peak_hour_start"], 11);
assert_eq!(rhythm["peak_hour_end"], 14);
assert_eq!(rhythm["top_days"], serde_json::json!([5, 7]));
}
async fn get_insights_json(
app: Router,
token: &str,
query: &str,
) -> (StatusCode, serde_json::Value) {
let response = app
.oneshot(with_ip(
Request::builder()
.uri(format!("/api/v1/insights{query}"))
.header("authorization", format!("Bearer {token}"))
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
let status = response.status();
let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
(status, json)
}
#[tokio::test]
async fn get_api_insights_ignores_remote_reviews() {
let (app, pool, user_id, token) = test_app_with_user().await;
seed_insights_diary(&pool, user_id).await;
// Federated rows stored under the same user_id are someone else's watches.
for i in 0..12u32 {
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Horror {i}"))
.bind(2019)
.execute(&pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(66600 + i as i64)
.bind("Horror")
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at, remote_actor_url) \
VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(5)
.bind("2024-05-05 09:00:00")
.bind("2024-05-05 09:00:00")
.bind("https://remote.example/users/someone")
.execute(&pool)
.await
.unwrap();
}
let (status, json) = get_insights_json(app, &token, "").await;
assert_eq!(status, StatusCode::OK);
let insights = json["insights"].as_array().unwrap();
assert!(
!insights.iter().any(|i| i["genre"] == "Horror"),
"remote reviews must not feed insights: {json}"
);
let thriller = insights
.iter()
.find(|i| i["kind"] == "rating_lift" && i["genre"] == "Thriller")
.unwrap();
assert_eq!(thriller["delta"], 1.5);
}
#[tokio::test]
async fn get_api_insights_buckets_rhythm_in_requested_time_zone() {
let (app, pool, user_id, token) = test_app_with_user().await;
seed_insights_diary(&pool, user_id).await;
// January in Warsaw is UTC+1: 22:30 → 23:30 and 13:00 → 14:00, so the
// earliest tied window moves from 11–14 to 12–15.
let (status, json) = get_insights_json(app, &token, "?tz=Europe/Warsaw").await;
assert_eq!(status, StatusCode::OK);
let rhythm = json["insights"]
.as_array()
.unwrap()
.iter()
.find(|i| i["kind"] == "rhythm")
.expect("rhythm expected");
assert_eq!(rhythm["peak_hour_start"], 12);
assert_eq!(rhythm["peak_hour_end"], 15);
}
#[tokio::test]
async fn get_api_insights_rejects_unknown_time_zone() {
let (app, _pool, _user_id, token) = test_app_with_user().await;
let (status, _) = get_insights_json(app, &token, "?tz=Mars/Olympus").await;
assert_eq!(status, StatusCode::BAD_REQUEST);
}

View File

@@ -0,0 +1,69 @@
//! The composition root assembles on a fresh database, with every adapter real.
mod support;
#[tokio::test]
async fn server_builds_on_a_fresh_database() {
let pool = support::memory_pool().await;
let router = bootstrap::build_server_on(&support::test_config(), pool)
.await
.expect("server builds");
let (status, _, _) = support::send(&router, support::get("/health", None)).await;
assert_eq!(status, 200);
}
#[tokio::test]
async fn worker_builds_on_a_fresh_database() {
let pool = support::memory_pool().await;
bootstrap::build_worker_on(&support::test_config(), pool)
.await
.expect("worker builds");
}
/// Without the federation feature, following another local user still works
/// (the request lands as pending for the target):
/// the local social ports are wired in place of the ActivityPub ones.
#[cfg(not(feature = "federation"))]
#[tokio::test]
async fn federation_off_build_wires_local_social() {
use axum::http::Method;
let (router, _pool) = support::server().await;
let alice = support::register(&router, "alice", "alice@example.com").await;
let bob = support::register(&router, "bob", "bob@example.com").await;
let (status, _, body) = support::send(
&router,
support::json_request(
Method::POST,
"/api/v1/social/follow",
Some(&alice.token),
serde_json::json!({ "handle": "bob@localhost:3000" }),
),
)
.await;
assert_eq!(status, 200, "follow: {body}");
let (status, _, body) = support::send(
&router,
axum::http::Request::builder()
.uri("/api/v1/social/followers/pending")
.header("authorization", format!("Bearer {}", bob.token))
.body(axum::body::Body::empty())
.unwrap(),
)
.await;
assert_eq!(status, 200, "pending followers: {body}");
let followers: serde_json::Value = serde_json::from_str(&body).unwrap();
let actors = followers["actors"].as_array().unwrap();
assert_eq!(
actors.len(),
1,
"bob should have one pending follower: {body}"
);
assert!(
body.contains(&alice.user_id) || body.contains("alice"),
"follower should be alice: {body}"
);
}

View File

@@ -1,14 +1,15 @@
//! Characterization of the classic HTML routes, feeds and the OpenAPI surface,
//! recorded before the HTTP layer moves. Statuses are what the app returns today.
mod support;
use axum::{
Router,
body::Body,
http::{Method, Request, header},
};
use tower::ServiceExt;
use support::{get, json_request, send};
const PASSWORD: &str = "correct-horse-battery";
const EMAIL: &str = "owner@example.com";
const USERNAME: &str = "owner";
@@ -83,135 +84,11 @@ struct Seed {
owner_cookie: String,
}
fn test_config() -> config::AppConfig {
config::AppConfig {
server: config::ServerConfig {
allow_registration: true,
base_url: "http://localhost:3000".into(),
..Default::default()
},
database: config::DatabaseConfig {
url: "sqlite::memory:".into(),
},
auth: config::AuthConfig {
jwt_secret: Some("route-smoke-test-secret-0123456789abcdef".into()),
..Default::default()
},
events: Default::default(),
nats: Default::default(),
storage: Default::default(),
image_conversion: Default::default(),
// Required to build; never called (seeding logs reviews with manual titles).
metadata: config::MetadataConfig {
tmdb_api_key: Some("test-key".into()),
..Default::default()
},
federation: Default::default(),
wrapup: Default::default(),
}
}
async fn send(router: &Router, mut req: Request<Body>) -> (u16, axum::http::HeaderMap, String) {
let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap();
req.extensions_mut()
.insert(axum::extract::ConnectInfo::<std::net::SocketAddr>(addr));
let resp = router.clone().oneshot(req).await.unwrap();
let status = resp.status().as_u16();
let headers = resp.headers().clone();
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
(
status,
headers,
String::from_utf8_lossy(&bytes).into_owned(),
)
}
fn get(uri: &str, cookie: Option<&str>) -> Request<Body> {
let mut b = Request::builder().method(Method::GET).uri(uri);
if let Some(c) = cookie {
b = b.header(header::COOKIE, c);
}
b.body(Body::empty()).unwrap()
}
fn cookie_value(headers: &axum::http::HeaderMap, name: &str) -> Option<String> {
headers
.get_all(header::SET_COOKIE)
.iter()
.filter_map(|v| v.to_str().ok())
.find_map(|c| {
c.split(';')
.next()?
.trim()
.strip_prefix(&format!("{name}="))
.map(str::to_string)
})
}
fn json_request(
method: Method,
uri: &str,
token: Option<&str>,
body: serde_json::Value,
) -> Request<Body> {
let mut b = Request::builder()
.method(method)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json");
if let Some(t) = token {
b = b.header(header::AUTHORIZATION, format!("Bearer {t}"));
}
b.body(Body::from(body.to_string())).unwrap()
}
/// One user (registered through the classic form), one movie with a spoiler review, one goal.
async fn seed() -> Seed {
let pool = sqlx::sqlite::SqlitePoolOptions::new()
.max_connections(1)
.connect("sqlite::memory:")
.await
.unwrap();
let router = bootstrap::build_server_on(&test_config(), pool.clone())
.await
.unwrap();
// Register through the form: GET for the csrf cookie, then POST with the token.
let (status, headers, _) = send(&router, get("/register", None)).await;
assert_eq!(status, 200, "GET /register");
let csrf = cookie_value(&headers, "csrf").expect("csrf cookie");
let form = format!(
"email={}&username={USERNAME}&password={PASSWORD}&_csrf={csrf}",
EMAIL.replace('@', "%40")
);
let req = Request::builder()
.method(Method::POST)
.uri("/register")
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
.header(header::COOKIE, format!("csrf={csrf}"))
.body(Body::from(form))
.unwrap();
let (status, headers, _) = send(&router, req).await;
assert_eq!(status, 303, "POST /register");
let session = cookie_value(&headers, "token").expect("session cookie after register");
let owner_cookie = format!("csrf={csrf}; token={session}");
// Bearer token for the JSON API.
let (status, _, body) = send(
&router,
json_request(
Method::POST,
"/api/v1/auth/login",
None,
serde_json::json!({ "email": EMAIL, "password": PASSWORD }),
),
)
.await;
assert_eq!(status, 200, "login: {body}");
let login: serde_json::Value = serde_json::from_str(&body).unwrap();
let token = login["token"].as_str().unwrap().to_string();
let user_id = login["user_id"].as_str().unwrap().to_string();
let (router, pool) = support::server().await;
let session = support::register(&router, USERNAME, EMAIL).await;
let (token, user_id, owner_cookie) = (session.token, session.user_id, session.cookie);
let (status, _, body) = send(
&router,

View File

@@ -0,0 +1,162 @@
//! Shared helpers for the bootstrap integration tests: a production app built
//! through `bootstrap` on an in-memory database, driven over HTTP.
#![allow(dead_code)]
use axum::{
Router,
body::Body,
http::{Method, Request, header},
};
use tower::ServiceExt;
pub const PASSWORD: &str = "correct-horse-battery";
pub fn test_config() -> config::AppConfig {
config::AppConfig {
server: config::ServerConfig {
allow_registration: true,
base_url: "http://localhost:3000".into(),
..Default::default()
},
database: config::DatabaseConfig {
url: "sqlite::memory:".into(),
},
auth: config::AuthConfig {
jwt_secret: Some("route-smoke-test-secret-0123456789abcdef".into()),
..Default::default()
},
events: Default::default(),
nats: Default::default(),
storage: Default::default(),
image_conversion: Default::default(),
// Required to build; never called (seeding logs reviews with manual titles).
metadata: config::MetadataConfig {
tmdb_api_key: Some("test-key".into()),
..Default::default()
},
federation: Default::default(),
wrapup: Default::default(),
}
}
pub async fn send(router: &Router, mut req: Request<Body>) -> (u16, axum::http::HeaderMap, String) {
let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap();
req.extensions_mut()
.insert(axum::extract::ConnectInfo::<std::net::SocketAddr>(addr));
let resp = router.clone().oneshot(req).await.unwrap();
let status = resp.status().as_u16();
let headers = resp.headers().clone();
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
.await
.unwrap();
(
status,
headers,
String::from_utf8_lossy(&bytes).into_owned(),
)
}
pub fn get(uri: &str, cookie: Option<&str>) -> Request<Body> {
let mut b = Request::builder().method(Method::GET).uri(uri);
if let Some(c) = cookie {
b = b.header(header::COOKIE, c);
}
b.body(Body::empty()).unwrap()
}
pub fn cookie_value(headers: &axum::http::HeaderMap, name: &str) -> Option<String> {
headers
.get_all(header::SET_COOKIE)
.iter()
.filter_map(|v| v.to_str().ok())
.find_map(|c| {
c.split(';')
.next()?
.trim()
.strip_prefix(&format!("{name}="))
.map(str::to_string)
})
}
pub fn json_request(
method: Method,
uri: &str,
token: Option<&str>,
body: serde_json::Value,
) -> Request<Body> {
let mut b = Request::builder()
.method(method)
.uri(uri)
.header(header::CONTENT_TYPE, "application/json");
if let Some(t) = token {
b = b.header(header::AUTHORIZATION, format!("Bearer {t}"));
}
b.body(Body::from(body.to_string())).unwrap()
}
/// A single-connection in-memory pool (every connection would otherwise get its own database).
pub async fn memory_pool() -> sqlx::SqlitePool {
sqlx::sqlite::SqlitePoolOptions::new()
.max_connections(1)
.connect("sqlite::memory:")
.await
.unwrap()
}
/// The production server router on a fresh in-memory database, plus its pool.
pub async fn server() -> (Router, sqlx::SqlitePool) {
let pool = memory_pool().await;
let router = bootstrap::build_server_on(&test_config(), pool.clone())
.await
.unwrap();
(router, pool)
}
pub struct Session {
pub user_id: String,
/// Bearer token for the JSON API.
pub token: String,
/// Cookie header for the classic UI (csrf + session).
pub cookie: String,
}
/// Registers through the classic form (GET for the csrf cookie, then POST with
/// the token), then logs in over the JSON API for a bearer token.
pub async fn register(router: &Router, username: &str, email: &str) -> Session {
let (status, headers, _) = send(router, get("/register", None)).await;
assert_eq!(status, 200, "GET /register");
let csrf = cookie_value(&headers, "csrf").expect("csrf cookie");
let form = format!(
"email={}&username={username}&password={PASSWORD}&_csrf={csrf}",
email.replace('@', "%40")
);
let req = Request::builder()
.method(Method::POST)
.uri("/register")
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
.header(header::COOKIE, format!("csrf={csrf}"))
.body(Body::from(form))
.unwrap();
let (status, headers, _) = send(router, req).await;
assert_eq!(status, 303, "POST /register");
let session = cookie_value(&headers, "token").expect("session cookie after register");
let cookie = format!("csrf={csrf}; token={session}");
let (status, _, body) = send(
router,
json_request(
Method::POST,
"/api/v1/auth/login",
None,
serde_json::json!({ "email": email, "password": PASSWORD }),
),
)
.await;
assert_eq!(status, 200, "login: {body}");
let login: serde_json::Value = serde_json::from_str(&body).unwrap();
Session {
user_id: login["user_id"].as_str().unwrap().to_string(),
token: login["token"].as_str().unwrap().to_string(),
cookie,
}
}

View File

@@ -21,18 +21,3 @@ tokio = { workspace = true }
anyhow = { workspace = true }
dotenvy = { workspace = true }
tracing = { workspace = true }
[dev-dependencies]
domain = { workspace = true, features = ["test-helpers"] }
application = { workspace = true }
http-axum = { path = "../adapters/http-axum", default-features = false }
rss = { workspace = true }
sqlite = { workspace = true }
sqlx = { workspace = true }
async-trait = { workspace = true }
bytes = { workspace = true }
futures = { workspace = true }
serde_json = { workspace = true }
uuid = { workspace = true }
tower = { version = "0.5", features = ["util"] }
http-body-util = "0.1"

View File

@@ -1,844 +0,0 @@
use std::sync::Arc;
use application::config::AppConfig;
use async_trait::async_trait;
use axum::{
Router,
body::Body,
http::{Request, StatusCode},
};
use bootstrap::Repositories;
use domain::{
errors::DomainError,
events::DomainEvent,
models::{
EntityType, ExternalPersonId, IndexableDocument, MetadataSearchCriteria, Movie, Person,
PersonCredits, PersonEnrichmentData, PersonId, SearchQuery, SearchResults, User,
},
ports::{
AuthService, EventPublisher, MetadataClient, ObjectStorage, PasswordHasher, PersonCommand,
PersonQuery, PosterFetcherClient, SearchCommand, SearchPort, UserRepository,
},
value_objects::{Email, ExternalMetadataId, PasswordHash, PosterUrl, UserId},
};
use http_axum::context::AppContext;
use http_axum::{router, state::AppState};
use http_body_util::BodyExt;
use rss::RssAdapter;
use sqlite::{
SqliteDiaryRepository, SqliteInsightsQuery, SqliteMovieRepository, SqliteReviewRepository,
SqliteStatsRepository, migrate as sqlite_migrate,
};
use sqlx::SqlitePool;
use tower::ServiceExt;
use uuid::Uuid;
struct NoopEventPublisher;
#[async_trait]
impl EventPublisher for NoopEventPublisher {
async fn publish(&self, _: &DomainEvent) -> Result<(), DomainError> {
Ok(())
}
}
struct PanicReviewLogger;
#[async_trait]
impl application::ports::ReviewLogger for PanicReviewLogger {
async fn log_review(
&self,
_: application::diary::commands::LogReviewCommand,
) -> Result<(), DomainError> {
panic!("review_logger not wired in tests")
}
}
struct PanicMeta;
#[async_trait]
impl MetadataClient for PanicMeta {
async fn fetch_movie_metadata(&self, _: &MetadataSearchCriteria) -> Result<Movie, DomainError> {
panic!("metadata not wired in tests")
}
async fn get_poster_url(
&self,
_: &ExternalMetadataId,
) -> Result<Option<PosterUrl>, DomainError> {
panic!()
}
}
struct PanicFetcher;
#[async_trait]
impl PosterFetcherClient for PanicFetcher {
async fn fetch_poster_bytes(&self, _: &PosterUrl) -> Result<Vec<u8>, DomainError> {
panic!()
}
}
struct PanicObjectStorage;
#[async_trait]
impl ObjectStorage for PanicObjectStorage {
async fn store(&self, _: &str, _: &[u8]) -> Result<String, DomainError> {
panic!()
}
async fn get(&self, _: &str) -> Result<Vec<u8>, DomainError> {
panic!()
}
async fn get_stream(
&self,
_: &str,
) -> Result<futures::stream::BoxStream<'static, Result<bytes::Bytes, DomainError>>, DomainError>
{
panic!()
}
async fn delete(&self, _: &str) -> Result<(), DomainError> {
panic!()
}
}
struct PanicHasher;
#[async_trait]
impl PasswordHasher for PanicHasher {
async fn hash(&self, _: &str) -> Result<PasswordHash, DomainError> {
panic!()
}
async fn verify(&self, _: &str, _: &PasswordHash) -> Result<bool, DomainError> {
panic!()
}
}
struct NobodyUserRepo;
#[async_trait]
impl UserRepository for NobodyUserRepo {
async fn find_by_email(&self, _: &Email) -> Result<Option<User>, DomainError> {
Ok(None)
}
async fn find_by_username(
&self,
_: &domain::value_objects::Username,
) -> Result<Option<User>, DomainError> {
Ok(None)
}
async fn save(&self, _: &User) -> Result<(), DomainError> {
panic!()
}
async fn find_by_id(&self, _: &UserId) -> Result<Option<User>, DomainError> {
panic!()
}
async fn list_with_stats(&self) -> Result<Vec<domain::models::UserSummary>, DomainError> {
panic!()
}
async fn update_profile(
&self,
_: &UserId,
_: &domain::models::UserProfile,
) -> Result<(), DomainError> {
Ok(())
}
}
struct PanicProfileFields;
#[async_trait]
impl domain::ports::UserProfileFieldsRepository for PanicProfileFields {
async fn get_fields(
&self,
_: &UserId,
) -> Result<Vec<domain::models::ProfileField>, DomainError> {
Ok(vec![])
}
async fn set_fields(
&self,
_: &UserId,
_: Vec<domain::models::ProfileField>,
) -> Result<(), DomainError> {
panic!()
}
}
struct PanicExporter;
impl domain::ports::DiaryExporter for PanicExporter {
fn stream_entries(
&self,
_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")
}
}
struct PanicImportSession;
#[async_trait]
impl domain::ports::ImportSessionRepository for PanicImportSession {
async fn create(&self, _: &domain::models::ImportSession) -> Result<(), DomainError> {
panic!()
}
async fn get(
&self,
_: &domain::value_objects::ImportSessionId,
_: &UserId,
) -> Result<Option<domain::models::ImportSession>, DomainError> {
panic!()
}
async fn update(&self, _: &domain::models::ImportSession) -> Result<(), DomainError> {
panic!()
}
async fn delete(&self, _: &domain::value_objects::ImportSessionId) -> Result<(), DomainError> {
panic!()
}
async fn delete_expired(&self) -> Result<u64, DomainError> {
panic!()
}
async fn delete_expired_for_user(&self, _: &UserId) -> Result<(), DomainError> {
panic!()
}
}
struct PanicDocumentParser;
impl domain::ports::DocumentParser for PanicDocumentParser {
fn parse(
&self,
_: &[u8],
_: domain::models::FileFormat,
) -> Result<domain::models::ParsedFile, domain::models::ImportError> {
panic!("DocumentParser not wired in tests")
}
fn apply_mapping(
&self,
_: &domain::models::ParsedFile,
_: &[domain::models::FieldMapping],
) -> Vec<domain::models::AnnotatedRow> {
panic!("DocumentParser not wired in tests")
}
}
struct PanicImportProfile;
struct PanicMovieProfile;
#[async_trait]
impl domain::ports::MovieProfileRepository for PanicMovieProfile {
async fn upsert(&self, _: &domain::models::MovieProfile) -> Result<(), DomainError> {
panic!()
}
async fn get_by_movie_id(
&self,
_: &domain::value_objects::MovieId,
) -> Result<Option<domain::models::MovieProfile>, DomainError> {
Ok(None)
}
async fn list_stale(
&self,
) -> Result<Vec<(domain::value_objects::MovieId, String)>, DomainError> {
Ok(vec![])
}
}
#[async_trait]
impl domain::ports::ImportProfileRepository for PanicImportProfile {
async fn save(&self, _: &domain::models::ImportProfile) -> Result<(), DomainError> {
panic!()
}
async fn list_for_user(
&self,
_: &UserId,
) -> Result<Vec<domain::models::ImportProfile>, DomainError> {
panic!()
}
async fn get(
&self,
_: &domain::value_objects::ImportProfileId,
_: &UserId,
) -> Result<Option<domain::models::ImportProfile>, DomainError> {
panic!()
}
async fn delete(&self, _: &domain::value_objects::ImportProfileId) -> Result<(), DomainError> {
panic!()
}
}
struct PanicWatchlist;
#[async_trait]
impl domain::ports::WatchlistRepository for PanicWatchlist {
async fn add(&self, _: &domain::models::WatchlistEntry) -> Result<(), DomainError> {
panic!()
}
async fn remove(
&self,
_: &domain::value_objects::UserId,
_: &domain::value_objects::MovieId,
) -> Result<(), DomainError> {
panic!()
}
async fn remove_if_present(
&self,
_: &domain::value_objects::UserId,
_: &domain::value_objects::MovieId,
) -> Result<bool, DomainError> {
Ok(false)
}
async fn get_for_user(
&self,
_: &domain::value_objects::UserId,
_: &domain::models::collections::PageParams,
) -> Result<
domain::models::collections::Paginated<domain::models::WatchlistWithMovie>,
DomainError,
> {
panic!()
}
async fn contains(
&self,
_: &domain::value_objects::UserId,
_: &domain::value_objects::MovieId,
) -> Result<bool, DomainError> {
Ok(false)
}
}
struct PanicPersonCommand;
#[async_trait]
impl PersonCommand for PanicPersonCommand {
async fn upsert_batch(&self, _: &[Person]) -> Result<(), DomainError> {
panic!()
}
async fn backfill_from_credits_batch(
&self,
_batch_size: u32,
) -> Result<(u64, bool), DomainError> {
panic!()
}
async fn update_enrichment(
&self,
_: &PersonId,
_: &PersonEnrichmentData,
) -> Result<(), DomainError> {
panic!()
}
}
struct PanicPersonQuery;
#[async_trait]
impl PersonQuery for PanicPersonQuery {
async fn get_by_id(&self, _: &PersonId) -> Result<Option<Person>, DomainError> {
panic!()
}
async fn get_by_external_id(
&self,
_: &ExternalPersonId,
) -> Result<Option<Person>, DomainError> {
panic!()
}
async fn get_credits(&self, _: &PersonId) -> Result<PersonCredits, DomainError> {
panic!()
}
async fn list_orphaned_persons(&self) -> Result<Vec<PersonId>, DomainError> {
panic!()
}
async fn list_page(
&self,
_limit: u32,
_offset: u32,
) -> Result<Vec<domain::models::Person>, DomainError> {
panic!()
}
}
struct PanicSearchPort;
#[async_trait]
impl SearchPort for PanicSearchPort {
async fn search(&self, _: &SearchQuery) -> Result<SearchResults, DomainError> {
panic!()
}
}
struct PanicSearchCommand;
#[async_trait]
impl SearchCommand for PanicSearchCommand {
async fn index(&self, _: IndexableDocument) -> Result<(), DomainError> {
panic!()
}
async fn remove(&self, _: EntityType, _: &str) -> Result<(), DomainError> {
panic!()
}
}
struct PanicRemoteWatchlist;
#[async_trait::async_trait]
impl domain::ports::RemoteWatchlistRepository for PanicRemoteWatchlist {
async fn save(&self, _: domain::models::RemoteWatchlistEntry) -> Result<(), DomainError> {
Ok(())
}
async fn remove_by_ap_id(&self, _: &str, _: &str) -> Result<(), DomainError> {
Ok(())
}
async fn get_by_actor_url(
&self,
_: &str,
) -> Result<Vec<domain::models::RemoteWatchlistEntry>, DomainError> {
Ok(vec![])
}
async fn remove_all_by_actor(&self, _: &str) -> Result<(), DomainError> {
Ok(())
}
async fn get_by_derived_uuid(
&self,
_: uuid::Uuid,
) -> Result<Vec<domain::models::RemoteWatchlistEntry>, DomainError> {
Ok(vec![])
}
}
async fn test_app() -> Router {
test_app_with(domain::testing::FakeAuthService).await.0
}
async fn test_app_with(auth: impl AuthService + 'static) -> (Router, SqlitePool) {
let pool = SqlitePool::connect("sqlite::memory:")
.await
.expect("in-memory SQLite failed");
sqlite_migrate(&pool).await.expect("migration failed");
let repos = Repositories {
movie_command: Arc::new(SqliteMovieRepository::new(pool.clone())) as _,
movie_query: Arc::new(SqliteMovieRepository::new(pool.clone())) as _,
review: Arc::new(SqliteReviewRepository::new(pool.clone())) as _,
diary: Arc::new(SqliteDiaryRepository::new(pool.clone())) as _,
stats: Arc::new(SqliteStatsRepository::new(pool.clone())) as _,
user: Arc::new(NobodyUserRepo),
import_session: Arc::new(PanicImportSession),
import_profile: Arc::new(PanicImportProfile),
movie_profile: Arc::new(PanicMovieProfile),
watchlist: Arc::new(PanicWatchlist),
watch_event_command: Arc::new(domain::testing::PanicWatchEventCommand),
watch_event_query: Arc::new(domain::testing::PanicWatchEventQuery),
webhook_token: Arc::new(domain::testing::PanicWebhookTokenRepository),
profile_fields: Arc::new(PanicProfileFields),
person_command: Arc::new(PanicPersonCommand),
person_query: Arc::new(PanicPersonQuery),
search_port: Arc::new(PanicSearchPort),
search_command: Arc::new(PanicSearchCommand),
remote_watchlist: Arc::new(PanicRemoteWatchlist),
social_command: Arc::new(domain::ports::noop::NoopSocialCommand),
follow_graph: Arc::new(domain::ports::noop::NoopSocialQuery),
block_query: Arc::new(domain::ports::noop::NoopSocialQuery),
federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery) as _,
wrapup_stats: Arc::new(domain::testing::PanicWrapUpStatsQuery) as _,
wrapup_repo: Arc::new(domain::testing::PanicWrapUpRepository) as _,
insights: Arc::new(SqliteInsightsQuery::new(pool.clone())) as _,
goal_command: Arc::new(domain::testing::NoopGoalCommand),
goal_query: Arc::new(domain::testing::NoopGoalQuery),
user_settings: Arc::new(domain::testing::NoopUserSettingsRepository),
remote_goal: Arc::new(domain::testing::NoopRemoteGoalRepository),
refresh_session: Arc::new(domain::testing::PanicRefreshSessionRepository),
federated_profile: None,
};
let services = application::Services {
auth: Arc::new(auth),
password_hasher: Arc::new(PanicHasher),
metadata: Arc::new(PanicMeta),
poster_fetcher: Arc::new(PanicFetcher),
object_storage: Arc::new(PanicObjectStorage),
event_publisher: Arc::new(NoopEventPublisher),
diary_exporter: Arc::new(PanicExporter),
document_parser: Arc::new(PanicDocumentParser),
review_logger: Arc::new(PanicReviewLogger),
person_enrichment: None,
};
let config = AppConfig {
allow_registration: false,
base_url: "http://localhost:3000".to_string(),
rate_limit: 20,
refresh_ttl_seconds: 2_592_000,
secure_cookies: false,
cors_origins: String::new(),
wrapup: application::config::WrapUpConfig {
font_path: None,
logo_path: None,
bg_dir: None,
},
};
let instance = domain::value_objects::InstanceIdentity::new("http://localhost:3000");
let deps = Arc::new(bootstrap::build_deps(&repos, &services, &config, &instance));
let state = AppState {
app_ctx: AppContext {
deps,
services,
config,
instance,
#[cfg(feature = "federation")]
ap_document: Arc::new(domain::ports::noop::NoopApDocument),
#[cfg(feature = "federation")]
ap_blocklist: Arc::new(domain::ports::noop::NoopInstanceBlocklist),
},
rss_renderer: Arc::new(RssAdapter::new("http://localhost:3000".into())),
};
let router = router::build_router(state, axum::Router::new());
(router, pool)
}
/// Inject a fake peer IP so the GovernorLayer can extract ConnectInfo.
fn with_ip(req: Request<Body>) -> Request<Body> {
let addr: std::net::SocketAddr = "127.0.0.1:12345".parse().unwrap();
let mut req = req;
req.extensions_mut()
.insert(axum::extract::ConnectInfo::<std::net::SocketAddr>(addr));
req
}
#[tokio::test]
async fn get_api_diary_returns_empty_list() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/diary")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let bytes = response.into_body().collect().await.unwrap().to_bytes();
let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(json["total_count"], 0);
assert_eq!(json["items"], serde_json::json!([]));
assert_eq!(json["limit"], 5);
assert_eq!(json["offset"], 0);
}
#[tokio::test]
async fn post_api_reviews_without_auth_returns_401() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.method("POST")
.uri("/api/v1/reviews")
.header("content-type", "application/json")
.body(Body::from(
r#"{"rating":4,"watched_at":"2026-01-01T20:00:00","manual_title":"Dune","manual_release_year":2021}"#,
))
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn post_api_auth_login_unknown_user_returns_401() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.method("POST")
.uri("/api/v1/auth/login")
.header("content-type", "application/json")
.body(Body::from(r#"{"email":"a@b.com","password":"x"}"#))
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[tokio::test]
async fn get_api_movie_detail_returns_404_for_unknown_id() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/movies/00000000-0000-0000-0000-000000000000")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn tags_moviesdiary_redirects_to_home() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/tags/moviesdiary")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::TEMPORARY_REDIRECT);
assert_eq!(response.headers().get("location").unwrap(), "/");
}
#[tokio::test]
async fn tags_other_redirects_to_search() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/tags/batman")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::TEMPORARY_REDIRECT);
assert_eq!(
response.headers().get("location").unwrap(),
"/?search=batman"
);
}
#[tokio::test]
async fn get_movie_detail_html_returns_404_for_unknown_id() {
let app = test_app().await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/movies/00000000-0000-0000-0000-000000000000")
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
// ── Insights ─────────────────────────────────────────────────────────────────
async fn seed_insights_diary(pool: &SqlitePool, user_id: Uuid) {
// 12 thrillers rated 5 late at night, 12 comedies rated 2 — enough for a
// rating lift once genres come from movie_genres.
for i in 0..12u32 {
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Thriller {i}"))
.bind(2020)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(44400 + i as i64)
.bind("Thriller")
.execute(pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at) \
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(5)
.bind("2024-01-05 22:30:00")
.bind("2024-01-05 22:30:00")
.execute(pool)
.await
.unwrap();
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Comedy {i}"))
.bind(2015)
.execute(pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(55300 + i as i64)
.bind("Comedy")
.execute(pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at) \
VALUES (?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(2)
.bind("2024-03-10 13:00:00")
.bind("2024-03-10 13:00:00")
.execute(pool)
.await
.unwrap();
}
}
#[tokio::test]
async fn get_api_insights_returns_rating_lift_for_own_user() {
let (app, pool) = test_app_with(domain::testing::FakeAuthService).await;
let user_id = Uuid::new_v4();
seed_insights_diary(&pool, user_id).await;
let response = app
.oneshot(with_ip(
Request::builder()
.uri("/api/v1/insights")
.header("authorization", format!("Bearer {user_id}"))
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
let (status, bytes) = (
response.status(),
response.into_body().collect().await.unwrap().to_bytes(),
);
assert_eq!(
status,
StatusCode::OK,
"body: {}",
String::from_utf8(bytes.to_vec()).unwrap()
);
let json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
let rating_lifts: Vec<&serde_json::Value> = json["insights"]
.as_array()
.unwrap()
.iter()
.filter(|i| i["kind"] == "rating_lift")
.collect();
assert_eq!(rating_lifts.len(), 2, "thriller up, comedy down: {json}");
let thriller = rating_lifts
.iter()
.find(|i| i["genre"] == "Thriller")
.unwrap();
assert_eq!(thriller["delta"], 1.5);
assert_eq!(thriller["sample_count"], 12);
// Timestamps are real → rhythm fires. Hours split evenly between 22
// (thrillers) and 13 (comedies); every window covering either bucket ties,
// so the earliest one, 11:00–14:00, wins. Days tie too (Fri/Sun), sorted
// ascending.
let rhythm = json["insights"]
.as_array()
.unwrap()
.iter()
.find(|i| i["kind"] == "rhythm")
.expect("rhythm expected for real timestamps");
assert_eq!(rhythm["peak_hour_start"], 11);
assert_eq!(rhythm["peak_hour_end"], 14);
assert_eq!(rhythm["top_days"], serde_json::json!([5, 7]));
}
async fn get_insights_json(
app: Router,
user_id: Uuid,
query: &str,
) -> (StatusCode, serde_json::Value) {
let response = app
.oneshot(with_ip(
Request::builder()
.uri(format!("/api/v1/insights{query}"))
.header("authorization", format!("Bearer {user_id}"))
.body(Body::empty())
.unwrap(),
))
.await
.unwrap();
let status = response.status();
let bytes = response.into_body().collect().await.unwrap().to_bytes();
let json = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
(status, json)
}
#[tokio::test]
async fn get_api_insights_ignores_remote_reviews() {
let (app, pool) = test_app_with(domain::testing::FakeAuthService).await;
let user_id = Uuid::new_v4();
seed_insights_diary(&pool, user_id).await;
// Federated rows stored under the same user_id are someone else's watches.
for i in 0..12u32 {
let movie_id = uuid::Uuid::new_v4().to_string();
sqlx::query("INSERT INTO movies (id, title, release_year) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(format!("Horror {i}"))
.bind(2019)
.execute(&pool)
.await
.unwrap();
sqlx::query("INSERT INTO movie_genres (movie_id, tmdb_id, name) VALUES (?, ?, ?)")
.bind(&movie_id)
.bind(66600 + i as i64)
.bind("Horror")
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at, remote_actor_url) \
VALUES (?, ?, ?, ?, ?, ?, ?)",
)
.bind(uuid::Uuid::new_v4().to_string())
.bind(&movie_id)
.bind(user_id.to_string())
.bind(5)
.bind("2024-05-05 09:00:00")
.bind("2024-05-05 09:00:00")
.bind("https://remote.example/users/someone")
.execute(&pool)
.await
.unwrap();
}
let (status, json) = get_insights_json(app, user_id, "").await;
assert_eq!(status, StatusCode::OK);
let insights = json["insights"].as_array().unwrap();
assert!(
!insights.iter().any(|i| i["genre"] == "Horror"),
"remote reviews must not feed insights: {json}"
);
let thriller = insights
.iter()
.find(|i| i["kind"] == "rating_lift" && i["genre"] == "Thriller")
.unwrap();
assert_eq!(thriller["delta"], 1.5);
}
#[tokio::test]
async fn get_api_insights_buckets_rhythm_in_requested_time_zone() {
let (app, pool) = test_app_with(domain::testing::FakeAuthService).await;
let user_id = Uuid::new_v4();
seed_insights_diary(&pool, user_id).await;
// January in Warsaw is UTC+1: 22:30 → 23:30 and 13:00 → 14:00, so the
// earliest tied window moves from 11–14 to 12–15.
let (status, json) = get_insights_json(app, user_id, "?tz=Europe/Warsaw").await;
assert_eq!(status, StatusCode::OK);
let rhythm = json["insights"]
.as_array()
.unwrap()
.iter()
.find(|i| i["kind"] == "rhythm")
.expect("rhythm expected");
assert_eq!(rhythm["peak_hour_start"], 12);
assert_eq!(rhythm["peak_hour_end"], 15);
}
#[tokio::test]
async fn get_api_insights_rejects_unknown_time_zone() {
let (app, _pool) = test_app_with(domain::testing::FakeAuthService).await;
let (status, _) = get_insights_json(app, Uuid::new_v4(), "?tz=Mars/Olympus").await;
assert_eq!(status, StatusCode::BAD_REQUEST);
}