GKaszewski/movies-diary
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: always-on Insights — correlation analysis over viewing habits
Some checks failed
CI / Check / Test (push) Failing after 9m16s
Details

Browse Source 
Conservative, live insights (ADR-0010) derived from each user's diary:
rating lift by genre (first-time watches only), genre seasonality vs a
uniform 1/12 month baseline, medium weekend-lift, and watch-time rhythm.

- domain: WatchRow model, InsightsQuery port, pure insights_analyzer with
  all honesty thresholds (sample minimums, strength cutoffs, suppressions
  for import noise), unit-tested
- adapters: thin Sqlite/Postgres fetch twins, reusing WrapUp's genre fetch
- application: get_insights use case, InsightsGroup in Deps
- composition/server: wiring through Repositories/DatabaseOutput
- presentation: GET /api/v1/insights handler, typed serde-tagged DTOs,
  openapi doc
- spa: /insights route (shadcn cards), insights feature hook, profile
  entry link, en locale strings
- server test: end-to-end seed sqlite → HTTP → rating lift + rhythm

Also records ADR-0010, defines "Insight" in the glossary, and includes
pre-existing worktree comment-stripping cleanups.
This commit is contained in:
Gabriel Kaszewski
2026-10-12 19:29:54 +02:00
parent 4d221e60de
commit 020fbb891e
126 changed files with 1922 additions and 639 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/adapters/activitypub/src/federation_event_bridge.rs
View File

@@ -64,7 +64,6 @@ impl k_ap::EventPublisher for FederationEventBridge {
if let Some(outbox) = outbox_url {
tracing::info!(outbox = %outbox, "importing remote outbox after follow accepted");
// Handled by FollowBackfillHandler reacting to FollowAccepted
}
Ok(())
}

13
crates/adapters/activitypub/src/federation_ports.rs
View File

