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

insights-correlations #22

Merged
GKaszewski merged 4 commits from insights-correlations into master 2026-10-12 18:57:10 +00:00
Conversation 0 Commits 4 Files Changed 139 +2371 -696
139 changed files with 2371 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@@ -47,6 +47,8 @@ ALLOW_REGISTRATION=true
# SECURE_COOKIES=true
# Handler-level logs come from the `presentation` crate; startup/wiring logs come
# from `server`. Include both — `server=info` alone silently drops handler logs.
# Crates not named here (adapters, etc.) still log warnings and errors unless
# you set a bare level yourself (e.g. `info,presentation=debug`).
# RUST_LOG=server=info,presentation=info,tower_http=info,worker=info
# CORS (for SPA development only)

4
CONTEXT.md
View File

@@ -36,6 +36,10 @@ _Avoid_: Challenge, resolution, target
A generated summary report of viewing activity over a date range — statistics, trends, highlights, top directors/actors/genres. Can be personal (one user) or global (all users). Generated asynchronously. Shown to users as "Year in Review."
_Avoid_: Stats page, recap, summary
**Insight**:
A statement about a user's viewing behavior derived from their Reviews and linked Movie metadata, produced by correlation analysis. Always a comparison — relative to the user's own baseline (their average rating, their genre frequencies, their typical hours) — never a plain ranking or count. "You rate thrillers higher than your average" is an Insight; "thriller is your top genre" is a statistic.
_Avoid_: Correlation (implementation), fun fact, finding
**User**:
A registered account with a username, email, and profile (display name, bio, avatar, banner). Can be Standard or Admin.
_Avoid_: Account, member, profile (as a synonym for the whole User)

70
Cargo.lock generated
View File

@@ -612,13 +612,13 @@ checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de"
[[package]]
name = "async-trait"
version = "0.1.89"
version = "0.1.92"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb"
checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
"syn 3.0.9",
]
[[package]]
@@ -1067,6 +1067,16 @@ dependencies = [
"windows-link",
]
[[package]]
name = "chrono-tz"
version = "0.10.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6139a8597ed92cf816dfb33f5dd6cf0bb93a6adc938f11039f371bc5bcd26c3"
dependencies = [
"chrono",
"phf 0.12.1",
]
[[package]]
name = "cipher"
version = "0.4.4"
@@ -1321,7 +1331,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb2a7d3066da2de787b7f032c736763eb7ae5d355f81a68bab2675a96008b0bf"
dependencies = [
"lab",
"phf",
"phf 0.11.3",
]
[[package]]
@@ -1333,7 +1343,7 @@ dependencies = [
"cssparser-macros",
"dtoa-short",
"itoa",
"phf",
"phf 0.11.3",
"smallvec",
]
@@ -3751,7 +3761,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078"
dependencies = [
"phf_macros",
"phf_shared",
"phf_shared 0.11.3",
]
[[package]]
name = "phf"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "913273894cec178f401a31ec4b656318d95473527be05c0752cc41cdc32be8b7"
dependencies = [
"phf_shared 0.12.1",
]
[[package]]
@@ -3761,7 +3780,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aef8048c789fa5e851558d709946d6d79a8ff88c0440c587967f8e94bfb1216a"
dependencies = [
"phf_generator",
"phf_shared",
"phf_shared 0.11.3",
]
[[package]]
@@ -3770,7 +3789,7 @@ version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [
"phf_shared",
"phf_shared 0.11.3",
"rand 0.8.6",
]
@@ -3781,7 +3800,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f84ac04429c13a7ff43785d75ad27569f2951ce0ffd30a3321230db2fc727216"
dependencies = [
"phf_generator",
"phf_shared",
"phf_shared 0.11.3",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -3796,6 +3815,15 @@ dependencies = [
"siphasher",
]
[[package]]
name = "phf_shared"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06005508882fb681fd97892ecff4b7fd0fee13ef1aa569f8695dae7ab9099981"
dependencies = [
"siphasher",
]
[[package]]
name = "pin-project"
version = "1.1.13"
@@ -4047,6 +4075,7 @@ dependencies = [
"axum-governor",
"bytes",
"chrono",
"chrono-tz",
"composition",
"domain",
"dotenvy",
@@ -5545,7 +5574,7 @@ checksum = "bf776ba3fa74f83bf4b63c3dcbbf82173db2632ed8452cb2d891d33f459de70f"
dependencies = [
"new_debug_unreachable",
"parking_lot",
"phf_shared",
"phf_shared 0.11.3",
"precomputed-hash",
"serde",
]
@@ -5557,7 +5586,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c711928715f1fe0fe509c53b43e993a9a557babc2d0a3567d0a3006f1ac931a0"
dependencies = [
"phf_generator",
"phf_shared",
"phf_shared 0.11.3",
"proc-macro2",
"quote",
]
@@ -5628,6 +5657,17 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "syn"
version = "3.0.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d78c8dee4c7bf0e14673097256fed6142ce9d3b85a408189d07482442145823b"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "sync_wrapper"
version = "1.0.2"
@@ -5737,7 +5777,7 @@ checksum = "d4ea810f0692f9f51b382fff5893887bb4580f5fa246fde546e0b13e7fcee662"
dependencies = [
"fnv",
"nom 7.1.3",
"phf",
"phf 0.11.3",
"phf_codegen",
]
@@ -5774,7 +5814,7 @@ dependencies = [
"ordered-float",
"pest",
"pest_derive",
"phf",
"phf 0.11.3",
"sha2",
"signal-hook",
"siphasher",
@@ -6654,7 +6694,7 @@ version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57ffde1dc01240bdf9992e3205668b235e59421fd085e8a317ed98da0178d414"
dependencies = [
"phf",
"phf 0.11.3",
"phf_codegen",
"string_cache",
"string_cache_codegen",
@@ -6801,7 +6841,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.48.0",
"windows-sys 0.61.2",
]
[[package]]

3
Cargo.toml
View File

@@ -52,9 +52,10 @@ anyhow = "1.0"
thiserror = "2.0"
tracing = "0.1"
tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }
async-trait = "0.1"
async-trait = "0.1.92"
uuid = { version = "1.23.0", features = ["v4", "v5", "serde"] }
chrono = { version = "0.4", features = ["serde"] }
chrono-tz = "0.10"
sqlx = { version = "0.8.6", features = [
"runtime-tokio-rustls",
"sqlite",

138
Makefile
View File

@@ -1,10 +1,8 @@
.DEFAULT_GOAL := check
# Run the full local check suite — same order as CI would.
check: fmt-check clippy test check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-social-crates-are-ap-free
@echo "✅ All checks passed"
# Enforce that no application use case imports AppContext (god-object guard).
check-appcontext:
@if grep -rn "AppContext" crates/application/src --include="*.rs" | grep -q .; then \
echo "❌ AppContext found in application crate:"; \
@@ -13,20 +11,6 @@ check-appcontext:
fi
@echo "✅ No AppContext in application crate"
# Enforce that presentation never hand-builds a deps struct (composition-root guard).
# Catches struct literals (`FooDeps { ... }`) and conversions (`FooDeps::from(...)`) —
# both are the composition-root pattern that belongs in crates/composition.
#
# Scope is the whole crate, not just handlers/, so that factoring a
# `fn build_login_deps(state) -> LoginDeps { .. }` helper into a sibling module
# does not slip past the guard. One exclusion:
# src/tests/ — test fixtures legitimately assemble their own state
# `src/main.rs` used to be excluded too ("the binary IS the wiring root") before
# crates/server took over the binary (ADR-0006) — presentation is lib-only now, has no
# main.rs, and the exclusion was dropped rather than left pointing at a file that no
# longer exists.
# NOTE: `\s` is a GNU grep extension, not POSIX ERE. Fine on GNU/Ubuntu runners;
# would need `[[:space:]]` if this ever runs under BusyBox/Alpine grep.
check-handler-deps:
@if grep -rnE "[A-Za-z0-9_]*Deps(\s*\{|::from\()" crates/presentation/src --include="*.rs" --exclude-dir=tests | grep -q .; then \
echo "❌ hand-built deps struct found in presentation:"; \
@@ -35,30 +19,6 @@ check-handler-deps:
fi
@echo "✅ No hand-built deps structs in presentation"
# Enforce that presentation calls use cases only, never repository methods directly
# (ADR-0006's guard; ADR-0007 records the widening below).
#
# Was: two shape-matching grep passes for `repos.<field>.<method>(` and
# `repos.<field>.clone().<method>(`, needed because `AppContext.repos` was still a
# legitimate way to reach a repository — ~40 sites cloned an Arc out of it to pass
# positionally into a use case, so the guard had to distinguish that legal shape
# from an illegal direct call. That shape-matching was demonstrably incomplete: it
# matches `repos.<field>.<method>(` textually, so it cannot see a bare reference
# pass (`&repos.diary`, no call at all at the read site) or a `let`-bound access
# (`if let Some(x) = repos.federated_profile` then `x.method()` on the *binding*,
# not on `repos` itself). Both shapes existed in production code — Plan C2's Task 6
# review caught it, Task 7 fixed both — while the two-pass guard printed a clean
# ✅ the whole time. ADR-0006's "handlers call use cases only" was false at that
# line for as long as the hole existed.
#
# Now: Plan C2 deleted `AppContext.repos` entirely (ADR-0007), so there is no
# longer any legitimate reason for the substring `app_ctx.repos` to appear in
# production code — not a call, not a clone, not a reference, not a let-binding.
# That makes the rule trivial and unevadeable: any mention at all is an offender.
# No shape to match means no shape to miss.
#
# One exclusion: src/tests/ — test helpers still legitimately build a
# `Repositories` and call `composition::build_deps` to produce test state.
check-handler-repos:
@if grep -rn "app_ctx\.repos" crates/presentation/src --include="*.rs" --exclude-dir=tests | grep -q .; then \
echo "❌ app_ctx.repos referenced in presentation:"; \
@@ -67,18 +27,7 @@ check-handler-repos:
fi
@echo "✅ No app_ctx.repos references in presentation"
# Enforce that presentation depends on at most one adapter crate.
#
# Presentation may name things that RENDER OUTPUT; it may not name things that
# REACH external systems or storage. `template-askama` is the HTML template
# engine, so it stays. See ADR-0008.
#
# This checks Cargo.toml rather than grepping src/, deliberately: `<crate>::`
# greps collide with same-named modules inside `application` and `presentation`
# (`auth::logout` is `application::auth`; `rss::get_user_feed` is
# `handlers::rss`), so the source-level grep cannot decide this. Every directory
# under crates/adapters/ is named exactly like its package, so directory names
# are a safe source of truth.
PRESENTATION_ALLOWED_ADAPTERS := template-askama
check-presentation-adapters:
@@ -96,41 +45,7 @@ check-presentation-adapters:
fi; \
echo "✅ presentation depends on no adapter crate but $(PRESENTATION_ALLOWED_ADAPTERS)"
# Enforce that the federation-off build wires real social behavior, not silent noops.
#
# With federation off, `NoopSocialCommand::follow` returns Ok(()) and writes nothing,
# and `NoopSocialQuery` always answers zero/empty. The write side is currently
# unreachable over HTTP (the social routes are federation-gated in presentation), but
# the query side is not: `get_local_profile`, `get_page_viewer` and `get_activity_feed`
# read it through fully ungated routes, so a noop there means a federation-off
# instance lies about its own follow graph on every profile view and activity feed
# request — contradicting ADR-0003, which established that local follows bypass
# ActivityPub entirely. `application::social::LocalSocialService` is what must be
# wired instead. See ADR-0009.
#
# Two checks, not one: absence of the noops is necessary but not sufficient — deleting
# the whole `#[cfg(not(feature = "federation"))]` wiring block would also make the
# first check pass, leaving federation-off with nothing wired at all, which is worse
# than the noops. So this also asserts `LocalSocialService` is present — and it must
# be present in the federation-OFF region specifically, not merely anywhere in the
# file: `LocalSocialService` is also constructed inside the federation-ON branch
# (feeding `CompositeSocialAdapter`), so a bare file-wide grep is satisfied by that
# occurrence alone and would not catch the federation-off block being deleted. The
# awk carves out the region from the first `#[cfg(not(feature = "federation"))]` to
# the next `#[cfg(feature = "federation")]` — the same kind of range
# `check-presentation-adapters` uses for `[dependencies]` — and greps only inside it.
# Not an occurrence count: that would work today but break the moment the two
# `LocalSocialService` constructions in `main.rs` are deduplicated (a recorded
# follow-up), which could legitimately drop the total to one.
#
# A grep rather than a test because the decision lives in `wire_dependencies()`, a
# private fn in a binary crate, unreachable from integration tests.
#
# Scope is the single file `crates/server/src/main.rs`, not the whole crate: it is the
# sole wiring site for these ports. `worker` and `tui` never name SocialCommand /
# FollowGraphQuery / BlockQuery at all, so there is nowhere else for the noops to
# reappear. Relocating the wiring out of main.rs would move it out of the guard's
# scope — widen this to `crates/server/src` if that ever happens.
check-federation-off-social:
@if grep -n "NoopSocialCommand\|NoopSocialQuery" crates/server/src/main.rs | grep -q .; then \
echo "❌ server wires social noops — federation-off would report an empty follow graph:"; \
@@ -143,24 +58,7 @@ check-federation-off-social:
fi
@echo "✅ server wires real social behavior in every configuration"
# Enforce that `<backend>-social` crates never depend on ActivityPub.
#
# ADR-0009's whole federation-optional property rests on one invariant: a
# `<backend>-social` crate speaks only `domain::ports`; a `<backend>-federation` crate
# speaks ActivityPub; the former must never depend on the latter. It holds today, and
# nothing but this guard enforces it. One manifest line —
# `k-ap = { version = "0.5.0", registry = "gitea" }` in a `-social` crate's Cargo.toml,
# the obvious move when someone wants to reuse a helper like `status_to_str` from
# `sqlite-federation/src/lib.rs` — would silently restore all 76 crates and 19MB that
# federation-off exists to shed, while `make check`, `cargo test` and CI all stay green
# without this guard to catch it. Matches `activitypub` with any suffix, so
# `activitypub_federation = "0.6"` (the upstream crate carrying all 76 transitives) is
# caught too, not just the local `activitypub` crate.
#
# Known gap: a dependency renamed via a `package = "k-ap"` key (`foo = { package =
# "k-ap", ... }`), or declared with a `[dependencies.k-ap]` table header instead of the
# inline form, would not match this line-level regex. Not attempted — nothing in the
# workspace does either today.
check-social-crates-are-ap-free:
@if grep -nE '^[[:space:]]*(k-ap|activitypub[A-Za-z0-9_-]*|[A-Za-z0-9_-]+-federation)[[:space:]]*=' crates/adapters/*-social/Cargo.toml | grep -q .; then \
echo "❌ social crate depends on ActivityPub:"; \
@@ -169,25 +67,43 @@ check-social-crates-are-ap-free:
fi
@echo "✅ social crates depend on no k-ap, activitypub, or *-federation crate"
# Apply rustfmt to all files.
fmt:
cargo fmt
# Check formatting without modifying files (CI-safe).
fmt-check:
cargo fmt --check
# Run Clippy and treat warnings as errors.
clippy:
cargo clippy -- -D warnings
# Run the test suite.
test:
cargo test
# Apply fmt + clippy auto-fixes in one shot.
fix:
cargo fmt
cargo clippy --fix --allow-dirty --allow-staged
.PHONY: check fmt fmt-check clippy test fix check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-social-crates-are-ap-free
# Run the dev stack:
# server (:3000, in the foreground, Ctrl-C stops it)
# worker (background, log at /tmp/movies-diary-worker.log)
# spa (:5173, background, log at /tmp/movies-diary-spa.log)
# Tear down the background processes with `make dev-stop`.
# .env enables EVENT_BUS_BACKEND=nats, so both binaries are built with the nats feature.
dev: dev-worker dev-spa
cargo run -p server --features nats
dev-stop:
-pkill -f '[t]arget.debug.worker' || true
-pkill -f '[s]pa/node_modules/[.]bin/vite' || true
@echo "dev stack stopped"
dev-worker:
@bash -c 'nohup cargo run -p worker --features nats >> /tmp/movies-diary-worker.log 2>&1 & echo "worker → pid $$! (log: /tmp/movies-diary-worker.log)"'
dev-spa:
@bash -c 'nohup npm --prefix spa run dev >> /tmp/movies-diary-spa.log 2>&1 & echo "spa → pid $$! (log: /tmp/movies-diary-spa.log, http://localhost:5173)"'
spa-dev:
@npm --prefix spa run dev
.PHONY: check fmt fmt-check clippy test fix dev dev-stop dev-worker dev-spa spa-dev check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-social-crates-are-ap-free

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 AND remote_actor_url IS NULL \
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,

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

@@ -0,0 +1,90 @@
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};
/// 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 = ? AND remote_actor_url IS NULL \
ORDER BY watched_at ASC",
)
.bind(user_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(adapter_common::map_sqlx_error)?;
// Subquery rather than an `IN (?, …)` bind list: a whole diary can
// exceed SQLite's bound-parameter limit.
let genre_rows = sqlx::query(
"SELECT movie_id, name FROM movie_genres \
WHERE movie_id IN ( \
SELECT movie_id FROM reviews \
WHERE user_id = ? AND remote_actor_url IS NULL \
) \
ORDER BY name",
)
.bind(user_id.value().to_string())
.fetch_all(&self.pool)
.await
.map_err(adapter_common::map_sqlx_error)?;
let mut genres_map: HashMap<String, Vec<String>> = HashMap::new();
for row in genre_rows {
let mid: String = row
.try_get("movie_id")
.map_err(adapter_common::map_sqlx_error)?;
let name: String = row
.try_get("name")
.map_err(adapter_common::map_sqlx_error)?;
genres_map.entry(mid).or_default().push(name);
}
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> {

1
crates/api-types/src/diary.rs
View File

@@ -60,7 +60,6 @@ pub struct FeedEntryDto {
#[derive(Debug, Clone, Deserialize, utoipa::IntoParams)]
#[into_params(parameter_in = Query)]
pub struct ExportQueryParams {
/// Output format: `csv` (default) or `json`
#[serde(default = "default_export_format")]
pub format: String,
}

7
crates/api-types/src/import.rs
View File

@@ -17,13 +17,9 @@ pub struct SessionStateResponse {
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct ApiFieldMapping {
/// Column name in the source file
pub source_column: String,
/// Domain field: title | release_year | director | rating | watched_at | comment | external_metadata_id
pub domain_field: String,
/// For rating fields: multiply raw value by this factor (e.g. 0.5 for 10-point → 5-point scale)
pub rating_scale: Option<f64>,
/// For watched_at fields: strftime format hint (e.g. "%d/%m/%Y")
pub date_format: Option<String>,
}
@@ -34,15 +30,12 @@ pub struct ApplyMappingRequest {
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct ConfirmRequest {
/// Indices (0-based) of rows from the mapping preview to import
pub confirmed_indices: Vec<usize>,
}
#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct SaveProfileRequest {
/// Session UUID whose current field_mappings to save
pub session_id: String,
/// Human-readable profile name (e.g. "Letterboxd")
pub name: String,
}

57
crates/api-types/src/insights.rs Normal file
View File

@@ -0,0 +1,57 @@
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;
/// One conservative, baseline-relative statement about the user's viewing
/// behavior. Tagged by `kind` so the SPA can switch on it; every variant
/// carries `sample_count` so the UI can explain a thin sample.
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum InsightDto {
/// `genre`'s average rating sits `delta` stars from the user's own
/// average (first-time watches only).
RatingLift {
genre: String,
genre_avg: f64,
user_avg: f64,
delta: f64,
sample_count: u32,
},
/// `genre`'s share of watches in `month` (1–12) is `multiplier` times the
/// user's own share of watches in that month.
Seasonality {
genre: String,
month: u32,
multiplier: f64,
sample_count: u32,
},
/// `medium`'s weekend share of watches is `multiplier` times the user's
/// overall weekend share.
MediumTiming {
medium: String,
multiplier: f64,
weekend_share: f64,
baseline_share: f64,
sample_count: u32,
},
/// Watching concentrates in a 3-hour window starting `peak_hour_start`
/// (0–23) on `top_days` (Mon=1 … Sun=7).
Rhythm {
peak_hour_start: u32,
peak_hour_end: u32,
top_days: Vec<u32>,
sample_count: u32,
},
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct InsightsResponse {
pub insights: Vec<InsightDto>,
}
#[derive(Debug, Clone, Deserialize, utoipa::IntoParams)]
#[into_params(parameter_in = Query)]
pub struct InsightsQueryParams {
/// IANA time zone (e.g. `Europe/Warsaw`) months, weekdays and hours are
/// bucketed in. Defaults to UTC.
pub tz: Option<String>,
}

2
crates/api-types/src/lib.rs
View File

@@ -3,6 +3,7 @@ pub mod common;
pub mod diary;
pub mod goals;
pub mod import;
pub mod insights;
pub mod movies;
pub mod rendering;
pub mod search;
@@ -17,6 +18,7 @@ pub use common::*;
pub use diary::*;
pub use goals::*;
pub use import::*;
pub use insights::*;
pub use movies::*;
pub use rendering::*;
pub use social::*;

3
crates/api-types/src/movies.rs
View File

@@ -10,11 +10,8 @@ use crate::common::SocialFeedResponse;
pub struct MoviesQueryParams {
pub limit: Option<u32>,
pub offset: Option<u32>,
/// Optional title filter (case-insensitive substring match)
pub search: Option<String>,
/// Filter by genre name (case-insensitive)
pub genre: Option<String>,
/// Filter by ISO language code (e.g. "en")
pub language: Option<String>,
}

8
crates/api-types/src/search.rs
View File

@@ -6,21 +6,13 @@ pub use crate::common::{PaginatedMovieHits, PaginatedPersonHits};
#[derive(Debug, Deserialize, IntoParams)]
pub struct SearchQueryParams {
/// Free-text query matched across title, cast, crew, genres and keywords.
pub q: Option<String>,
/// Filter by genre name (exact match, case-sensitive).
pub genre: Option<String>,
/// Filter by release year.
pub year: Option<u16>,
/// Filter by person ID (UUID).
pub person_id: Option<Uuid>,
/// Filter crew results by department (e.g. "Directing", "Writing").
pub department: Option<String>,
/// Filter by original language code (e.g. "en", "fr").
pub language: Option<String>,
/// Max results to return (default 20).
pub limit: Option<u32>,
/// Offset for pagination (default 0).
pub offset: Option<u32>,
}

4
crates/api-types/src/social.rs
View File

@@ -16,7 +16,6 @@ pub struct RemoteActorDto {
pub handle: String,
pub display_name: Option<String>,
pub url: String,
/// `Some` for local actors, so the SPA can link internally to `/users/{id}`.
pub user_id: Option<Uuid>,
pub avatar_url: Option<String>,
}
@@ -79,12 +78,11 @@ pub struct PendingCountResponse {
pub count: usize,
}
//todo: move out to separate file
#[cfg(test)]
mod tests {
use super::*;
/// Task 7's SPA zod schema parses these exact lowercase literals —
/// a casing or naming drift here breaks the SPA at runtime.
#[test]
fn follow_state_dto_serializes_to_lowercase_strings() {
assert_eq!(

1
crates/api-types/src/users.rs
View File

@@ -23,7 +23,6 @@ pub struct UsersResponse {
#[derive(Debug, Clone, Deserialize, utoipa::IntoParams)]
#[into_params(parameter_in = Query)]
pub struct UserProfileQueryParams {
/// One of: `recent` (default), `ratings`, `history`, `trends`
pub view: Option<String>,
pub limit: Option<u32>,
pub offset: Option<u32>,

6
crates/application/src/deps.rs
View File

@@ -13,6 +13,7 @@ use crate::import::deps::{
DeleteImportProfileDeps, ExecuteImportDeps, GetMappingStageDeps, GetPreviewStageDeps,
GetSessionStateDeps, ListImportProfilesDeps, SaveProfileDeps,
};
use crate::insights::deps::GetInsightsDeps;
use crate::integrations::deps::{
ConfirmWatchEventsDeps, DismissWatchEventsDeps, GenerateWebhookTokenDeps, GetWatchQueueDeps,
GetWebhookTokensDeps, IngestWatchEventDeps, RevokeWebhookTokenDeps,
@@ -92,6 +93,10 @@ pub struct IntegrationsGroup {
pub plex_parser: Arc<dyn MediaServerParser>,
}
pub struct InsightsGroup {
pub get: GetInsightsDeps,
}
pub struct MoviesGroup {
pub sync_poster: SyncPosterDeps,
pub get_movie_profile: GetMovieProfileDeps,
@@ -159,6 +164,7 @@ pub struct Deps {
pub diary: DiaryGroup,
pub goals: GoalsGroup,
pub import: ImportGroup,
pub insights: InsightsGroup,
pub integrations: IntegrationsGroup,
pub movies: MoviesGroup,
pub person: PersonGroup,

3
crates/application/src/diary/get_user_feed.rs
View File

@@ -7,9 +7,6 @@ use crate::diary::deps::{GetDiaryDeps, GetUserFeedDeps};
use crate::diary::get_diary;
use crate::diary::queries::GetDiaryQuery;
/// The RSS feed's author line — derived the same way the deleted handler code
/// derived its page title: from the local part of the user's email, not their
/// username.
pub struct FeedAuthor {
pub display_name: String,
}

2
crates/application/src/diary/tests/review_logger.rs
View File

@@ -242,8 +242,6 @@ async fn watchlist_not_present_does_not_publish_removed() {
);
}
/// A metadata client that returns a movie with an external_metadata_id,
/// triggering the MovieDiscovered event path.
struct MetadataClientWithExternalId;
#[async_trait]

2
crates/application/src/import/apply_profile.rs
View File

@@ -5,8 +5,6 @@ use domain::{
use super::{commands::ApplyImportProfileCommand, deps::ApplyProfileDeps};
/// Copies the profile's field_mappings onto the session. Caller must then invoke
/// apply_import_mapping to regenerate row_results with the new mappings.
pub async fn execute(
deps: &ApplyProfileDeps,
cmd: ApplyImportProfileCommand,

5
crates/application/src/import/deps.rs
View File

@@ -50,11 +50,6 @@ pub struct ListImportProfilesDeps {
pub import_profile: Arc<dyn ImportProfileRepository>,
}
/// Backs `apply_profile_and_map`, which internally drives `apply_profile::execute`
/// then `apply_mapping::execute` — these fields are exactly the union of
/// `ApplyProfileDeps` and `ApplyMappingDeps`'s fields, cloned once here and used to
/// build each nested deps struct inline at the call site (see that file's doc
/// comment for why: no use-case signature changes, per this task's constraints).
pub struct ApplyProfileAndMapDeps {
pub import_profile: Arc<dyn ImportProfileRepository>,
pub import_session: Arc<dyn ImportSessionRepository>,

8
crates/application/src/import/get_mapping_stage.rs
View File

@@ -1,16 +1,8 @@
//! The mapping-page stage gate: a session must exist and have a `parsed_file`
//! before its columns/sample rows can be shown for field mapping. Absorbs
//! `handlers/import.rs::get_mapping_page`'s two early-return checks (session
//! missing, `parsed_file` absent) — both collapse to `NotFound` here since the
//! handler redirected to the same place (`/import`) for either.
use domain::{errors::DomainError, value_objects::ImportSessionId};
use uuid::Uuid;
use crate::import::deps::GetMappingStageDeps;
/// Cap on sample rows shown on the mapping page — was a bare `.take(5)` in the
/// handler.
pub const SAMPLE_ROW_LIMIT: usize = 5;
pub struct MappingStage {

16
crates/application/src/import/get_preview_stage.rs
View File

@@ -1,10 +1,3 @@
//! The preview-page stage gate: a session must have `row_results` (i.e. a
//! mapping has already been applied) before its rows can be previewed. Serves
//! both the HTML preview handler and the API preview handler —
//! `handlers/import.rs::get_preview_page` and `::api_get_preview` — which
//! render/respond to `NotYetMapped` differently (redirect vs. status code); that
//! decision stays in the handlers, not here.
use domain::{
errors::DomainError,
models::AnnotatedRow,
@@ -14,15 +7,6 @@ use uuid::Uuid;
use crate::import::deps::GetPreviewStageDeps;
/// The columns and mapped/annotated rows for a session whose mapping has
/// already been applied. `columns` comes from the session's `parsed_file` —
/// the HTML preview template renders it as the table header — while `rows`
/// comes from `row_results`. Not in the brief's `PreviewStage::Ready(Vec<AnnotatedRow>)`
/// sketch: the deleted `get_preview_page` handler code read both
/// `session.parsed_file.columns` and `session.row_results` to render the page,
/// so dropping `columns` here would either blank the preview table's header or
/// force the handler to re-fetch the session itself (forbidden — that's the
/// exact repo call this task removes). See task-2 report for detail.
pub struct PreviewRows {
pub columns: Vec<String>,
pub rows: Vec<AnnotatedRow>,

2
crates/application/src/import/tests/apply_mapping.rs
View File

@@ -86,8 +86,6 @@ async fn fails_when_session_not_found() {
assert!(result.is_err());
}
/// A document parser that returns rows with specific field values for testing
/// the mark_duplicates logic.
struct DuplicateTestParser {
rows: Vec<ImportRow>,
}

7
crates/application/src/insights/deps.rs Normal file
View File

@@ -0,0 +1,7 @@
use std::sync::Arc;
use domain::ports::InsightsQuery;
pub struct GetInsightsDeps {
pub insights: Arc<dyn InsightsQuery>,
}

24
crates/application/src/insights/get.rs Normal file
View File

@@ -0,0 +1,24 @@
//! Live Insights for the requesting user (ADR-0010): fetch the whole diary as
//! `WatchRow`s, run the pure analyzer, hand back whatever clears the honesty
//! thresholds. No caching, no snapshots — the same on-the-fly shape as the
//! existing live stats.
use chrono::TimeZone;
use domain::{errors::DomainError, models::Insight, value_objects::UserId};
use crate::insights::deps::GetInsightsDeps;
pub async fn execute<Tz: TimeZone + Sync>(
deps: &GetInsightsDeps,
user_id: UserId,
tz: &Tz,
) -> Result<Vec<Insight>, DomainError> {
let rows = deps.insights.get_watch_rows(&user_id).await?;
Ok(domain::services::insights_analyzer::build_insights(
&rows, tz,
))
}
#[cfg(test)]
#[path = "tests/get.rs"]
mod tests;

2
crates/application/src/insights/mod.rs Normal file
View File

@@ -0,0 +1,2 @@
pub mod deps;
pub mod get;

99
crates/application/src/insights/tests/get.rs Normal file
View File

@@ -0,0 +1,99 @@
use std::sync::Arc;
use chrono::NaiveDate;
use uuid::Uuid;
use domain::{models::insights::WatchRow, testing::InMemoryInsightsQuery, value_objects::UserId};
use super::*;
fn watch(genre: &str, rating: u8, day: u32, hour: u32) -> WatchRow {
WatchRow {
movie_id: Uuid::new_v4(),
rating,
watched_at: NaiveDate::from_ymd_opt(2024, 1, day)
.unwrap()
.and_hms_opt(hour, 0, 0)
.unwrap(),
watch_medium: Some("streaming".to_string()),
genres: vec![genre.to_string()],
}
}
#[tokio::test]
async fn returns_analyzer_output_for_the_user() {
let user = Uuid::new_v4();
let rows: Vec<WatchRow> = (0..12)
.map(|i| watch("Thriller", 5, 1 + i, 22))
.chain((0..12).map(|i| watch("Comedy", 2, 13 + i, 22)))
.collect();
let deps = GetInsightsDeps {
insights: InMemoryInsightsQuery::with_rows(user, rows),
};
let insights = execute(&deps, UserId::from_uuid(user), &chrono::Utc)
.await
.unwrap();
assert!(insights.iter().any(
|i| matches!(i, domain::models::Insight::RatingLift { genre, .. } if genre == "Thriller")
));
}
#[tokio::test]
async fn other_users_rows_are_not_seen() {
let user = Uuid::new_v4();
let other = Uuid::new_v4();
let rows: Vec<WatchRow> = (0..12)
.map(|i| watch("Thriller", 5, 1 + i, 22))
.chain((0..12).map(|i| watch("Comedy", 2, 13 + i, 22)))
.collect();
let deps = GetInsightsDeps {
insights: InMemoryInsightsQuery::with_rows(other, rows),
};
let insights = execute(&deps, UserId::from_uuid(user), &chrono::Utc)
.await
.unwrap();
assert!(insights.is_empty());
}
#[tokio::test]
async fn empty_diary_returns_empty_list() {
let user = Uuid::new_v4();
let deps = GetInsightsDeps {
insights: InMemoryInsightsQuery::new(),
};
let insights = execute(&deps, UserId::from_uuid(user), &chrono::Utc)
.await
.unwrap();
assert!(insights.is_empty());
}
#[tokio::test]
async fn port_errors_propagate() {
struct Failing;
#[async_trait::async_trait]
impl domain::ports::InsightsQuery for Failing {
async fn get_watch_rows(
&self,
_: &UserId,
) -> Result<Vec<WatchRow>, domain::errors::DomainError> {
Err(domain::errors::DomainError::InfrastructureError(
"db down".into(),
))
}
}
let deps = GetInsightsDeps {
insights: Arc::new(Failing),
};
let err = execute(&deps, UserId::from_uuid(Uuid::new_v4()), &chrono::Utc)
.await
.unwrap_err();
assert!(matches!(
err,
domain::errors::DomainError::InfrastructureError(_)
));
}

1
crates/application/src/lib.rs
View File

@@ -9,6 +9,7 @@ pub mod auth;
pub mod diary;
pub mod goals;
pub mod import;
pub mod insights;
pub mod integrations;
pub mod movies;
pub mod person;

3
crates/application/src/movies/discovery_indexer.rs
View File

@@ -8,9 +8,6 @@ use domain::{
ports::{EventHandler, MovieQuery, SearchCommand},
};
/// Reacts to `MovieDiscovered` and inserts a bare search index entry immediately,
/// so movies are findable before TMDb enrichment runs.
/// Enrichment will later overwrite this with the full document (cast, genres, etc.).
pub struct MovieDiscoveryIndexer {
movie_repository: Arc<dyn MovieQuery>,
search_command: Arc<dyn SearchCommand>,

8
crates/application/src/movies/enrich_movie.rs
View File

@@ -8,22 +8,18 @@ use domain::{
use crate::movies::{commands::EnrichMovieCommand, deps::EnrichMovieDeps};
pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<(), DomainError> {
// 1. Persist the enriched profile (also handles movie_cast, movie_crew, genres, keywords)
deps.movie_profile.upsert(&cmd.profile).await?;
// 2. Upsert persons extracted from cast + crew (no reads — only upsert)
let persons = extract_persons(&cmd.profile.cast, &cmd.profile.crew);
if !persons.is_empty() {
deps.person_command.upsert_batch(&persons).await?;
}
// 3. Fetch the movie for the search index document
let Some(movie) = deps.movie_query.get_movie_by_id(&cmd.movie_id).await? else {
tracing::warn!(movie_id = %cmd.movie_id.value(), "enrich_movie: movie not found after profile upsert");
return Ok(());
};
// 4. Index the movie in search
deps.search_command
.index(IndexableDocument::Movie {
id: cmd.movie_id.clone(),
@@ -32,7 +28,6 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<
})
.await?;
// 5. Index each unique person in search (no reads — persons built from in-memory data)
for person in &persons {
deps.search_command
.index(IndexableDocument::Person {
@@ -50,9 +45,6 @@ pub async fn execute(deps: &EnrichMovieDeps, cmd: EnrichMovieCommand) -> Result<
Ok(())
}
/// Build unique Person values from cast and crew.
/// Uses deterministic UUIDv5 so the same tmdb_person_id always maps to the same PersonId.
/// No DB reads — persons are built entirely from in-memory TMDb data.
fn extract_persons(cast: &[CastMember], crew: &[CrewMember]) -> Vec<Person> {
let mut seen: HashMap<u64, Person> = HashMap::new();

3
crates/application/src/movies/resolve.rs
View File

@@ -9,9 +9,6 @@ use domain::{
use crate::diary::commands::MovieInput;
use crate::diary::movie_resolver::{MovieResolver, MovieResolverDeps};
/// Resolves a movie from input, persists it, and publishes `MovieDiscovered` if new.
///
/// Returns `(movie, is_new_movie)`.
pub async fn resolve_and_persist_movie(
input: &MovieInput,
movie_command: &dyn MovieCommand,

3
crates/application/src/services.rs
View File

@@ -7,9 +7,6 @@ use domain::ports::{
use crate::ports::ReviewLogger;
/// Services the application layer needs, assembled by the composition root.
/// Adapter-typed ports do not belong here — the AP port inversion removed the
/// last of them; see ADR-0008.
#[derive(Clone)]
pub struct Services {
pub auth: Arc<dyn AuthService>,

7
crates/application/src/social/local_service.rs
View File

@@ -13,13 +13,6 @@ use domain::{
},
};
/// The subset of `SocialCommand`/`FollowGraphQuery`/`BlockQuery` that a single
/// instance can serve with only its own database — no ActivityPub involved.
///
/// This is the local half of `activitypub::CompositeSocialAdapter`, extracted
/// here so `application` needs no dependency on the `activitypub` crate to
/// offer it. It implements `domain::ports::LocalSocial` via that trait's
/// blanket impl over the five ports below.
pub struct LocalSocialService {
user_repo: Arc<dyn UserRepository>,
follow_command: Arc<dyn FollowCommand>,

4
crates/application/src/tests/services.rs
View File

@@ -6,10 +6,6 @@ use domain::testing::{
};
use std::sync::Arc;
/// Services must be constructible from domain ports alone — no adapter types.
/// If this file ever needs an adapter crate import, the boundary has regressed.
/// Asserts on Clone specifically: AppState is Clone, so a Services that clones
/// its Arcs by value instead of sharing them would silently duplicate state.
#[test]
fn services_clone_shares_the_same_port_instances() {
let logger: Arc<dyn crate::ports::ReviewLogger> = Arc::new(NoopReviewLogger);

9
crates/application/src/users/authorize_admin.rs
View File

@@ -5,15 +5,6 @@ use uuid::Uuid;
use crate::users::deps::AuthorizeAdminDeps;
/// Whether `user_id` is an admin, for the two extractors (`AdminApiUser`,
/// `AdminUser`) that gate admin-only routes.
///
/// Returns `Ok(None)` when the user row does not exist — a distinct success
/// value, not folded into `DomainError::NotFound`, specifically so callers can
/// tell "row missing" apart from "repository call failed" without depending on
/// which `DomainError` variant a lookup failure happens to produce. Each
/// extractor rejects those two cases differently (missing row is a 401/404,
/// a repository error is a 500) and that distinction must stay exact.
pub async fn execute(
deps: &AuthorizeAdminDeps,
user_id: Uuid,

12
crates/application/src/users/deps.rs
View File

@@ -7,9 +7,6 @@ use domain::ports::{
};
use domain::value_objects::InstanceIdentity;
/// The local half of the former `GetProfileDeps` split (ADR-0004). `user` and
/// `instance` build the always-populated `ProfileIdentity`; `stats`/`diary`/
/// `social_query` are unchanged from before the split.
pub struct GetLocalProfileDeps {
pub stats: Arc<dyn StatsRepository>,
pub diary: Arc<dyn DiaryQuery>,
@@ -18,18 +15,12 @@ pub struct GetLocalProfileDeps {
pub instance: InstanceIdentity,
}
/// The federated half of the former `GetProfileDeps` split. No `user`/`instance` —
/// the federated handler builds identity from the resolved remote actor and never
/// reads a local user row for this path.
pub struct GetFederatedProfileStatsDeps {
pub stats: Arc<dyn StatsRepository>,
pub diary: Arc<dyn DiaryQuery>,
pub social_query: Arc<dyn FollowGraphQuery>,
}
/// Backs page-chrome data (email/role badge, pending-follow badge). See the error
/// policy note on `get_page_viewer::execute`: this deliberately propagates errors
/// rather than degrading internally — the caller decides chrome degrades.
pub struct GetPageViewerDeps {
pub user: Arc<dyn UserRepository>,
pub follow_graph: Arc<dyn FollowGraphQuery>,
@@ -82,9 +73,6 @@ pub struct AuthorizeAdminDeps {
pub user: Arc<dyn UserRepository>,
}
/// `Option` mirrors `Repositories::federated_profile` — genuine optional
/// configuration (federation on/off), not a container-shape workaround. Same
/// exemption as `Services::person_enrichment`.
pub struct GetFederatedProfileDeps {
pub federated_profile: Option<Arc<dyn FederatedProfileQuery>>,
}

12
crates/application/src/users/get_federated_profile.rs
View File

@@ -4,18 +4,6 @@ use uuid::Uuid;
use crate::users::deps::GetFederatedProfileDeps;
/// Looks up a federated (remote) profile by synthetic user id — the fallback
/// path `get_user_profile` (`handlers/users.rs`) takes when the local-profile
/// lookup comes back `NotFound`.
///
/// Federation being disabled (`deps.federated_profile` is `None`) collapses to
/// `Ok(None)`, the same value a present-but-empty lookup would return. That's
/// deliberate: the handler's `if let Ok(Some(fed)) = ...` treats "port absent",
/// "call returned `Ok(None)`", and "call returned `Err`" identically — all three
/// fall through to a 404 — and this use case must not make any of those three
/// distinguishable to a caller that cannot act on the difference. An `Err` from
/// the port is propagated unchanged; the swallow stays at the handler, exactly
/// where it already was.
pub async fn execute(
deps: &GetFederatedProfileDeps,
user_id: Uuid,

4
crates/application/src/users/get_local_profile.rs
View File

@@ -30,10 +30,6 @@ pub struct ProfileIdentity {
pub actor_url: String,
pub avatar_url: Option<String>,
pub banner_url: Option<String>,
/// Only the HTML profile handler reads this (it derives its "display name"
/// from the email's local part, not from `username` — see
/// `handlers/users.rs::get_user_profile_html`). Sourced from the same local
/// `User` row as every other identity field.
pub email: String,
}

13
crates/application/src/users/get_profile_settings.rs
View File

@@ -1,12 +1,3 @@
//! Absorbs BOTH of `handlers/users.rs::get_profile_settings`'s repository calls:
//! the user lookup (`repos.user.find_by_id`) and the profile_fields join
//! (`repos.profile_fields.get_fields`) — plus the avatar/banner URL derivation via
//! `InstanceIdentity`, following the same pattern `get_local_profile` uses.
//!
//! Error policy: profile fields are primary content for this page (it's the page
//! whose entire job is editing them), not page chrome, so a failed join
//! propagates as `Err` rather than silently degrading to an empty list.
use domain::{errors::DomainError, value_objects::UserId};
use uuid::Uuid;
@@ -18,10 +9,6 @@ pub struct ProfileSettings {
pub bio: Option<String>,
pub avatar_url: Option<String>,
pub banner_url: Option<String>,
/// Not in the brief's interface sketch, but the deleted handler code read
/// `user.also_known_as()` and rendered it in `ProfileSettingsTemplate` — this
/// use case absorbs that read too, or the settings page would silently stop
/// showing it.
pub also_known_as: Option<String>,
pub fields: Vec<(String, String)>,
}

8
crates/application/src/users/tests/get_federated_profile_stats.rs
View File

@@ -3,14 +3,6 @@ use crate::users::deps::GetFederatedProfileStatsDeps;
use crate::users::get_federated_profile_stats;
use crate::users::queries::{GetUserProfileQuery, ProfileView};
/// Mirrors exactly how `build_federated_profile_response`
/// (`crates/presentation/src/handlers/users.rs`) calls `execute`: a `user_id` that
/// has no row in the local `users` table (a federated/remote profile uses a
/// synthetic id — see `FederatedProfileQuery`), `is_own_profile: false`,
/// `include_remote: true`. Before ADR-0004's split this was the tolerant branch of
/// the unified `get_profile`; now it's simply this function's only behavior — it
/// never looks at the local `users` table at all, so there is nothing to be
/// tolerant about.
#[tokio::test]
async fn succeeds_for_a_user_id_with_no_local_row() {
let b = TestContextBuilder::new();

8
crates/application/src/users/tests/get_local_profile.rs
View File

@@ -54,10 +54,6 @@ fn unknown_query(user_id: uuid::Uuid) -> GetUserProfileQuery {
}
}
/// `get_local_profile` is the local-only half of the former `get_profile` split
/// (ADR-0004's closed wart). Unlike the old tolerant behavior, a user id with no
/// local row must be `NotFound` here — the empty-string identity sentinel this
/// replaces cannot be represented in `LocalProfileData` at all.
#[tokio::test]
async fn get_local_profile_is_not_found_for_unknown_user() {
let b = TestContextBuilder::new();
@@ -272,10 +268,6 @@ async fn populates_handle_and_actor_url_for_a_local_user() {
let user = user_repo.find_by_email(&email).await.unwrap().unwrap();
let uid = user.id().value();
// `InMemoryUserRepository::update_profile` now actually persists (previously a
// no-op stub) — use it to give this test a real display_name/bio to assert,
// rather than asserting `None` for a mapping that could just as easily be
// wrong, inverted, or hardcoded.
let update_deps = UpdateProfileDeps {
user: user_repo.clone(),
object_storage: b.object_storage.clone(),

3
crates/application/src/users/tests/get_page_viewer.rs
View File

@@ -34,9 +34,6 @@ async fn setup_user(b: &TestContextBuilder, email: &str, username: &str) -> uuid
user.id().value()
}
/// `get_page_viewer` absorbs `helpers.rs::build_page_context`'s inline pending-count
/// computation. It must reuse `FollowGraphQuery::count_pending_followers` rather than
/// re-deriving the count from raw follow rows.
#[tokio::test]
async fn get_page_viewer_reports_pending_follower_count() {
let b = TestContextBuilder::new();

4
crates/application/src/users/tests/get_profile_settings.rs
View File

@@ -8,10 +8,6 @@ use crate::test_helpers::TestContextBuilder;
use crate::users::deps::GetProfileSettingsDeps;
use crate::users::get_profile_settings;
/// `get_profile_settings` absorbs BOTH of the handler's calls: the user lookup
/// (`repos.user.find_by_id`) and the profile_fields join
/// (`repos.profile_fields.get_fields`). This proves the join actually moved in,
/// not just the user lookup.
#[tokio::test]
async fn get_profile_settings_includes_profile_fields() {
let b = TestContextBuilder::new();

3
crates/application/src/users/tests/resolve_username_to_id.rs
View File

@@ -4,9 +4,6 @@ use crate::test_helpers::TestContextBuilder;
use crate::users::deps::ResolveUsernameDeps;
use crate::users::resolve_username_to_id;
/// `resolve_username_to_id` absorbs `handlers/users.rs::get_user_by_username`'s
/// `repos.user.find_by_username()` call. Unknown usernames must resolve to `None`,
/// not an error — the handler turns `None` into a 404, same as today.
#[tokio::test]
async fn resolve_username_to_id_returns_none_for_unknown() {
let b = TestContextBuilder::new();

16
crates/application/src/watchlist/get_watchlist_for_owner.rs
View File

@@ -9,22 +9,11 @@ use crate::watchlist::deps::{GetWatchlistDeps, GetWatchlistForOwnerDeps};
use crate::watchlist::get;
use crate::watchlist::queries::GetWatchlistQuery;
/// Which data source answered a watchlist page request. The former handler
/// decided this by probing for a local user row — that decision now lives here,
/// not in `handlers/watchlist.rs`.
pub enum WatchlistView {
Local(Paginated<WatchlistWithMovie>),
Remote(Vec<RemoteWatchlistEntry>),
}
/// `limit`/`offset` are taken raw, not as a pre-built `PageParams`, on purpose:
/// they're only validated on the local branch, by delegating to
/// `watchlist::get::execute` (whose `PageParams::new` call is where the validation
/// actually happens). The remote branch ignores them entirely, unconditionally
/// fetching the full federated watchlist. The delegation call below must stay
/// inside the `if is_local` branch — calling it before the local/remote decision
/// would turn an out-of-range `limit` into a 400 on a remote owner's page, which
/// never happened before and must not start happening now.
pub async fn execute(
deps: &GetWatchlistForOwnerDeps,
owner_id: Uuid,
@@ -33,9 +22,6 @@ pub async fn execute(
) -> Result<WatchlistView, DomainError> {
let user_id = UserId::from_uuid(owner_id);
// Matches the deleted handler's `.map(|u| u.is_some()).unwrap_or(false)`
// exactly: a lookup error is treated the same as "no local row", falling
// through to the remote arm rather than propagating.
let is_local = deps
.user
.find_by_id(&user_id)
@@ -58,8 +44,6 @@ pub async fn execute(
.await?;
Ok(WatchlistView::Local(paginated))
} else {
// Matches the deleted handler's `.unwrap_or_default()`: a federation
// lookup error yields an empty list, not a propagated error.
let entries = deps
.remote_watchlist
.get_by_derived_uuid(owner_id)

12
crates/application/src/watchlist/tests/get_watchlist_for_owner.rs
View File

@@ -44,8 +44,6 @@ async fn watchlist_for_owner_returns_local_when_user_row_exists() {
let user = b.user_repo.find_by_email(&email).await.unwrap().unwrap();
let uid = user.id().value();
// If the local-row decision were wrong and this reached the remote arm, the
// panic port below would fail the test instead of silently passing.
let deps = GetWatchlistForOwnerDeps {
user: b.user_repo.clone(),
watchlist: InMemoryWatchlistRepository::new(),
@@ -75,9 +73,6 @@ async fn watchlist_for_owner_returns_remote_when_no_local_row() {
},
]);
// No user row exists for `owner_id` in this fresh, empty user repo — the
// local-vs-remote decision must fall through to the remote arm. The panic
// port below proves the local arm is never touched.
let deps = GetWatchlistForOwnerDeps {
user: InMemoryUserRepository::new(),
watchlist: Arc::new(PanicWatchlistRepository),
@@ -97,10 +92,6 @@ async fn watchlist_for_owner_returns_remote_when_no_local_row() {
}
}
/// Pins a pre-existing quirk: the old handler validated `limit`/`offset` only on
/// the local branch (inside `watchlist::get::execute`'s `PageParams::new` call) —
/// the remote branch never looked at them at all. An out-of-range limit must
/// still surface as a validation error on the local arm.
#[tokio::test]
async fn watchlist_for_owner_local_arm_validates_limit() {
let b = TestContextBuilder::new();
@@ -126,9 +117,6 @@ async fn watchlist_for_owner_local_arm_validates_limit() {
));
}
/// Same quirk, other direction: an out-of-range limit is silently ignored on the
/// remote arm, exactly as before — it must not turn into a validation error just
/// because the use case now owns the branch decision.
#[tokio::test]
async fn watchlist_for_owner_remote_arm_ignores_invalid_limit() {
let owner_id = Uuid::new_v4();

17
crates/application/src/wrapup/get_ready_report.rs
View File

@@ -6,23 +6,6 @@ use domain::{
use crate::wrapup::deps::GetReadyReportDeps;
/// Absorbs the year -> date-range conversion and the `status == Ready` +
/// `report.is_some()` gating both HTML wrap-up handlers used to do by hand.
///
/// Takes `domain::models::wrapup::WrapUpScope` directly (the same type
/// `wrapup::handle_requested` already matches on) rather than a second,
/// use-case-local `WrapUpScope` — there is exactly one concept here, one type.
/// `None` can no longer be silently passed for "global": callers must pick
/// `WrapUpScope::User(uid)` or `WrapUpScope::Global` explicitly.
///
/// Error shape mirrors the deleted handlers exactly:
/// - An unrepresentable `year` (can't form a valid `NaiveDate`) is
/// `ValidationError` — the handlers used this to return a bare 400.
/// - Everything else — no record, a record whose `status` isn't `Ready`, a
/// `Ready` record with `report: None`, or even a repository error — collapses
/// to `NotFound`, matching the old handlers' `_ => return StatusCode::NOT_FOUND`
/// catch-all (which really did swallow repo errors into a 404, not just the
/// "not ready yet" cases; preserved here rather than tidied).
pub async fn execute(
deps: &GetReadyReportDeps,
scope: WrapUpScope,

4
crates/application/src/wrapup/tests/get_ready_report.rs
View File

@@ -133,10 +133,6 @@ async fn ready_report_global_scope_ignores_user_id() {
.unwrap();
}
/// The old handlers guarded `NaiveDate::from_ymd_opt` up front and returned a bare
/// 400 for a year that can't form a valid date — that guard moved into the use
/// case along with the rest of the year->date-range conversion. A distinguishable
/// error (ValidationError, not NotFound) lets the handler still map it to 400.
#[tokio::test]
async fn ready_report_is_validation_error_for_unrepresentable_year() {
let repo = InMemoryWrapUpRepository::new();

11
crates/composition/src/build.rs
View File

@@ -2,8 +2,9 @@ use application::auth::deps::{
LoginDeps, LogoutDeps, RefreshDeps, RegisterAndLoginDeps, RegisterDeps,
};
use application::deps::{
AuthGroup, Deps, DiaryGroup, GoalsGroup, ImportGroup, IntegrationsGroup, MoviesGroup,
PersonGroup, SearchGroup, SocialGroup, UsersGroup, WatchlistGroup, WorkerDeps, WrapupGroup,
AuthGroup, Deps, DiaryGroup, GoalsGroup, ImportGroup, InsightsGroup, IntegrationsGroup,
MoviesGroup, PersonGroup, SearchGroup, SocialGroup, UsersGroup, WatchlistGroup, WorkerDeps,
WrapupGroup,
};
use application::diary::deps::{
DeleteReviewDeps, EditReviewDeps, ExportDiaryDeps, GetActivityFeedDeps, GetDiaryDeps,
@@ -15,6 +16,7 @@ use application::import::deps::{
DeleteImportProfileDeps, ExecuteImportDeps, GetMappingStageDeps, GetPreviewStageDeps,
GetSessionStateDeps, ListImportProfilesDeps, SaveProfileDeps,
};
use application::insights::deps::GetInsightsDeps;
use application::integrations::deps::{
ConfirmWatchEventsDeps, DismissWatchEventsDeps, GenerateWebhookTokenDeps, GetWatchQueueDeps,
GetWebhookTokensDeps, IngestWatchEventDeps, RevokeWebhookTokenDeps,
@@ -175,6 +177,11 @@ pub fn build_deps(
import_profile: repos.import_profile.clone(),
},
},
insights: InsightsGroup {
get: GetInsightsDeps {
insights: repos.insights.clone(),
},
},
integrations: IntegrationsGroup {
ingest_watch_event: IngestWatchEventDeps {
webhook_token: repos.webhook_token.clone(),

3
crates/composition/src/factory.rs
View File

@@ -32,6 +32,7 @@ pub struct DatabaseOutput {
pub ap_content: Arc<dyn LocalApContentQuery>,
pub wrapup_stats: Arc<dyn domain::ports::WrapUpStatsQuery>,
pub wrapup_repo: Arc<dyn domain::ports::WrapUpRepository>,
pub insights: Arc<dyn domain::ports::InsightsQuery>,
pub goal_command: Arc<dyn domain::ports::GoalCommand>,
pub goal_query: Arc<dyn domain::ports::GoalQuery>,
pub user_settings: Arc<dyn domain::ports::UserSettingsRepository>,
@@ -91,6 +92,7 @@ pub async fn build_database_adapters(
ap_content: w.ap_content,
wrapup_stats: w.wrapup_stats,
wrapup_repo: w.wrapup_repo,
insights: w.insights,
goal_command: w.goal_command,
goal_query: w.goal_query,
user_settings: w.user_settings,
@@ -145,6 +147,7 @@ pub async fn build_database_adapters(
ap_content: w.ap_content,
wrapup_stats: w.wrapup_stats,
wrapup_repo: w.wrapup_repo,
insights: w.insights,
goal_command: w.goal_command,
goal_query: w.goal_query,
user_settings: w.user_settings,

13
crates/composition/src/repositories.rs
View File

@@ -2,12 +2,12 @@ use std::sync::Arc;
use domain::ports::{
BlockQuery, DiaryQuery, FederatedProfileQuery, FederationAdminQuery, FollowGraphQuery,
GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository, MovieCommand,
MovieProfileRepository, MovieQuery, PersonCommand, PersonQuery, RefreshSessionRepository,
RemoteGoalRepository, RemoteWatchlistRepository, ReviewRepository, SearchCommand, SearchPort,
SocialCommand, StatsRepository, UserProfileFieldsRepository, UserRepository,
UserSettingsRepository, WatchEventCommand, WatchEventQuery, WatchlistRepository,
WebhookTokenRepository, WrapUpRepository, WrapUpStatsQuery,
GoalCommand, GoalQuery, ImportProfileRepository, ImportSessionRepository, InsightsQuery,
MovieCommand, MovieProfileRepository, MovieQuery, PersonCommand, PersonQuery,
RefreshSessionRepository, RemoteGoalRepository, RemoteWatchlistRepository, ReviewRepository,
SearchCommand, SearchPort, SocialCommand, StatsRepository, UserProfileFieldsRepository,
UserRepository, UserSettingsRepository, WatchEventCommand, WatchEventQuery,
WatchlistRepository, WebhookTokenRepository, WrapUpRepository, WrapUpStatsQuery,
};
#[derive(Clone)]
@@ -37,6 +37,7 @@ pub struct Repositories {
pub federation_admin: Arc<dyn FederationAdminQuery>,
pub wrapup_stats: Arc<dyn WrapUpStatsQuery>,
pub wrapup_repo: Arc<dyn WrapUpRepository>,
pub insights: Arc<dyn InsightsQuery>,
pub goal_command: Arc<dyn GoalCommand>,
pub goal_query: Arc<dyn GoalQuery>,
pub user_settings: Arc<dyn UserSettingsRepository>,

16
crates/composition/src/tests/build.rs
View File

@@ -5,13 +5,13 @@ use domain::testing::{
FakeAuthService, FakeDocumentParser, FakeMetadataClient, FakePasswordHasher, FakePersonQuery,
FakePosterFetcher, FakeSearchCommand, FakeSearchPort, FakeStatsRepository,
InMemoryGoalRepository, InMemoryImportProfileRepository, InMemoryImportSessionRepository,
InMemoryMovieProfileRepository, InMemoryMovieRepository, InMemoryProfileFieldsRepo,
InMemoryRefreshSessionRepository, InMemoryReviewRepository, InMemorySocialRepository,
InMemoryUserRepository, InMemoryUserSettingsRepository, InMemoryWatchEventRepository,
InMemoryWatchlistRepository, InMemoryWebhookTokenRepository, InMemoryWrapUpRepository,
InMemoryWrapUpStatsQuery, NoopEventPublisher, NoopFederationAdminQuery, NoopObjectStorage,
NoopRemoteGoalRepository, NoopRemoteWatchlistRepository, PanicDiaryExporter,
PanicPersonCommand,
InMemoryInsightsQuery, InMemoryMovieProfileRepository, InMemoryMovieRepository,
InMemoryProfileFieldsRepo, InMemoryRefreshSessionRepository, InMemoryReviewRepository,
InMemorySocialRepository, InMemoryUserRepository, InMemoryUserSettingsRepository,
InMemoryWatchEventRepository, InMemoryWatchlistRepository, InMemoryWebhookTokenRepository,
InMemoryWrapUpRepository, InMemoryWrapUpStatsQuery, NoopEventPublisher,
NoopFederationAdminQuery, NoopObjectStorage, NoopRemoteGoalRepository,
NoopRemoteWatchlistRepository, PanicDiaryExporter, PanicPersonCommand,
};
use crate::{DatabaseOutput, Repositories, build::build_deps, build::build_worker_deps};
@@ -254,6 +254,7 @@ fn test_repositories() -> Repositories {
federation_admin: Arc::new(NoopFederationAdminQuery),
wrapup_stats: InMemoryWrapUpStatsQuery::new(),
wrapup_repo: InMemoryWrapUpRepository::new(),
insights: InMemoryInsightsQuery::new(),
goal_command: Arc::clone(&goals) as _,
goal_query: goals as _,
user_settings: InMemoryUserSettingsRepository::new(),
@@ -312,6 +313,7 @@ fn test_database_output() -> DatabaseOutput {
ap_content: Arc::new(NoopLocalApContentQuery),
wrapup_stats: InMemoryWrapUpStatsQuery::new(),
wrapup_repo: InMemoryWrapUpRepository::new(),
insights: InMemoryInsightsQuery::new(),
goal_command: Arc::clone(&goals) as _,
goal_query: goals as _,
user_settings: Arc::clone(&user_settings) as _,

7
crates/domain/src/models/federation.rs
View File

@@ -39,10 +39,6 @@ pub struct FederatedProfile {
pub banner_url: Option<String>,
}
/// A domain-blocklist entry as presented to admin surfaces. Mirrors what the
/// federation adapter can supply; `blocked_at` is a pre-formatted string
/// because that is what the underlying store returns and every consumer
/// renders it verbatim.
#[derive(Debug, Clone)]
pub struct BlockedDomainInfo {
pub domain: String,
@@ -50,9 +46,6 @@ pub struct BlockedDomainInfo {
pub blocked_at: String,
}
/// A remote actor the local instance follows, reduced to the two fields any
/// consumer actually reads. Deliberately narrower than the federation
/// library's actor type — widening it is a decision, not an oversight.
#[derive(Debug, Clone)]
pub struct FollowedActorInfo {
pub url: String,

66
crates/domain/src/models/insights.rs Normal file
View File

@@ -0,0 +1,66 @@
use chrono::NaiveDateTime;
use uuid::Uuid;
/// One denormalized row per Review — everything the insight analyzer needs,
/// nothing it doesn't (ADR-0010). Lighter than `WrapUpMovieRow` on purpose:
/// WrapUp's row is a report-era artifact with poster paths, budgets and cast
/// that Insights must not inherit.
#[derive(Clone, Debug)]
pub struct WatchRow {
pub movie_id: Uuid,
pub rating: u8,
pub watched_at: NaiveDateTime,
pub watch_medium: Option<String>,
pub genres: Vec<String>,
}
/// A single conservative, baseline-relative statement about a user's viewing
/// behavior (see CONTEXT.md "Insight"). Every variant carries the numbers the
/// SPA needs to render the claim plus the `sample_count` behind it, so the UI
/// can show why a category is empty without another round trip.
///
/// Ordering contract: `build_insights` returns insights grouped by category in
/// a fixed order (rating lift, seasonality, medium timing, rhythm), strongest
/// first within each category, at most `MAX_PER_CATEGORY` per category.
#[derive(Clone, Debug, PartialEq)]
pub enum Insight {
/// The genre's average rating sits `delta` stars away from the user's own
/// average. First-time watches only — a rewatch is chosen because it is
/// already liked, so counting it would conflate choice with genre.
RatingLift {
genre: String,
genre_avg: f64,
user_avg: f64,
delta: f64,
sample_count: u32,
},
/// The genre's share of watches in one month of the year is `multiplier`
/// times the user's own share of watches in that month.
Seasonality {
genre: String,
/// 1–12
month: u32,
multiplier: f64,
sample_count: u32,
},
/// The medium's weekend share of watches is `multiplier` times the user's
/// overall weekend share.
MediumTiming {
medium: String,
multiplier: f64,
weekend_share: f64,
baseline_share: f64,
sample_count: u32,
},
/// When watching actually happens: the 3-hour window and the weekdays that
/// carry the most real-timestamped watches.
Rhythm {
/// 0–23, start of the peak 3-hour window.
peak_hour_start: u32,
/// 0–23, exclusive end of the peak window (`start + 3`, mod 24).
peak_hour_end: u32,
/// Chrono weekday numbers, strongest first: Mon=1 … Sun=7.
top_days: Vec<u32>,
sample_count: u32,
},
}

2
crates/domain/src/models/mod.rs
View File

@@ -12,6 +12,7 @@ pub mod goal;
pub mod import;
pub mod import_profile;
pub mod import_session;
pub mod insights;
pub mod person;
pub mod remote_goal;
pub mod remote_watchlist;
@@ -24,6 +25,7 @@ pub mod wrapup;
pub use enrichment::*;
pub use federation::*;
pub use feed::*;
pub use insights::*;
pub use movie::*;
pub use review::*;
pub use stats::{

1
crates/domain/src/models/person.rs
View File

@@ -12,7 +12,6 @@ impl ExternalPersonId {
&self.0
}
/// Parse the TMDb numeric ID from "tmdb:12345". Returns None for other formats.
pub fn tmdb_id(&self) -> Option<i64> {
self.0.strip_prefix("tmdb:").and_then(|s| s.parse().ok())
}

2
crates/domain/src/models/review.rs
View File

@@ -28,7 +28,6 @@ pub struct PersistedReview {
pub watch_medium: Option<WatchMedium>,
}
/// Partial update for a review. `None` = unchanged, `Some(None)` = clear, `Some(Some(v))` = set.
#[derive(Clone, Debug, Default)]
pub struct ReviewEdit {
pub rating: Option<Rating>,
@@ -133,7 +132,6 @@ impl Review {
pub fn watch_medium(&self) -> Option<&WatchMedium> {
self.watch_medium.as_ref()
}
/// Returns [star1_filled, star2_filled, ..., star5_filled]
pub fn stars(&self) -> [bool; 5] {
let r = self.rating.value();
[r >= 1, r >= 2, r >= 3, r >= 4, r >= 5]

5
crates/domain/src/models/search.rs
View File

@@ -42,13 +42,9 @@ pub struct PersonSearchHit {
pub name: String,
pub known_for_department: Option<String>,
pub profile_path: Option<String>,
/// Top movie titles this person is known for — populated at query time
/// by joining relational tables, never from the index.
pub known_for_titles: Vec<String>,
}
/// Document submitted to the search index.
/// Add a new variant here to make a new entity type searchable — the port never changes.
pub enum IndexableDocument {
Movie {
id: MovieId,
@@ -58,7 +54,6 @@ pub enum IndexableDocument {
Person {
id: PersonId,
person: Box<Person>,
// known_for_titles intentionally absent — no reads inside a command flow
},
}

3
crates/domain/src/ports/events.rs
View File

@@ -11,9 +11,6 @@ pub trait EventPublisher: Send + Sync {
}
pub trait EventConsumer: Send + Sync {
/// Returns a stream of event envelopes. Each envelope carries a domain event
/// and an ack handle — callers ack after successful dispatch, nack on failure.
/// Implementations decide transport (NATS, DB queue, in-memory channel).
fn consume(&self) -> futures::stream::BoxStream<'_, Result<EventEnvelope, DomainError>>;
}

6
crates/domain/src/ports/federation.rs
View File

@@ -6,8 +6,6 @@ use crate::{
models::{BlockedDomainInfo, FollowedActorInfo},
};
/// Serves ActivityPub documents for content negotiation. Presentation calls
/// this when a peer sends `Accept: application/activity+json`.
#[async_trait]
pub trait ApDocumentPort: Send + Sync {
async fn actor_json(&self, user_id: &str) -> Result<String, DomainError>;
@@ -23,8 +21,6 @@ pub trait ApDocumentPort: Send + Sync {
) -> Result<String, DomainError>;
}
/// Instance-wide domain blocklist administration. Presentation calls this from
/// the admin API and the admin HTML pages; nothing outside presentation does.
#[async_trait]
pub trait InstanceBlocklistPort: Send + Sync {
async fn get_blocked_domains(&self) -> Result<Vec<BlockedDomainInfo>, DomainError>;
@@ -36,8 +32,6 @@ pub trait InstanceBlocklistPort: Send + Sync {
async fn remove_blocked_domain(&self, domain: &str) -> Result<(), DomainError>;
}
/// Pulling remote content in, and pushing local content out, after a follow
/// is established. Worker-side only — no HTTP handler calls this.
#[async_trait]
pub trait ApBackfillPort: Send + Sync {
async fn get_following(

1
crates/domain/src/ports/images.rs
View File

@@ -4,7 +4,6 @@ use crate::{errors::DomainError, value_objects::PosterUrl};
#[async_trait]
pub trait ObjectStorage: Send + Sync {
/// Stores `image_bytes` at `key` and returns the stored key.
async fn store(&self, key: &str, image_bytes: &[u8]) -> Result<String, DomainError>;
async fn get(&self, key: &str) -> Result<Vec<u8>, DomainError>;
async fn get_stream(

12
crates/domain/src/ports/insights.rs Normal file
View File

@@ -0,0 +1,12 @@
use async_trait::async_trait;
use crate::{errors::DomainError, models::insights::WatchRow, value_objects::UserId};
/// Read side of Insights (ADR-0010): fetches one denormalized `WatchRow` per
/// review across the user's whole diary. All thresholds and baseline
/// comparisons live in the pure `insights_analyzer`, never in adapters —
/// SQL's job here is "rows in, rows out" and nothing else.
#[async_trait]
pub trait InsightsQuery: Send + Sync {
async fn get_watch_rows(&self, user_id: &UserId) -> Result<Vec<WatchRow>, DomainError>;
}

2
crates/domain/src/ports/media_server.rs
View File

@@ -12,7 +12,6 @@ pub trait MediaServerParser: Send + Sync {
-> Result<Option<ParsedPlaybackEvent>, DomainError>;
}
/// Write port — mutates watch events.
#[async_trait]
pub trait WatchEventCommand: Send + Sync {
async fn save(&self, event: &WatchEvent) -> Result<(), DomainError>;
@@ -32,7 +31,6 @@ pub trait WatchEventCommand: Send + Sync {
) -> Result<u64, DomainError>;
}
/// Read port — queries watch events. No mutations.
#[async_trait]
pub trait WatchEventQuery: Send + Sync {
async fn list_pending(&self, user_id: &UserId) -> Result<Vec<WatchEvent>, DomainError>;

2
crates/domain/src/ports/mod.rs
View File

@@ -8,6 +8,7 @@ pub mod goals;
pub mod image_fetcher;
pub mod images;
pub mod import;
pub mod insights;
pub mod jobs;
pub mod media_server;
pub mod movie;
@@ -29,6 +30,7 @@ pub use goals::*;
pub use image_fetcher::*;
pub use images::*;
pub use import::*;
pub use insights::*;
pub use jobs::*;
pub use media_server::*;
pub use movie::*;

9
crates/domain/src/ports/movie.rs
View File

@@ -9,14 +9,12 @@ use crate::{
value_objects::{ExternalMetadataId, MovieId, MovieTitle, PosterUrl, ReleaseYear},
};
/// Write port — mutates the movies table.
#[async_trait]
pub trait MovieCommand: Send + Sync {
async fn upsert_movie(&self, movie: &Movie) -> Result<(), DomainError>;
async fn delete_movie(&self, movie_id: &MovieId) -> Result<(), DomainError>;
}
/// Read port — queries movies. No mutations.
#[async_trait]
pub trait MovieQuery: Send + Sync {
async fn get_movie_by_external_id(
@@ -42,7 +40,6 @@ pub trait MovieQuery: Send + Sync {
page: &PageParams,
filter: &MovieFilter,
) -> Result<Paginated<MovieSummary>, DomainError>;
/// Returns all movies that have an external_metadata_id set. Used for deduplication.
async fn list_movies_with_external_id(&self) -> Result<Vec<Movie>, DomainError>;
}
@@ -62,14 +59,11 @@ pub trait MetadataClient: Send + Sync {
pub trait MovieProfileRepository: Send + Sync {
async fn upsert(&self, profile: &MovieProfile) -> Result<(), DomainError>;
async fn get_by_movie_id(&self, id: &MovieId) -> Result<Option<MovieProfile>, DomainError>;
/// Returns (movie_id, external_metadata_id) for movies with no profile or a stale one
/// (enriched_at older than 30 days).
async fn list_stale(&self) -> Result<Vec<(MovieId, String)>, DomainError>;
}
#[async_trait]
pub trait MovieEnrichmentClient: Send + Sync {
/// Resolves an external ID (TMDb or IMDb) and fetches the full movie profile.
async fn fetch_profile(
&self,
movie_id: MovieId,
@@ -79,9 +73,6 @@ pub trait MovieEnrichmentClient: Send + Sync {
#[async_trait]
pub trait MovieDeduplicator: Send + Sync {
/// Atomically re-points all foreign keys (reviews, watchlist entries, movie profiles)
/// from `old_id` to `canonical_id`, upserts the canonical movie record, then deletes
/// the old duplicate. Returns the number of rows re-pointed across all tables.
async fn merge_into_canonical(
&self,
old_id: &MovieId,

5
crates/domain/src/ports/noop.rs
View File

@@ -7,7 +7,6 @@ use crate::{
// ── NoopRemoteWatchlistRepository ─────────────────────────────────────────────
/// Stub used when federation is disabled — every operation is a no-op.
pub struct NoopRemoteWatchlistRepository;
#[async_trait]
@@ -113,7 +112,6 @@ impl super::BlockQuery for NoopSocialQuery {
// ── NoopFederationAdminQuery ─────────────────────────────────────────────────
/// Stub used when federation is disabled — returns empty results.
pub struct NoopFederationAdminQuery;
#[async_trait]
@@ -127,7 +125,6 @@ impl super::FederationAdminQuery for NoopFederationAdminQuery {
// ── NoopApDocument ───────────────────────────────────────────────────────────
/// Stub used when federation is disabled — every operation is a no-op.
pub struct NoopApDocument;
#[async_trait]
@@ -153,7 +150,6 @@ impl super::ApDocumentPort for NoopApDocument {
// ── NoopInstanceBlocklist ────────────────────────────────────────────────────
/// Stub used when federation is disabled — every operation is a no-op.
pub struct NoopInstanceBlocklist;
#[async_trait]
@@ -173,7 +169,6 @@ impl super::InstanceBlocklistPort for NoopInstanceBlocklist {
// ── NoopApBackfill ───────────────────────────────────────────────────────────
/// Stub used when federation is disabled — every operation is a no-op.
pub struct NoopApBackfill;
#[async_trait]

8
crates/domain/src/ports/person.rs
View File

@@ -10,13 +10,9 @@ pub trait PersonEnrichmentClient: Send + Sync {
async fn fetch_details(&self, external_id: &str) -> Result<PersonEnrichmentData, DomainError>;
}
/// Write port — mutates the persons table. No reads.
#[async_trait]
pub trait PersonCommand: Send + Sync {
/// Upsert a batch of persons. Uses INSERT OR REPLACE (SQLite) / ON CONFLICT DO UPDATE (Postgres).
async fn upsert_batch(&self, persons: &[Person]) -> Result<(), DomainError>;
/// Insert a batch of missing persons from movie_cast/movie_crew into the persons table.
/// Returns (inserted_count, has_more).
async fn backfill_from_credits_batch(
&self,
batch_size: u32,
@@ -28,7 +24,6 @@ pub trait PersonCommand: Send + Sync {
) -> Result<(), DomainError>;
}
/// Read port — queries persons and credits. No mutations.
#[async_trait]
pub trait PersonQuery: Send + Sync {
async fn get_by_id(&self, id: &PersonId) -> Result<Option<Person>, DomainError>;
@@ -36,10 +31,7 @@ pub trait PersonQuery: Send + Sync {
&self,
id: &ExternalPersonId,
) -> Result<Option<Person>, DomainError>;
/// Returns the person's full cast and crew credit history across all indexed movies.
async fn get_credits(&self, id: &PersonId) -> Result<PersonCredits, DomainError>;
/// Returns persons who have no remaining entries in movie_cast or movie_crew.
/// Called after movie deletion to find index entries that can be pruned.
async fn list_orphaned_persons(&self) -> Result<Vec<PersonId>, DomainError>;
async fn list_page(&self, limit: u32, offset: u32) -> Result<Vec<Person>, DomainError>;
}

4
crates/domain/src/ports/search.rs
View File

@@ -5,17 +5,13 @@ use crate::{
models::{EntityType, IndexableDocument, SearchQuery, SearchResults},
};
/// Read port — executes search queries. No mutations.
#[async_trait]
pub trait SearchPort: Send + Sync {
async fn search(&self, query: &SearchQuery) -> Result<SearchResults, DomainError>;
}
/// Write port — manages the search index. No reads.
#[async_trait]
pub trait SearchCommand: Send + Sync {
/// Add or replace a document in the search index.
async fn index(&self, doc: IndexableDocument) -> Result<(), DomainError>;
/// Remove a document from the search index by entity type and internal ID string.
async fn remove(&self, entity_type: EntityType, id: &str) -> Result<(), DomainError>;
}

35
crates/domain/src/ports/social.rs
View File

@@ -127,35 +127,11 @@ pub trait LocalApContentQuery: Send + Sync {
) -> Result<Vec<DiaryEntry>, DomainError>;
}
/// Resolves a `FollowTarget` (a handle or an already-known identity) to the
/// `SocialIdentity` that should be dispatched on — local vs. remote.
///
/// Split out from `SocialCommand::follow` because the decision must happen
/// once, before dispatch, and both `LocalSocialService` and
/// `CompositeSocialAdapter` need to observe its result rather than each
/// re-deriving it (which would need `UserRepository` on the composite and
/// duplicate the local/remote fallthrough logic).
#[async_trait]
pub trait FollowTargetResolver: Send + Sync {
async fn resolve_target(&self, target: &FollowTarget) -> Result<SocialIdentity, DomainError>;
}
/// Performs the local follow write for a target whose `SocialIdentity` a caller
/// has *already* resolved via `FollowTargetResolver::resolve_target`.
///
/// Exists so `CompositeSocialAdapter` — which must call `resolve_target` first
/// to decide local vs. remote dispatch — doesn't then hand the raw
/// `FollowTarget` to `SocialCommand::follow` and pay for a second resolution of
/// the same handle. That second resolution wasn't just wasteful: on a
/// federation-ON deployment it could also change the answer (the target user
/// deleted between the two calls, e.g.), turning a local follow into a bogus
/// "federation is not enabled" error.
///
/// `LocalSocialService::follow` is `resolve_target_identity` then
/// `follow_resolved` — the self-follow guard and the two-sided write live only
/// in the latter, so a caller that already has the identity (the composite)
/// and one that doesn't (federation-off `SocialCommand::follow`, wired
/// directly) both end up running the exact same write path.
#[async_trait]
pub trait ResolvedFollow: Send + Sync {
async fn follow_resolved(
@@ -165,15 +141,6 @@ pub trait ResolvedFollow: Send + Sync {
) -> Result<(), DomainError>;
}
/// The subset of social behavior that needs no ActivityPub — everything a
/// single-instance deployment can do with only its own database.
///
/// A marker supertrait with a blanket impl, so any type implementing all five
/// parts is usable as `Arc<dyn LocalSocial>` without a separate registration.
/// Same shape as `k_ap::FollowRepository` over its five follow traits.
///
/// `CompositeSocialAdapter` takes one of these for its local branches, which is
/// what lets the `activitypub` crate stay unaware of `application`.
pub trait LocalSocial:
SocialCommand + FollowGraphQuery + BlockQuery + FollowTargetResolver + ResolvedFollow
{
@@ -333,8 +300,6 @@ mod local_social_tests {
}
}
/// The blanket impl must make any type implementing all five usable as
/// `Arc<dyn LocalSocial>`, and `LocalSocial` must stay object-safe.
#[test]
fn blanket_impl_yields_a_trait_object() {
let local: Arc<dyn LocalSocial> = Arc::new(Stub);

3
crates/domain/src/ports/watchlist.rs
View File

@@ -11,13 +11,10 @@ use crate::{
#[async_trait]
pub trait WatchlistRepository: Send + Sync {
/// Add a new entry. Silently succeeds if the entry already exists.
async fn add(&self, entry: &WatchlistEntry) -> Result<(), DomainError>;
/// Remove an entry. Returns NotFound if the entry does not exist.
async fn remove(&self, user_id: &UserId, movie_id: &MovieId) -> Result<(), DomainError>;
/// Remove an entry if it exists. Never returns NotFound.
async fn remove_if_present(
&self,
user_id: &UserId,

348
crates/domain/src/services/insights_analyzer.rs Normal file
View File

@@ -0,0 +1,348 @@
//! Conservative insight extraction over a user's whole diary (ADR-0010).
//!
//! Every threshold lives here, not in adapters, so the honesty rules are
//! unit-testable without a database. An insight that cannot clear its
//! thresholds is silently absent — an empty page is the correct rendering of
//! a small or noisy diary, not a bug to fix by lowering thresholds.
use std::collections::HashMap;
use chrono::{Datelike, NaiveDateTime, TimeZone, Timelike, Weekday};
use crate::models::{Insight, WatchRow};
// ── Honesty thresholds ───────────────────────────────────────────────────────
/// First-time ratings a genre needs before its average means anything.
const MIN_GENRE_RATINGS: usize = 5;
/// First-time ratings overall before the user's own average is a stable baseline.
const MIN_BASELINE_RATINGS: usize = 10;
/// Minimum distance from the user's average before a rating lift is worth saying.
const MIN_RATING_DELTA: f64 = 0.3;
/// Rows a genre needs before month-of-year distribution is more than noise.
const MIN_GENRE_SEASON_ROWS: usize = 15;
/// Minimum lift over the user's own share of watches in that month.
const MIN_SEASON_MULTIPLIER: f64 = 1.5;
/// Genre watches the peak month needs — a lift built on one or two rows in a
/// quiet month is noise, however large the ratio.
const MIN_SEASON_PEAK_ROWS: u32 = 4;
/// Rows a medium needs before its weekend share means anything.
const MIN_MEDIUM_ROWS: usize = 5;
/// Minimum lift over the user's own weekend share.
const MIN_MEDIUM_MULTIPLIER: f64 = 1.5;
/// More than this share of rows without a medium → the medium-timing category
/// would describe half a diary; skip it entirely.
const MAX_MISSING_MEDIUM_SHARE: f64 = 0.3;
/// Real-timestamped rows needed to say anything about rhythm.
const MIN_RHYTHM_ROWS: usize = 10;
/// More than this share of rows at exactly 00:00 → timestamps are import noise
/// (CSV imports carry no time); skip rhythm entirely.
const MAX_MIDNIGHT_SHARE: f64 = 0.5;
/// Insights shown per category — more than this and the page is a wall.
const MAX_PER_CATEGORY: usize = 3;
/// Width of the peak-hour window in the rhythm insight.
const RHYTHM_WINDOW_HOURS: u32 = 3;
/// `rows` carry `watched_at` in UTC as stored; `tz` is the viewer's zone, so
/// months, weekdays and hours are bucketed as the user lived them.
pub fn build_insights<Tz: TimeZone>(rows: &[WatchRow], tz: &Tz) -> Vec<Insight> {
let mut insights = Vec::new();
insights.extend(rating_lifts(rows));
insights.extend(seasonalities(rows, tz));
insights.extend(medium_timings(rows, tz));
insights.extend(rhythm(rows, tz));
insights
}
fn local<Tz: TimeZone>(tz: &Tz, utc: &NaiveDateTime) -> NaiveDateTime {
tz.from_utc_datetime(utc).naive_local()
}
// ── Rating lift ──────────────────────────────────────────────────────────────
fn rating_lifts(rows: &[WatchRow]) -> Vec<Insight> {
let first_time = first_time_rows(rows);
if first_time.len() < MIN_BASELINE_RATINGS {
return vec![];
}
let user_avg = avg_rating(&first_time);
let mut by_genre: HashMap<&str, Vec<u8>> = HashMap::new();
for row in first_time {
for genre in &row.genres {
by_genre.entry(genre).or_default().push(row.rating);
}
}
let mut candidates: Vec<Insight> = by_genre
.into_iter()
.filter(|(_, ratings)| ratings.len() >= MIN_GENRE_RATINGS)
.filter_map(|(genre, ratings)| {
let genre_avg =
ratings.iter().map(|&r| f64::from(r)).sum::<f64>() / ratings.len() as f64;
let delta = round2(genre_avg - user_avg);
if delta.abs() < MIN_RATING_DELTA {
return None;
}
Some(Insight::RatingLift {
genre: genre.to_string(),
genre_avg: round2(genre_avg),
user_avg: round2(user_avg),
delta,
sample_count: ratings.len() as u32,
})
})
.collect();
candidates.sort_by(strongest_first);
candidates.truncate(MAX_PER_CATEGORY);
candidates
}
/// Strongest first; ties broken by label ascending so output is deterministic.
fn strongest_first(a: &Insight, b: &Insight) -> std::cmp::Ordering {
strength_of(b)
.partial_cmp(&strength_of(a))
.unwrap_or(std::cmp::Ordering::Equal)
.then_with(|| label_of(a).cmp(label_of(b)))
}
/// The earliest review of each movie; later reviews of the same movie are
/// rewatches (the same definition `ReviewHistory` uses).
fn first_time_rows(rows: &[WatchRow]) -> Vec<&WatchRow> {
let mut seen: HashMap<uuid::Uuid, &WatchRow> = HashMap::new();
for row in rows {
match seen.get(&row.movie_id) {
Some(existing) if existing.watched_at <= row.watched_at => {}
_ => {
seen.insert(row.movie_id, row);
}
}
}
let mut first: Vec<&WatchRow> = seen.into_values().collect();
first.sort_by_key(|r| r.watched_at);
first
}
fn avg_rating(rows: &[&WatchRow]) -> f64 {
rows.iter().map(|r| f64::from(r.rating)).sum::<f64>() / rows.len() as f64
}
// ── Seasonality ──────────────────────────────────────────────────────────────
fn seasonalities<Tz: TimeZone>(rows: &[WatchRow], tz: &Tz) -> Vec<Insight> {
if rows.is_empty() {
return vec![];
}
// The user's own month-of-year distribution is the baseline: a diary
// started in October, or a user who simply watches more in winter, must
// not make every big genre look seasonal.
let mut user_months = [0u32; 12];
let mut by_genre: HashMap<&str, [u32; 12]> = HashMap::new();
for row in rows {
let m = local(tz, &row.watched_at).month0() as usize;
user_months[m] += 1;
for genre in &row.genres {
by_genre.entry(genre).or_insert([0; 12])[m] += 1;
}
}
let user_total = rows.len() as f64;
let mut candidates: Vec<Insight> = by_genre
.into_iter()
.filter_map(|(genre, histogram)| {
let genre_total: u32 = histogram.iter().sum();
if (genre_total as usize) < MIN_GENRE_SEASON_ROWS {
return None;
}
// Month with the largest lift over the user's share for that
// month; smallest month number wins ties — deterministic output.
let (peak_month, multiplier) = histogram
.iter()
.enumerate()
.filter(|(_, c)| **c >= MIN_SEASON_PEAK_ROWS)
.map(|(i, &c)| {
let genre_share = f64::from(c) / f64::from(genre_total);
let user_share = f64::from(user_months[i]) / user_total;
(i as u32 + 1, round2(genre_share / user_share))
})
.fold(None, |best: Option<(u32, f64)>, cur| match best {
Some(b) if b.1 >= cur.1 => Some(b),
_ => Some(cur),
})?;
if multiplier < MIN_SEASON_MULTIPLIER {
return None;
}
Some(Insight::Seasonality {
genre: genre.to_string(),
month: peak_month,
multiplier,
sample_count: genre_total,
})
})
.collect();
candidates.sort_by(strongest_first);
candidates.truncate(MAX_PER_CATEGORY);
candidates
}
// ── Medium timing ────────────────────────────────────────────────────────────
fn medium_timings<Tz: TimeZone>(rows: &[WatchRow], tz: &Tz) -> Vec<Insight> {
if rows.is_empty() {
return vec![];
}
let with_medium_count = rows.iter().filter(|r| r.watch_medium.is_some()).count();
let missing_share = 1.0 - with_medium_count as f64 / rows.len() as f64;
if missing_share > MAX_MISSING_MEDIUM_SHARE {
return vec![];
}
// (total watches, weekend watches) per medium.
let mut by_medium: HashMap<&str, (usize, usize)> = HashMap::new();
let mut weekend_total = 0usize;
for row in rows.iter().filter(|r| r.watch_medium.is_some()) {
let weekend = matches!(
local(tz, &row.watched_at).weekday(),
Weekday::Sat | Weekday::Sun
);
weekend_total += usize::from(weekend);
let entry = by_medium
.entry(row.watch_medium.as_deref().expect("filtered above"))
.or_insert((0, 0));
entry.0 += 1;
entry.1 += usize::from(weekend);
}
let baseline_share = weekend_total as f64 / with_medium_count as f64;
// No weekend watches at all → no baseline to lift against.
if baseline_share == 0.0 {
return vec![];
}
let mut candidates: Vec<Insight> = by_medium
.into_iter()
.filter(|(_, (count, _))| *count >= MIN_MEDIUM_ROWS)
.filter_map(|(medium, (count, weekend))| {
let weekend_share = weekend as f64 / count as f64;
let multiplier = round2(weekend_share / baseline_share);
if multiplier < MIN_MEDIUM_MULTIPLIER {
return None;
}
Some(Insight::MediumTiming {
medium: medium.to_string(),
multiplier,
weekend_share: round2(weekend_share),
baseline_share: round2(baseline_share),
sample_count: count as u32,
})
})
.collect();
candidates.sort_by(strongest_first);
candidates.truncate(MAX_PER_CATEGORY);
candidates
}
// ── Rhythm ───────────────────────────────────────────────────────────────────
fn rhythm<Tz: TimeZone>(rows: &[WatchRow], tz: &Tz) -> Vec<Insight> {
// Import noise is detected on the stored value: date-only imports land at
// exactly 00:00 as stored, which no longer reads as midnight once shifted
// into the viewer's zone.
let real: Vec<NaiveDateTime> = rows
.iter()
.filter(|r| !(r.watched_at.hour() == 0 && r.watched_at.minute() == 0))
.map(|r| local(tz, &r.watched_at))
.collect();
let midnight_share = 1.0 - real.len() as f64 / rows.len().max(1) as f64;
if rows.is_empty() || real.len() < MIN_RHYTHM_ROWS || midnight_share > MAX_MIDNIGHT_SHARE {
return vec![];
}
let mut hours = [0u32; 24];
let mut days = [0u32; 7]; // index 0 = Monday (chrono weekday number − 1)
for at in &real {
hours[at.hour() as usize] += 1;
days[at.weekday().number_from_monday() as usize - 1] += 1;
}
// Circular 3-hour window with the most watches; smallest start hour wins ties.
let mut best_start = 0usize;
let mut best_count = 0u32;
for start in 0..24 {
let count: u32 = (0..RHYTHM_WINDOW_HOURS)
.map(|o| hours[(start + o as usize) % 24])
.sum();
if count > best_count {
best_count = count;
best_start = start;
}
}
// Days carrying at least 60% of the busiest day's load, strongest first,
// capped at three — "mostly Tue–Thu" without the tail of noise.
let max_day = days.iter().copied().max().unwrap_or(0);
let mut top_days: Vec<(u32, u32)> = days
.into_iter()
.enumerate()
.filter(|(_, c)| *c > 0 && f64::from(*c) >= f64::from(max_day) * 0.6)
.map(|(i, c)| (i as u32 + 1, c))
.collect();
// Count descending, then day number ascending — deterministic.
top_days.sort_unstable_by_key(|&(day, count)| (std::cmp::Reverse(count), day));
let top_days: Vec<u32> = top_days.into_iter().map(|(day, _)| day).take(3).collect();
vec![Insight::Rhythm {
peak_hour_start: best_start as u32,
peak_hour_end: (best_start as u32 + RHYTHM_WINDOW_HOURS) % 24,
top_days,
sample_count: real.len() as u32,
}]
}
// ── Shared helpers ───────────────────────────────────────────────────────────
/// Strength = effect size scaled by evidence volume. A 1.6× lift on 60 rows
/// outranks 1.7× on 3; a raw lift would not.
fn strength(effect: f64, sample: usize) -> f64 {
effect * (1.0 + (sample as f64).ln())
}
fn strength_of(insight: &Insight) -> f64 {
match insight {
Insight::RatingLift {
delta,
sample_count,
..
} => strength(delta.abs(), *sample_count as usize),
Insight::Seasonality {
multiplier,
sample_count,
..
}
| Insight::MediumTiming {
multiplier,
sample_count,
..
} => strength(multiplier - 1.0, *sample_count as usize),
Insight::Rhythm { .. } => 0.0,
}
}
fn label_of(insight: &Insight) -> &str {
match insight {
Insight::RatingLift { genre, .. } | Insight::Seasonality { genre, .. } => genre,
Insight::MediumTiming { medium, .. } => medium,
Insight::Rhythm { .. } => "",
}
}
fn round2(v: f64) -> f64 {
(v * 100.0).round() / 100.0
}
#[cfg(test)]
#[path = "tests/insights_analyzer.rs"]
mod tests;

1
crates/domain/src/services/mod.rs
View File

@@ -1,2 +1,3 @@
pub mod insights_analyzer;
pub mod review_history;
pub mod wrapup_analyzer;

435
crates/domain/src/services/tests/insights_analyzer.rs Normal file
View File

@@ -0,0 +1,435 @@
use chrono::{Datelike, FixedOffset, NaiveDate, Utc, Weekday};
use uuid::Uuid;
use crate::models::{Insight, WatchRow};
use super::super::insights_analyzer::build_insights;
fn watch(genres: &[&str], rating: u8, date: NaiveDate, hour: u32) -> WatchRow {
WatchRow {
movie_id: Uuid::new_v4(),
rating,
watched_at: date.and_hms_opt(hour, 30, 0).unwrap(),
watch_medium: Some("streaming".to_string()),
genres: genres.iter().map(|g| g.to_string()).collect(),
}
}
fn date(year: i32, month: u32, day: u32) -> NaiveDate {
NaiveDate::from_ymd_opt(year, month, day).unwrap()
}
fn kinds(insights: &[Insight]) -> Vec<&str> {
insights
.iter()
.map(|i| match i {
Insight::RatingLift { .. } => "rating_lift",
Insight::Seasonality { .. } => "seasonality",
Insight::MediumTiming { .. } => "medium_timing",
Insight::Rhythm { .. } => "rhythm",
})
.collect()
}
// ── Rating lift ──────────────────────────────────────────────────────────────
#[test]
fn rating_lift_compares_genre_to_personal_average() {
// 12 thrillers rated 5, 12 comedies rated 2 — user average 3.5.
let mut rows: Vec<WatchRow> = (0..12)
.map(|i| watch(&["Thriller"], 5, date(2024, 1, 1 + i), 20))
.collect();
rows.extend((0..12).map(|i| watch(&["Comedy"], 2, date(2024, 3, 1 + i), 20)));
let insights = build_insights(&rows, &Utc);
let thrill = insights.iter().find_map(|i| match i {
Insight::RatingLift {
genre,
delta,
sample_count,
..
} if genre == "Thriller" => Some((*delta, *sample_count)),
_ => None,
});
let (delta, count) = thrill.expect("Thriller lift expected");
assert!((delta - 1.5).abs() < 1e-9, "delta was {delta}");
assert_eq!(count, 12);
let comedy = insights.iter().find_map(|i| match i {
Insight::RatingLift { genre, delta, .. } if genre == "Comedy" => Some(*delta),
_ => None,
});
assert!((comedy.expect("Comedy lift expected") + 1.5).abs() < 1e-9);
}
#[test]
fn rating_lift_ignores_rewatches() {
// 10 first-time dramas rated 3 (baseline), 5 thrillers rated 5…
let mut rows: Vec<WatchRow> = (0..10)
.map(|i| watch(&["Drama"], 3, date(2024, 1, 1 + i), 20))
.collect();
let mut thriller_ids: Vec<uuid::Uuid> = Vec::new();
for i in 0..5 {
let row = watch(&["Thriller"], 5, date(2024, 2, 1 + i), 20);
thriller_ids.push(row.movie_id);
rows.push(row);
}
// …and 3 rewatches of those same thrillers rated 1, which must not count.
for (i, id) in thriller_ids.iter().enumerate() {
let mut rewatch = watch(&["Thriller"], 1, date(2024, 6, 1 + i as u32), 20);
rewatch.movie_id = *id;
rows.push(rewatch);
}
let insights = build_insights(&rows, &Utc);
let thrill = insights.iter().find_map(|i| match i {
Insight::RatingLift {
genre,
delta,
sample_count,
..
} if genre == "Thriller" => Some((*delta, *sample_count)),
_ => None,
});
let (delta, count) = thrill.expect("Thriller lift expected");
// 15 first-time rows: user_avg = (10*3 + 5*5)/15 = 55/15 ≈ 3.67,
// genre_avg = 5.0, delta ≈ 1.33.
assert!((delta - 1.33).abs() < 0.01, "delta was {delta}");
assert_eq!(count, 5, "rewatches must not join the genre sample");
}
#[test]
fn rating_lift_below_sample_threshold_is_absent() {
let mut rows: Vec<WatchRow> = (0..10)
.map(|i| watch(&["Drama"], 3, date(2024, 1, 1 + i), 20))
.collect();
// Only 4 thriller ratings — below MIN_GENRE_RATINGS, so no matter how
// high they are rated, the genre must not yield an insight.
rows.extend((0..4).map(|i| watch(&["Thriller"], 5, date(2024, 2, 1 + i), 20)));
let insights = build_insights(&rows, &Utc);
assert!(
!insights
.iter()
.any(|i| matches!(i, Insight::RatingLift { genre, .. } if genre == "Thriller")),
"a genre with 4 ratings must not yield a lift"
);
}
#[test]
fn rating_lift_below_delta_threshold_is_absent() {
// 5 thrillers rated 4; dramas mix to put the personal average at 3.85,
// so the thriller delta is 0.15 < 0.3.
let mut rows: Vec<WatchRow> = (0..12)
.map(|i| watch(&["Drama"], 4, date(2024, 1, 1 + i % 28), 20))
.collect();
rows.extend((0..3).map(|i| watch(&["Drama"], 3, date(2024, 1, 13 + i), 20)));
rows.extend((0..5).map(|i| watch(&["Thriller"], 4, date(2024, 2, 1 + i), 20)));
let insights = build_insights(&rows, &Utc);
assert!(!kinds(&insights).contains(&"rating_lift"));
}
// ── Seasonality ──────────────────────────────────────────────────────────────
#[test]
fn seasonality_detects_month_peak_against_own_monthly_baseline() {
// 16 horrors: 10 in October across two years, 6 scattered.
let mut rows: Vec<WatchRow> = Vec::new();
for i in 0..10u32 {
let year = if i < 5 { 2023 } else { 2024 };
rows.push(watch(&["Horror"], 4, date(year, 10, 1 + i % 27), 21));
}
for (m, d) in [(1u32, 3u32), (2, 7), (4, 11), (6, 15), (8, 19), (12, 21)] {
rows.push(watch(&["Horror"], 4, date(2024, m, d), 21));
}
// A perfectly spread genre (2 watches per month) must NOT appear.
rows.extend((0..24).map(|i| watch(&["Drama"], 3, date(2024, 1 + i % 12, 5), 21)));
let insights = build_insights(&rows, &Utc);
let horror = insights.iter().find_map(|i| match i {
Insight::Seasonality {
genre,
month,
multiplier,
sample_count,
..
} if genre == "Horror" => Some((*month, *multiplier, *sample_count)),
_ => None,
});
let (month, multiplier, count) = horror.expect("Horror seasonality expected");
assert_eq!(month, 10);
// Horror: 10 of 16 watches in October (62.5%); the whole diary puts
// 12 of 40 watches there (30%) → 0.625 / 0.3 ≈ 2.08× the user's own share.
assert!(
(multiplier - 2.08).abs() < 1e-9,
"multiplier was {multiplier}"
);
assert_eq!(count, 16);
assert!(
!insights
.iter()
.any(|i| matches!(i, Insight::Seasonality { genre, .. } if genre == "Drama")),
"a spread-out genre must not yield a seasonality insight"
);
}
#[test]
fn seasonality_ignores_when_the_diary_itself_is_concentrated() {
// A diary started in October: every genre peaks in Oct/Nov only because
// the user logged nothing else. Relative to their own months, nothing is
// seasonal.
let mut rows: Vec<WatchRow> = Vec::new();
for i in 0..20u32 {
let month = if i % 2 == 0 { 10 } else { 11 };
rows.push(watch(&["Drama"], 3, date(2024, month, 1 + i), 21));
rows.push(watch(&["Comedy"], 3, date(2024, month, 1 + i), 21));
}
assert!(!kinds(&build_insights(&rows, &Utc)).contains(&"seasonality"));
}
#[test]
fn seasonality_below_sample_threshold_is_absent() {
let rows: Vec<WatchRow> = (0..14)
.map(|i| watch(&["Horror"], 4, date(2024, 10, 1 + i), 21))
.collect();
assert!(!kinds(&build_insights(&rows, &Utc)).contains(&"seasonality"));
}
// ── Medium timing ────────────────────────────────────────────────────────────
#[test]
fn medium_timing_detects_weekend_lift() {
let saturday = date(2024, 3, 9); // a Saturday
assert_eq!(saturday.weekday(), Weekday::Sat);
let friday = date(2024, 3, 8);
let mut rows: Vec<WatchRow> = Vec::new();
// 6 cinema watches, 5 of them on Saturdays.
let cinema_days = [
saturday,
saturday.with_day(16).unwrap(),
saturday.with_day(23).unwrap(),
saturday.with_day(30).unwrap(),
date(2024, 4, 6), // next month's first Saturday
friday,
];
for day in cinema_days {
let mut row = watch(&["Drama"], 4, day, 19);
row.watch_medium = Some("cinema".to_string());
rows.push(row);
}
// 10 streaming watches on weekdays only (two Mon–Fri weeks) to keep the
// baseline weekend share driven by cinema alone.
for i in 0..10u32 {
let mut row = watch(&["Drama"], 4, friday, 21);
row.watch_medium = Some("streaming".to_string());
row.watched_at = date(2024, 3, 4 + i % 5 + 7 * (i / 5))
.and_hms_opt(21, 0, 0)
.unwrap();
rows.push(row);
}
let insights = build_insights(&rows, &Utc);
let cinema = insights.iter().find_map(|i| match i {
Insight::MediumTiming {
medium,
multiplier,
sample_count,
..
} if medium == "cinema" => Some((*multiplier, *sample_count)),
_ => None,
});
let (multiplier, count) = cinema.expect("cinema timing expected");
// Baseline weekend share = 5/16; cinema share = 5/6 → lift = (5/6)/(5/16) ≈ 2.67.
assert!(
(multiplier - 2.67).abs() < 0.01,
"multiplier was {multiplier}"
);
assert_eq!(count, 6);
}
#[test]
fn medium_timing_skipped_when_medium_mostly_missing() {
let mut rows: Vec<WatchRow> = Vec::new();
let first_saturday = date(2024, 3, 9);
for i in 0..6u32 {
let mut row = watch(
&["Drama"],
4,
first_saturday + chrono::Duration::weeks(i64::from(i)),
20,
);
row.watch_medium = Some("cinema".to_string());
rows.push(row);
}
for i in 0..10u32 {
let mut row = watch(&["Drama"], 4, date(2024, 3, 1 + i), 20);
row.watch_medium = None; // imports without a medium
rows.push(row);
}
assert!(!kinds(&build_insights(&rows, &Utc)).contains(&"medium_timing"));
}
// ── Rhythm ───────────────────────────────────────────────────────────────────
#[test]
fn rhythm_finds_peak_window_and_days() {
// 12 watches at 22:30: five Tuesdays, four Wednesdays, three Thursdays.
let first_tuesday = date(2024, 3, 5);
assert_eq!(first_tuesday.weekday(), Weekday::Tue);
let mut rows: Vec<WatchRow> = Vec::new();
for day in 0..3u32 {
let count = 5 - day; // Tue ×5, Wed ×4, Thu ×3
for week in 0..count {
rows.push(watch(
&["Drama"],
4,
first_tuesday
+ chrono::Duration::days(i64::from(day))
+ chrono::Duration::weeks(i64::from(week)),
22,
));
}
}
let insights = build_insights(&rows, &Utc);
let rhythm = insights.iter().find_map(|i| match i {
Insight::Rhythm {
peak_hour_start,
peak_hour_end,
top_days,
sample_count,
} => Some((
*peak_hour_start,
*peak_hour_end,
top_days.clone(),
*sample_count,
)),
_ => None,
});
let (start, end, days, count) = rhythm.expect("rhythm expected");
// All watches sit at hour 22; the earliest 3-hour window that covers
// them starts at 20 (20, 21, 22).
assert_eq!(start, 20);
assert_eq!(end, 23);
// Every day carries ≥60% of the busiest day (5): Tue(5) Wed(4) Thu(3).
assert_eq!(days, vec![2, 3, 4], "Tue=2, Wed=3, Thu=4");
assert_eq!(count, 12);
}
#[test]
fn rhythm_buckets_in_the_viewers_time_zone() {
// 12 watches stored at 22:30 UTC on Tue/Wed/Thu are 00:30 the next day
// for a viewer at UTC+2.
let first_tuesday = date(2024, 3, 5);
let mut rows: Vec<WatchRow> = Vec::new();
for day in 0..3u32 {
for week in 0..(5 - day) {
rows.push(watch(
&["Drama"],
4,
first_tuesday
+ chrono::Duration::days(i64::from(day))
+ chrono::Duration::weeks(i64::from(week)),
22,
));
}
}
let plus_two = FixedOffset::east_opt(2 * 3600).unwrap();
let insights = build_insights(&rows, &plus_two);
let (start, end, days) = insights
.iter()
.find_map(|i| match i {
Insight::Rhythm {
peak_hour_start,
peak_hour_end,
top_days,
..
} => Some((*peak_hour_start, *peak_hour_end, top_days.clone())),
_ => None,
})
.expect("rhythm expected");
// Local hour 0; the scan starts at 0, so the first window covering it is 0–3.
assert_eq!((start, end), (0, 3));
assert_eq!(days, vec![3, 4, 5], "Wed=3, Thu=4, Fri=5 locally");
}
#[test]
fn rhythm_import_noise_detected_before_zone_shift() {
// Date-only imports are stored at 00:00 UTC; at UTC+2 they read 02:00,
// but they are still noise.
let rows: Vec<WatchRow> = (0..12u32)
.map(|i| WatchRow {
movie_id: Uuid::new_v4(),
rating: 4,
watched_at: date(2024, 1, 1 + i).and_hms_opt(0, 0, 0).unwrap(),
watch_medium: None,
genres: vec!["Drama".to_string()],
})
.collect();
let plus_two = FixedOffset::east_opt(2 * 3600).unwrap();
assert!(!kinds(&build_insights(&rows, &plus_two)).contains(&"rhythm"));
}
#[test]
fn rhythm_skipped_when_timestamps_are_import_noise() {
let mut rows: Vec<WatchRow> = Vec::new();
for i in 0..12u32 {
// Exactly midnight — the signature of a CSV import with no time data.
let row = WatchRow {
movie_id: Uuid::new_v4(),
rating: 4,
watched_at: date(2024, 1, 1 + i).and_hms_opt(0, 0, 0).unwrap(),
watch_medium: None,
genres: vec!["Drama".to_string()],
};
rows.push(row);
}
assert!(!kinds(&build_insights(&rows, &Utc)).contains(&"rhythm"));
}
// ── Overall ──────────────────────────────────────────────────────────────────
#[test]
fn empty_diary_yields_nothing() {
assert!(build_insights(&[], &Utc).is_empty());
}
#[test]
fn small_diary_yields_nothing() {
let rows: Vec<WatchRow> = (0..4)
.map(|i| watch(&["Drama"], 4, date(2024, 1, 1 + i), 20))
.collect();
assert!(build_insights(&rows, &Utc).is_empty());
}
#[test]
fn max_three_insights_per_category() {
// Six genres, all with a strong rating lift — only the three strongest survive.
let mut rows: Vec<WatchRow> = Vec::new();
let genres = [
("Thriller", 5u8),
("Horror", 5),
("Sci-Fi", 5),
("Comedy", 5),
("Romance", 5),
("Documentary", 5),
];
for (genre, rating) in genres {
rows.extend((0..5).map(|i| watch(&[genre], rating, date(2024, 1, 1 + i), 20)));
}
// Baseline dramas rated 1 make every lift large.
rows.extend((0..10).map(|i| watch(&["Drama"], 1, date(2024, 3, 1 + i), 20)));
let insights = build_insights(&rows, &Utc);
let rating_lifts = insights
.iter()
.filter(|i| matches!(i, Insight::RatingLift { .. }))
.count();
assert_eq!(rating_lifts, 3);
}

3
crates/domain/src/testing/fakes.rs
View File

@@ -103,9 +103,6 @@ impl FakeDiaryQuery {
.insert(movie.id().value(), (movie, reviews));
}
/// Configures what `query_diary` returns. Without this, `query_diary` keeps
/// its original always-empty behavior — this is purely opt-in for tests that
/// need to prove entries actually flow through a use case.
pub fn set_diary_page(&self, page: Paginated<DiaryEntry>) {
*self.diary_page.lock().unwrap() = Some(page);
}

Some files were not shown because too many files have changed in this diff Show More