@@ -9,15 +9,6 @@ use domain::{
use k_ap::ActivityPubService;
use uuid::Uuid;
/// Adapts the federation library's service to the three domain-owned ports.
///
/// A wrapper rather than a bare `impl ... for ActivityPubService` because the
/// orphan rule forbids implementing a foreign trait for a foreign type, and
/// from this crate both `ApDocumentPort` (owned by `domain`) and
/// `ActivityPubService` (owned by the external `k-ap` crate) are foreign.
///
/// One type carrying all three impls mirrors `CompositeSocialAdapter`, which
/// serves `SocialCommand`, `FollowGraphQuery`, and `BlockQuery` the same way.
pub struct ApServiceAdapter {
service: Arc<ActivityPubService>,
}
@@ -28,10 +19,6 @@ impl ApServiceAdapter {
}
}
/// Single conversion point from the federation library's `anyhow` errors to
/// `DomainError`. The log line and the resulting error string reproduce what
/// `presentation::handlers::social::ap_to_domain` produced before the port
/// inversion moved the boundary here.
fn ap_err(e: anyhow::Error) -> DomainError {
tracing::error!("ActivityPub error: {:?}", e);
DomainError::InfrastructureError(e.to_string())

7
crates/adapters/activitypub/src/lib.rs
View File

@@ -14,7 +14,6 @@ pub mod watchlist_handler;
pub const INSTANCE_ACTOR_ID: uuid::Uuid =
uuid::Uuid::from_bytes([0, 0, 0, 0, 0, 0, 0x40, 0, 0x80, 0, 0, 0, 0, 0, 0, 0]);
// Re-export the generic base types that callers need
pub use k_ap::{
ActivityPubService, ActivityRepository, ActorRepository, ApContentReader, ApFederationConfig,
ApObjectHandler, ApUser, ApUserRepository, BlocklistRepository, FederationData,
@@ -41,15 +40,9 @@ pub struct FederationRepos {
}
pub struct ActivityPubWire {
/// AP document serving. Prefer this over `service` from outside this crate.
pub document: std::sync::Arc<dyn domain::ports::ApDocumentPort>,
/// Instance domain blocklist. Prefer this over `service` from outside this crate.
pub blocklist: std::sync::Arc<dyn domain::ports::InstanceBlocklistPort>,
/// Post-follow content backfill. Prefer this over `service` from outside this crate.
pub backfill: std::sync::Arc<dyn domain::ports::ApBackfillPort>,
/// The concrete service, consumed by `crates/server` to construct
/// `CompositeSocialAdapter`. Everything else should use
/// `document`/`blocklist`/`backfill`.
pub service: std::sync::Arc<ActivityPubService>,
pub router: axum::Router,
pub event_handler: std::sync::Arc<dyn domain::ports::EventHandler>,

3
crates/adapters/activitypub/src/urls.rs
View File

@@ -1,13 +1,11 @@
use domain::value_objects::{InstanceIdentity, ReviewId, UserId};
use url::Url;
/// Builds the canonical actor URL: `{base_url}/users/{user_id}`
pub fn actor_url(instance: &InstanceIdentity, user_id: uuid::Uuid) -> Url {
Url::parse(&instance.actor_url_for(&UserId::from_uuid(user_id)))
.expect("base_url is always a valid URL prefix")
}
/// Builds the canonical review URL: `{base_url}/reviews/{review_id}`
pub fn review_url(instance: &InstanceIdentity, review_id: &ReviewId) -> Url {
Url::parse(&format!(
"{}/reviews/{}",
@@ -27,7 +25,6 @@ pub fn goal_url(instance: &InstanceIdentity, user_id: uuid::Uuid, year: u16) ->
.expect("base_url is always a valid URL prefix")
}
/// Builds the canonical watchlist entry URL: `{base_url}/users/{user_id}/watchlist/{movie_id}`
pub fn watchlist_entry_url(
instance: &InstanceIdentity,
user_id: uuid::Uuid,

27
crates/adapters/adapter-common/src/lib.rs
View File

@@ -10,31 +10,25 @@ use domain::{
},
};
/// Map a [`sqlx::Error`] to a [`DomainError::InfrastructureError`], logging the
/// underlying database error at `error` level.
pub fn map_sqlx_error(e: sqlx::Error) -> DomainError {
tracing::error!("Database error: {:?}", e);
DomainError::InfrastructureError("Database operation failed".into())
}
/// Parse a string as a UUID, returning a [`DomainError`] on failure.
pub fn parse_uuid(s: &str) -> Result<uuid::Uuid, DomainError> {
uuid::Uuid::parse_str(s)
.map_err(|e| DomainError::InfrastructureError(format!("Invalid UUID '{}': {}", s, e)))
}
/// Parse a `%Y-%m-%d %H:%M:%S` string into a [`chrono::NaiveDateTime`].
pub fn parse_datetime(s: &str) -> Result<chrono::NaiveDateTime, DomainError> {
chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S")
.map_err(|e| DomainError::InfrastructureError(format!("Invalid datetime '{}': {}", s, e)))
}
/// Format a [`chrono::NaiveDateTime`] as `%Y-%m-%d %H:%M:%S`.
pub fn datetime_to_str(dt: &chrono::NaiveDateTime) -> String {
dt.format("%Y-%m-%d %H:%M:%S").to_string()
}
/// Convert a `YYYY-MM` string into a human-readable label like `Jan '24`.
pub fn format_year_month(ym: &str) -> String {
let parts: Vec<&str> = ym.splitn(2, '-').collect();
if parts.len() != 2 {
@@ -59,16 +53,6 @@ pub fn format_year_month(ym: &str) -> String {
format!("{} '{}", month, year)
}
// ---------------------------------------------------------------------------
// Shared row-to-domain conversion functions
//
// Each database adapter keeps its own `FromRow` structs (sqlite vs postgres
// derive different impls) but the conversion from parsed row fields into
// domain types is identical. These functions capture that shared logic so
// each adapter's `into_domain()` becomes a one-liner delegation.
// ---------------------------------------------------------------------------
/// Convert raw movie row fields into a [`Movie`] domain object.
pub fn movie_row_to_domain(
id: String,
external_metadata_id: Option<String>,
@@ -94,7 +78,6 @@ pub fn movie_row_to_domain(
))
}
/// Convert raw review row fields into a [`Review`] domain object.
#[allow(clippy::too_many_arguments)]
pub fn review_row_to_domain(
id: String,
@@ -132,10 +115,6 @@ pub fn review_row_to_domain(
}))
}
/// Assemble a [`MovieSummary`] from an already-converted [`Movie`] and extra
/// metadata fields. The caller is responsible for converting genres into a
/// `Vec<String>` (sqlite splits a comma-separated string, postgres receives a
/// `Vec` directly).
pub fn movie_summary_to_domain(
movie: Movie,
genres: Vec<String>,
@@ -154,7 +133,6 @@ pub fn movie_summary_to_domain(
}
}
/// Convert raw aggregate stats into a [`MovieStats`] domain object.
pub fn movie_stats_to_domain(
total_count: i64,
avg_rating: Option<f64>,
@@ -175,7 +153,6 @@ pub fn movie_stats_to_domain(
}
}
/// Convert raw user summary row fields into a [`UserSummary`] domain object.
#[allow(clippy::too_many_arguments)]
pub fn user_summary_to_domain(
id: String,
@@ -197,7 +174,6 @@ pub fn user_summary_to_domain(
))
}
/// Convert raw watchlist entry fields into a [`WatchlistEntry`] domain object.
pub fn watchlist_entry_to_domain(
id: String,
user_id: String,
@@ -212,9 +188,6 @@ pub fn watchlist_entry_to_domain(
})
}
/// Convert raw watchlist+movie row fields into a [`WatchlistWithMovie`].
///
/// Takes the watchlist entry fields and a pre-converted [`Movie`].
pub fn watchlist_with_movie_to_domain(entry: WatchlistEntry, movie: Movie) -> WatchlistWithMovie {
WatchlistWithMovie { entry, movie }
}

2
crates/adapters/image-converter/src/backfill.rs
View File

@@ -27,7 +27,7 @@ impl ConversionBackfillJob {
#[async_trait]
impl PeriodicJob for ConversionBackfillJob {
fn interval(&self) -> Duration {
Duration::from_secs(60 * 60 * 24) // 24h
Duration::from_secs(60 * 60 * 24)
}
async fn run(&self) -> Result<(), DomainError> {

2
crates/adapters/image-converter/src/tests/handler.rs
View File

@@ -130,8 +130,6 @@ async fn converts_jpeg_to_avif_and_swaps_key() {
vec![("avatars/u1".into(), "avatars/u1.avif".into())]
);
assert!(storage.get("avatars/u1.avif").await.is_ok());
// Old raw key deleted — fallback resolves to .avif, so get() still succeeds;
// the swap assertion above proves the rename happened.
}
#[tokio::test]

2
crates/adapters/importer/src/parsers/xlsx.rs
View File

@@ -58,14 +58,12 @@ fn cell_to_string(cell: &Data) -> String {
Data::Int(i) => i.to_string(),
Data::Bool(b) => b.to_string(),
Data::DateTime(dt) => {
// ExcelDateTime::to_ymd_hms_milli() works without the chrono feature.
let (year, month, day, _, _, _, _) = dt.to_ymd_hms_milli();
format!("{:04}-{:02}-{:02}", year, month, day)
}
Data::DateTimeIso(s) => s.clone(),
Data::DurationIso(s) => s.clone(),
Data::Empty | Data::Error(_) => String::new(),
// Fallback for unexpected calamine Data variants; renders as debug string
other => format!("{other:?}"),
}
}

9
crates/adapters/importer/src/tests/mapper.rs
View File

@@ -7,7 +7,7 @@ fn sample_file() -> ParsedFile {
rows: vec![
vec!["Inception".into(), "10".into(), "2024-01-15".into()],
vec!["Dune".into(), "8".into(), "2024-02-20".into()],
vec!["".into(), "3".into(), "2024-03-01".into()], // missing title → invalid
vec!["".into(), "3".into(), "2024-03-01".into()],
],
}
}
@@ -36,10 +36,8 @@ fn full_mappings() -> Vec<FieldMapping> {
fn maps_valid_rows() {
let results = apply_mapping(&sample_file(), &full_mappings());
assert_eq!(results.len(), 3);
// First two rows are valid
assert!(matches!(results[0].result, RowResult::Valid(_)));
assert!(matches!(results[1].result, RowResult::Valid(_)));
// is_duplicate defaults to false
assert!(!results[0].is_duplicate);
}
@@ -47,7 +45,6 @@ fn maps_valid_rows() {
fn applies_rating_scale_transform() {
let results = apply_mapping(&sample_file(), &full_mappings());
if let RowResult::Valid(row) = &results[0].result {
// 10 * 0.5 = 5
assert_eq!(row.rating.as_deref(), Some("5"));
} else {
panic!("expected Valid");
@@ -57,7 +54,6 @@ fn applies_rating_scale_transform() {
#[test]
fn marks_missing_required_fields_invalid() {
let results = apply_mapping(&sample_file(), &full_mappings());
// Row 2 has empty title
assert!(matches!(results[2].result, RowResult::Invalid { .. }));
}
@@ -74,7 +70,6 @@ fn ignores_unmapped_columns() {
};
let results = apply_mapping(&file, &mappings);
assert_eq!(results.len(), 1);
// Missing rating and watched_at → invalid
assert!(matches!(results[0].result, RowResult::Invalid { .. }));
}
@@ -90,7 +85,6 @@ fn nonexistent_source_column_skipped() {
rows: vec![vec!["Inception".into()]],
};
let results = apply_mapping(&file, &mappings);
// Column not found → field not set → invalid (missing title, rating, watched_at)
assert!(matches!(results[0].result, RowResult::Invalid { .. }));
}
@@ -107,7 +101,6 @@ fn collects_all_errors_not_just_first() {
domain_field: DomainField::Rating,
transform: Transform::RatingScale(0.5),
},
// no watched_at mapping
];
let file = ParsedFile {
columns: vec!["Name".into(), "Stars".into()],

3
crates/adapters/metadata/src/tmdb.rs
View File

@@ -68,7 +68,7 @@ impl TmdbProvider {
struct Details {
imdb_id: Option<String>,
title: String,
release_date: String, // "YYYY-MM-DD"
release_date: String,
poster_path: Option<String>,
credits: Credits,
}
@@ -88,7 +88,6 @@ impl TmdbProvider {
))
})?;
// Prefer IMDB ID; fall back to "tmdb:{id}" so the record is still usable.
let raw_id = d
.imdb_id
.filter(|s| !s.is_empty())

1
crates/adapters/nats/src/consumer.rs
View File

@@ -153,7 +153,6 @@ impl NatsJetStreamConsumer {
tracing::debug!("message sent to consumer channel");
}
// messages() stream ended (fetch expired in strict mode) — restart
}
});

1
crates/adapters/object-storage/src/lib.rs
View File

@@ -55,7 +55,6 @@ impl ObjectStorage for ObjectStorageAdapter {
Err(DomainError::NotFound(_)) if !has_image_ext(key) => {}
Err(e) => return Err(e),
}
// Key may reference a pre-conversion path; try converted extensions.
for ext in [".webp", ".avif"] {
let candidate = format!("{key}{ext}");
if let Ok(bytes) = self.get_exact(&candidate).await {

3
crates/adapters/plex/src/lib.rs
View File

@@ -4,9 +4,6 @@ use serde::Deserialize;
pub struct PlexParser;
impl MediaServerParser for PlexParser {
/// Plex sends multipart form data with a `payload` JSON field.
/// The caller must extract the JSON string from the multipart body
/// and pass it here as raw bytes.
fn parse_playback_event(
&self,
body: &[u8],

1
crates/adapters/poster-sync/src/lib.rs
View File

@@ -116,7 +116,6 @@ impl EventHandler for PosterSyncHandler {
movie_id,
external_metadata_id,
} => {
// Only sync poster if the movie doesn't have one yet
let already_has_poster = self
.movie_query
.get_movie_by_id(&MovieId::from_uuid(movie_id.value()))

2
crates/adapters/postgres-event-queue/src/lib.rs
View File

@@ -114,7 +114,6 @@ impl EventConsumer for PostgresEventQueue {
return;
}
}
// no sleep — re-poll immediately when batch was non-empty
}
}
}
@@ -140,7 +139,6 @@ async fn claim_batch(
pool: &PgPool,
config: &DbEventQueueConfig,
) -> Result<Vec<QueueRow>, DomainError> {
// CTE with FOR UPDATE SKIP LOCKED — atomic and safe for multiple workers
let rows = sqlx::query_as::<_, QueueRow>(
r#"
WITH claimed AS (

8
crates/adapters/postgres-social/src/lib.rs
View File

@@ -11,14 +11,6 @@ pub use remote_goals::PostgresRemoteGoalRepository;
use sqlx::PgPool;
/// Postgres-backed implementations of the *domain* social ports.
///
/// Deliberately separate from `postgres-federation`: this crate knows nothing
/// about ActivityPub, which is what allows a build with the `federation`
/// feature off to exclude the federation stack entirely. See ADR-0009.
///
/// Shares the `ap_followers` / `ap_following` tables with
/// `postgres-federation`; neither crate owns migrations.
pub struct PostgresSocialRepository {
pub(crate) pool: PgPool,
pub(crate) instance: domain::value_objects::InstanceIdentity,

5
crates/adapters/postgres/src/diary.rs
View File

@@ -276,7 +276,6 @@ impl DiaryQuery for PostgresDiaryRepository {
let offset = page.offset as i64;
let has_search = search.map(|s| !s.is_empty()).unwrap_or(false);
// Dynamic param counter
let mut p: i32 = 0;
let mut next_param = || {
p += 1;
@@ -323,9 +322,6 @@ impl DiaryQuery for PostgresDiaryRepository {
let where_clause = where_parts.join(" AND ");
// Reset counter for count query (reuse same where_clause string but re-bind)
// We need a separate counter for count SQL — but since where_clause is already built
// with the right $N references, both queries share it.
let count_sql = format!(
"SELECT COUNT(*) FROM reviews r
INNER JOIN movies m ON m.id = r.movie_id
@@ -352,7 +348,6 @@ impl DiaryQuery for PostgresDiaryRepository {
where_clause, order_clause, limit_param, offset_param
);
// Bind helper closure — binds search + following params in order
macro_rules! bind_filter_params {
($q:expr) => {{
let mut q = $q;

81
crates/adapters/postgres/src/insights.rs Normal file
View File

@@ -0,0 +1,81 @@
use std::collections::HashMap;
use async_trait::async_trait;
use domain::{
errors::DomainError, models::insights::WatchRow, ports::InsightsQuery, value_objects::UserId,
};
use sqlx::{PgPool, Row};
use adapter_common::{parse_datetime, parse_uuid};
use crate::wrapup::fetch_genres_pg;
/// Postgres twin of `SqliteInsightsQuery` — same shape, `$n` binds, and the
/// same thin "rows in, rows out" contract (ADR-0010).
pub struct PostgresInsightsQuery {
pool: PgPool,
}
impl PostgresInsightsQuery {
pub fn new(pool: PgPool) -> Self {
Self { pool }
}
}
#[async_trait]
impl InsightsQuery for PostgresInsightsQuery {
async fn get_watch_rows(&self, user_id: &UserId) -> Result<Vec<WatchRow>, DomainError> {
let rows = sqlx::query(
"SELECT movie_id, rating, \
to_char(watched_at AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS') AS watched_at, \
watch_medium \
FROM reviews \
WHERE user_id = $1 \
ORDER BY watched_at ASC",
)
.bind(user_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(adapter_common::map_sqlx_error)?;
let mut movie_ids: Vec<String> = Vec::new();
let mut seen = std::collections::HashSet::new();
for row in &rows {
let mid: String = row
.try_get("movie_id")
.map_err(adapter_common::map_sqlx_error)?;
if seen.insert(mid.clone()) {
movie_ids.push(mid);
}
}
let genres_map: HashMap<String, Vec<String>> =
fetch_genres_pg(&self.pool, &movie_ids).await?;
let mut result = Vec::with_capacity(rows.len());
for row in &rows {
let movie_id: String = row
.try_get("movie_id")
.map_err(adapter_common::map_sqlx_error)?;
let rating: i64 = row
.try_get("rating")
.map_err(adapter_common::map_sqlx_error)?;
let watched_at: String = row
.try_get("watched_at")
.map_err(adapter_common::map_sqlx_error)?;
let watch_medium: Option<String> = row
.try_get("watch_medium")
.map_err(adapter_common::map_sqlx_error)?;
result.push(WatchRow {
movie_id: parse_uuid(&movie_id)?,
rating: rating as u8,
watched_at: parse_datetime(&watched_at)?,
watch_medium,
genres: genres_map.get(&movie_id).cloned().unwrap_or_default(),
});
}
Ok(result)
}
}

4
crates/adapters/postgres/src/lib.rs
View File

@@ -6,6 +6,7 @@ mod goals;
mod image_ref;
mod import_profile;
mod import_session;
mod insights;
mod models;
mod movie;
mod movie_dedup;
@@ -25,6 +26,7 @@ pub use diary::PostgresDiaryRepository;
pub use image_ref::{PostgresImageRefAdapter, create_image_ref};
pub use import_profile::PostgresImportProfileRepository;
pub use import_session::PostgresImportSessionRepository;
pub use insights::PostgresInsightsQuery;
pub use movie::PostgresMovieRepository;
pub use movie_dedup::PostgresMovieDeduplicator;
pub use persons::{PostgresPersonAdapter, create_person_adapter};
@@ -68,6 +70,7 @@ pub struct PostgresWireOutput {
pub ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
pub wrapup_repo: std::sync::Arc<dyn domain::ports::WrapUpRepository>,
pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
pub insights: std::sync::Arc<dyn domain::ports::InsightsQuery>,
pub goal_command: std::sync::Arc<dyn domain::ports::GoalCommand>,
pub goal_query: std::sync::Arc<dyn domain::ports::GoalQuery>,
pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
@@ -111,6 +114,7 @@ pub async fn wire(database_url: &str) -> anyhow::Result<PostgresWireOutput> {
ap_content: std::sync::Arc::new(PostgresApContentQuery::new(pool.clone())) as _,
wrapup_repo: std::sync::Arc::new(PostgresWrapUpRepository::new(pool.clone())) as _,
wrapup_stats: std::sync::Arc::new(PostgresWrapUpStatsQuery::new(pool.clone())) as _,
insights: std::sync::Arc::new(PostgresInsightsQuery::new(pool.clone())) as _,
goal_command: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
goal_query: std::sync::Arc::new(goals::PostgresGoalRepository::new(pool.clone())) as _,
user_settings: std::sync::Arc::clone(&user_settings_repo) as _,

6
crates/adapters/postgres/src/movie_dedup.rs
View File

@@ -37,7 +37,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
.await
.map_err(adapter_common::map_sqlx_error)?;
// 1. Upsert canonical movie record
sqlx::query(
"INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
VALUES ($1, $2, $3, $4, $5, $6)
@@ -48,7 +47,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
.bind(&new).bind(&ext_id).bind(&title).bind(year).bind(&director).bind(&poster)
.execute(&mut *tx).await.map_err(adapter_common::map_sqlx_error)?;
// 2. Re-point simple FK tables
let reviews = sqlx::query("UPDATE reviews SET movie_id = $1 WHERE movie_id = $2")
.bind(&new)
.bind(&old)
@@ -74,7 +72,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
.map_err(adapter_common::map_sqlx_error)?
.rows_affected();
// 3. Re-point movie_profiles (PK — move only if canonical has none)
let profiles = sqlx::query("UPDATE movie_profiles SET movie_id = $1 WHERE movie_id = $2")
.bind(&new)
.bind(&old)
@@ -83,8 +80,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
.map_err(adapter_common::map_sqlx_error)?
.rows_affected();
// 4. Re-point enrichment tables with composite PKs (INSERT … ON CONFLICT DO NOTHING + DELETE)
// Canonical's existing rows win on conflict — old duplicates are discarded.
sqlx::query(
"INSERT INTO movie_genres (movie_id, tmdb_id, name)
SELECT $1, tmdb_id, name FROM movie_genres WHERE movie_id = $2
@@ -139,7 +134,6 @@ impl MovieDeduplicator for PostgresMovieDeduplicator {
.await
.map_err(adapter_common::map_sqlx_error)?;
// 5. Delete the now-empty old movie record (remaining cascades are safe: all FKs cleared above)
sqlx::query("DELETE FROM movies WHERE id = $1")
.bind(&old)
.execute(&mut *tx)

3
crates/adapters/postgres/src/wrapup.rs
View File

@@ -414,7 +414,8 @@ struct CastEntry {
profile_path: Option<String>,
}
async fn fetch_genres_pg(
/// Shared by WrapUp and Insights: movie-id → genre names.
pub(crate) async fn fetch_genres_pg(
pool: &PgPool,
movie_ids: &[String],
) -> Result<HashMap<String, Vec<String>>, DomainError> {

8
crates/adapters/sqlite-social/src/lib.rs
View File

@@ -11,14 +11,6 @@ pub use remote_goals::SqliteRemoteGoalRepository;
use sqlx::SqlitePool;
/// SQLite-backed implementations of the *domain* social ports.
///
/// Deliberately separate from `sqlite-federation`: this crate knows nothing
/// about ActivityPub, which is what allows a build with the `federation`
/// feature off to exclude the federation stack entirely. See ADR-0009.
///
/// Shares the `ap_followers` / `ap_following` tables with
/// `sqlite-federation`; neither crate owns migrations.
pub struct SqliteSocialRepository {
pub(crate) pool: SqlitePool,
pub(crate) instance: domain::value_objects::InstanceIdentity,

80
crates/adapters/sqlite/src/insights.rs Normal file
View File

@@ -0,0 +1,80 @@
use std::collections::HashMap;
use async_trait::async_trait;
use domain::{
errors::DomainError, models::insights::WatchRow, ports::InsightsQuery, value_objects::UserId,
};
use sqlx::{Row, SqlitePool};
use adapter_common::{parse_datetime, parse_uuid};
use crate::wrapup::fetch_genres_sqlite;
/// SQLite read side of Insights (ADR-0010). Deliberately thin: one query for
/// review basics, one for genres — every threshold and comparison lives in
/// the pure analyzer, not here.
pub struct SqliteInsightsQuery {
pool: SqlitePool,
}
impl SqliteInsightsQuery {
pub fn new(pool: SqlitePool) -> Self {
Self { pool }
}
}
#[async_trait]
impl InsightsQuery for SqliteInsightsQuery {
async fn get_watch_rows(&self, user_id: &UserId) -> Result<Vec<WatchRow>, DomainError> {
let rows = sqlx::query(
"SELECT movie_id, rating, watched_at, watch_medium \
FROM reviews \
WHERE user_id = ? \
ORDER BY watched_at ASC",
)
.bind(user_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(adapter_common::map_sqlx_error)?;
let mut movie_ids: Vec<String> = Vec::new();
let mut seen = std::collections::HashSet::new();
for row in &rows {
let mid: String = row
.try_get("movie_id")
.map_err(adapter_common::map_sqlx_error)?;
if seen.insert(mid.clone()) {
movie_ids.push(mid);
}
}
let genres_map: HashMap<String, Vec<String>> =
fetch_genres_sqlite(&self.pool, &movie_ids).await?;
let mut result = Vec::with_capacity(rows.len());
for row in &rows {
let movie_id: String = row
.try_get("movie_id")
.map_err(adapter_common::map_sqlx_error)?;
let rating: i64 = row
.try_get("rating")
.map_err(adapter_common::map_sqlx_error)?;
let watched_at: String = row
.try_get("watched_at")
.map_err(adapter_common::map_sqlx_error)?;
let watch_medium: Option<String> = row
.try_get("watch_medium")
.map_err(adapter_common::map_sqlx_error)?;
result.push(WatchRow {
movie_id: parse_uuid(&movie_id)?,
rating: rating as u8,
watched_at: parse_datetime(&watched_at)?,
watch_medium,
genres: genres_map.get(&movie_id).cloned().unwrap_or_default(),
});
}
Ok(result)
}
}

4
crates/adapters/sqlite/src/lib.rs
View File

@@ -5,6 +5,7 @@ mod goals;
mod image_ref;
mod import_profile;
mod import_session;
mod insights;
mod migrations;
mod models;
mod movie;
@@ -25,6 +26,7 @@ pub use diary::SqliteDiaryRepository;
pub use image_ref::{SqliteImageRefAdapter, create_image_ref};
pub use import_profile::SqliteImportProfileRepository;
pub use import_session::SqliteImportSessionRepository;
pub use insights::SqliteInsightsQuery;
pub use movie::SqliteMovieRepository;
pub use movie_dedup::SqliteMovieDeduplicator;
pub use persons::{SqlitePersonAdapter, create_person_adapter};
@@ -64,6 +66,7 @@ pub struct SqliteWireOutput {
pub ap_content: std::sync::Arc<dyn domain::ports::LocalApContentQuery>,
pub wrapup_repo: std::sync::Arc<dyn domain::ports::WrapUpRepository>,
pub wrapup_stats: std::sync::Arc<dyn domain::ports::WrapUpStatsQuery>,
pub insights: std::sync::Arc<dyn domain::ports::InsightsQuery>,
pub goal_command: std::sync::Arc<dyn domain::ports::GoalCommand>,
pub goal_query: std::sync::Arc<dyn domain::ports::GoalQuery>,
pub user_settings: std::sync::Arc<dyn domain::ports::UserSettingsRepository>,
@@ -114,6 +117,7 @@ pub async fn wire(database_url: &str) -> anyhow::Result<SqliteWireOutput> {
ap_content: std::sync::Arc::new(SqliteApContentQuery::new(pool.clone())) as _,
wrapup_repo: std::sync::Arc::new(SqliteWrapUpRepository::new(pool.clone())) as _,
wrapup_stats: std::sync::Arc::new(SqliteWrapUpStatsQuery::new(pool.clone())) as _,
insights: std::sync::Arc::new(SqliteInsightsQuery::new(pool.clone())) as _,
goal_command: std::sync::Arc::new(goals::SqliteGoalRepository::new(pool.clone())) as _,
goal_query: std::sync::Arc::new(goals::SqliteGoalRepository::new(pool.clone())) as _,
user_settings: std::sync::Arc::clone(&user_settings_repo) as _,

2
crates/adapters/sqlite/src/models.rs
View File

@@ -99,7 +99,6 @@ impl ReviewRow {
}
}
// Used by query_diary JOIN — r.id aliased to review_id to avoid ambiguity with m.id
#[derive(sqlx::FromRow)]
pub(crate) struct DiaryRow {
pub id: String,
@@ -173,7 +172,6 @@ impl MovieStatsRow {
}
}
// Like DiaryRow but includes user_email from JOIN with users table
#[derive(sqlx::FromRow)]
pub(crate) struct FeedRow {
pub id: String,

6
crates/adapters/sqlite/src/movie_dedup.rs
View File

@@ -37,7 +37,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator {
.await
.map_err(adapter_common::map_sqlx_error)?;
// 1. Upsert canonical movie record
sqlx::query(
"INSERT INTO movies (id, external_metadata_id, title, release_year, director, poster_path)
VALUES (?, ?, ?, ?, ?, ?)
@@ -48,7 +47,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator {
.bind(&new).bind(&ext_id).bind(&title).bind(year).bind(&director).bind(&poster)
.execute(&mut *tx).await.map_err(adapter_common::map_sqlx_error)?;
// 2. Re-point simple FK tables
let reviews = sqlx::query("UPDATE reviews SET movie_id = ? WHERE movie_id = ?")
.bind(&new)
.bind(&old)
@@ -73,7 +71,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator {
.map_err(adapter_common::map_sqlx_error)?
.rows_affected();
// 3. Re-point movie_profiles (PK — move only if canonical has none)
let profiles = sqlx::query("UPDATE movie_profiles SET movie_id = ? WHERE movie_id = ?")
.bind(&new)
.bind(&old)
@@ -82,8 +79,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator {
.map_err(adapter_common::map_sqlx_error)?
.rows_affected();
// 4. Re-point enrichment tables with composite PKs (INSERT OR IGNORE + DELETE)
// Canonical's existing rows win on conflict — old duplicates are discarded.
sqlx::query(
"INSERT OR IGNORE INTO movie_genres (movie_id, tmdb_id, name)
SELECT ?, tmdb_id, name FROM movie_genres WHERE movie_id = ?",
@@ -134,7 +129,6 @@ impl MovieDeduplicator for SqliteMovieDeduplicator {
.await
.map_err(adapter_common::map_sqlx_error)?;
// 5. Delete the now-empty old movie record (remaining cascades are safe: all FKs cleared above)
sqlx::query("DELETE FROM movies WHERE id = ?")
.bind(&old)
.execute(&mut *tx)

17
crates/adapters/sqlite/src/tests/diary.rs
View File

@@ -9,8 +9,6 @@ use sqlx::SqlitePool;
async fn setup(pool: &SqlitePool) {
sqlx::migrate!("./migrations").run(pool).await.unwrap();
// carol is a remote actor; we still need a non-null user_id for the schema,
// so we create a local "ghost" user and link the remote review via remote_actor_url.
sqlx::query(
"INSERT INTO users (id, email, username, password_hash, created_at) VALUES
('11111111-1111-1111-1111-111111111111', 'alice@example.com', 'alice', 'hash', '2024-01-01 00:00:00'),
@@ -31,7 +29,6 @@ async fn setup(pool: &SqlitePool) {
.await
.unwrap();
// carol's review: local user_id=33333333, remote_actor_url set → remote review
sqlx::query(
"INSERT INTO reviews (id, movie_id, user_id, rating, watched_at, created_at, remote_actor_url) VALUES
('a1a1a1a1-a1a1-a1a1-a1a1-a1a1a1a1a1a1', 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa', '11111111-1111-1111-1111-111111111111', 5, '2024-01-01 00:00:00', '2024-01-01 00:00:00', NULL),
@@ -97,7 +94,7 @@ async fn test_following_filter() {
.await
.unwrap();
assert_eq!(result.items.len(), 2); // alice + carol, NOT bob
assert_eq!(result.items.len(), 2);
let titles: Vec<String> = result
.items
.iter()
@@ -113,7 +110,6 @@ async fn test_get_movie_stats_local() {
setup(&pool).await;
let repo = SqliteDiaryRepository::new(pool);
// Inception: 1 local review, rating=5, no federated
let movie_id = domain::value_objects::MovieId::from_uuid(
uuid::Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(),
);
@@ -122,8 +118,8 @@ async fn test_get_movie_stats_local() {
assert_eq!(stats.total_count, 1);
assert_eq!(stats.federated_count, 0);
assert!((stats.avg_rating.unwrap() - 5.0).abs() < 0.001);
assert_eq!(stats.rating_histogram[4], 1); // 5★ bucket
assert_eq!(stats.rating_histogram[0], 0); // 1★ bucket
assert_eq!(stats.rating_histogram[4], 1);
assert_eq!(stats.rating_histogram[0], 0);
}
#[tokio::test]
@@ -176,7 +172,6 @@ async fn test_get_movie_social_feed_pagination() {
let movie_id = domain::value_objects::MovieId::from_uuid(
uuid::Uuid::parse_str("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa").unwrap(),
);
// offset beyond results: total_count still correct, items empty
let page = PageParams::new(Some(10), Some(5)).unwrap();
let result = repo.get_movie_social_feed(&movie_id, &page).await.unwrap();
@@ -190,7 +185,6 @@ async fn test_get_movie_stats_federated() {
setup(&pool).await;
let repo = SqliteDiaryRepository::new(pool);
// Dune: 1 federated review, rating=4
let movie_id = domain::value_objects::MovieId::from_uuid(
uuid::Uuid::parse_str("cccccccc-cccc-cccc-cccc-cccccccccccc").unwrap(),
);
@@ -198,8 +192,8 @@ async fn test_get_movie_stats_federated() {
assert_eq!(stats.total_count, 1);
assert_eq!(stats.federated_count, 1);
assert_eq!(stats.rating_histogram[3], 1); // 4★ bucket
assert_eq!(stats.rating_histogram[4], 0); // 5★ bucket
assert_eq!(stats.rating_histogram[3], 1);
assert_eq!(stats.rating_histogram[4], 0);
}
#[tokio::test]
@@ -208,7 +202,6 @@ async fn count_local_posts_excludes_remote_reviews() {
setup(&pool).await;
let repo = SqliteDiaryRepository::new(pool);
// setup() seeds 3 reviews: 2 local (alice, bob) + 1 remote (carol)
let count = repo.count_local_posts().await.unwrap();
assert_eq!(count, 2);
}

3
crates/adapters/sqlite/src/wrapup.rs
View File

@@ -437,7 +437,8 @@ fn in_placeholders(n: usize) -> String {
s
}
async fn fetch_genres_sqlite(
/// Shared by WrapUp and Insights: movie-id → genre names.
pub(crate) async fn fetch_genres_sqlite(
pool: &SqlitePool,
movie_ids: &[String],
) -> Result<HashMap<String, Vec<String>>, DomainError> {