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

refactor: extract business logic from handlers to application layer
Some checks failed
CI / Check / Test (push) Has been cancelled
Details
CI / Release build (push) Has been cancelled
Details

Browse Source 
Move domain logic out of 7 handlers into use cases:
- activity feed: FollowingFilter construction
- user profile: social counts + pending followers
- users list: parallel local+remote actor loading
- watchlist page: local-vs-remote branching
- sync_poster: movie lookup + validation
- get_profile: avatar URL construction
- post_register: register+login orchestration

Add SocialQueryPort.{count_following,count_accepted_followers,
get_pending_followers} to AppContext behind federation feature gate.
This commit is contained in:
Gabriel Kaszewski
2026-05-29 11:41:16 +02:00
parent 2355f89bed
commit c3b89f6dc6
22 changed files with 644 additions and 350 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
crates/adapters/postgres-federation/src/lib.rs
View File

@@ -748,6 +748,64 @@ impl domain::ports::SocialQueryPort for PostgresFederationRepository {
)
.collect())
}
async fn count_following(
&self,
user_id: uuid::Uuid,
) -> Result<usize, domain::errors::DomainError> {
let uid = user_id.to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_following WHERE local_user_id = $1 AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn count_accepted_followers(
&self,
user_id: uuid::Uuid,
) -> Result<usize, domain::errors::DomainError> {
let uid = user_id.to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_followers WHERE local_user_id = $1 AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn get_pending_followers(
&self,
user_id: uuid::Uuid,
) -> Result<Vec<domain::ports::PendingFollowerInfo>, domain::errors::DomainError> {
let uid = user_id.to_string();
let rows = sqlx::query_as::<_, (String, String, Option<String>, Option<String>)>(
"SELECT ar.url, ar.handle, ar.display_name, ar.avatar_url
FROM ap_followers f
JOIN ap_remote_actors ar ON ar.url = f.remote_actor_url
WHERE f.local_user_id = $1 AND f.status = 'pending'",
)
.bind(&uid)
.fetch_all(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(
|(url, handle, display_name, avatar_url)| domain::ports::PendingFollowerInfo {
url,
handle,
display_name,
avatar_url,
},
)
.collect())
}
}
#[async_trait]

58
crates/adapters/sqlite-federation/src/lib.rs
View File

@@ -924,6 +924,64 @@ impl domain::ports::SocialQueryPort for SqliteFederationRepository {
)
.collect())
}
async fn count_following(
&self,
user_id: uuid::Uuid,
) -> Result<usize, domain::errors::DomainError> {
let uid = user_id.to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_following WHERE local_user_id = ? AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn count_accepted_followers(
&self,
user_id: uuid::Uuid,
) -> Result<usize, domain::errors::DomainError> {
let uid = user_id.to_string();
let count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM ap_followers WHERE local_user_id = ? AND status = 'accepted'",
)
.bind(&uid)
.fetch_one(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(count as usize)
}
async fn get_pending_followers(
&self,
user_id: uuid::Uuid,
) -> Result<Vec<domain::ports::PendingFollowerInfo>, domain::errors::DomainError> {
let uid = user_id.to_string();
let rows = sqlx::query_as::<_, (String, String, Option<String>, Option<String>)>(
"SELECT ar.url, ar.handle, ar.display_name, ar.avatar_url
FROM ap_followers f
JOIN ap_remote_actors ar ON ar.url = f.remote_actor_url
WHERE f.local_user_id = ? AND f.status = 'pending'",
)
.bind(&uid)
.fetch_all(&self.pool)
.await
.map_err(|e| domain::errors::DomainError::InfrastructureError(e.to_string()))?;
Ok(rows
.into_iter()
.map(
|(url, handle, display_name, avatar_url)| domain::ports::PendingFollowerInfo {
url,
handle,
display_name,
avatar_url,
},
)
.collect())
}
}
#[async_trait]

7
crates/application/src/commands.rs
View File

@@ -21,7 +21,6 @@ pub struct LogReviewCommand {
#[derive(Clone)]
pub struct SyncPosterCommand {
pub movie_id: Uuid,
pub external_metadata_id: String,
}
pub struct RegisterCommand {
@@ -103,3 +102,9 @@ pub struct RemoveFromWatchlistCommand {
pub user_id: Uuid,
pub movie_id: Uuid,
}
pub struct RegisterAndLoginCommand {
pub email: String,
pub username: String,
pub password: String,
}

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

@@ -6,8 +6,8 @@ use domain::ports::{
AuthService, DiaryExporter, DiaryRepository, DocumentParser, EventPublisher, ImageStorage,
ImportProfileRepository, ImportSessionRepository, MetadataClient, MovieProfileRepository,
MovieRepository, PasswordHasher, PersonCommand, PersonQuery, PosterFetcherClient,
ReviewRepository, SearchCommand, SearchPort, StatsRepository, UserProfileFieldsRepository,
UserRepository, WatchlistRepository,
ReviewRepository, SearchCommand, SearchPort, SocialQueryPort, StatsRepository,
UserProfileFieldsRepository, UserRepository, WatchlistRepository,
};
use crate::config::AppConfig;
@@ -38,5 +38,7 @@ pub struct AppContext {
pub profile_fields_repository: Arc<dyn UserProfileFieldsRepository>,
#[cfg(feature = "federation")]
pub remote_watchlist_repository: Arc<dyn RemoteWatchlistRepository>,
#[cfg(feature = "federation")]
pub social_query: Arc<dyn SocialQueryPort>,
pub config: AppConfig,
}

8
crates/application/src/queries.rs
View File

@@ -28,7 +28,8 @@ pub struct GetActivityFeedQuery {
pub offset: u32,
pub sort_by: domain::ports::FeedSortBy,
pub search: Option<String>,
pub following: Option<domain::ports::FollowingFilter>,
pub viewer_user_id: Option<Uuid>,
pub filter_following: bool,
}
pub struct GetUsersQuery;
@@ -73,6 +74,7 @@ pub struct GetUserProfileQuery {
pub offset: Option<u32>,
pub sort_by: domain::ports::FeedSortBy,
pub search: Option<String>,
pub is_own_profile: bool,
}
pub struct GetMovieSocialPageQuery {
@@ -99,3 +101,7 @@ pub struct IsOnWatchlistQuery {
pub user_id: Uuid,
pub movie_id: Uuid,
}
pub struct GetCurrentProfileQuery {
pub user_id: Uuid,
}

4
crates/application/src/test_helpers.rs
View File

@@ -2,6 +2,8 @@ use std::sync::Arc;
#[cfg(feature = "federation")]
use domain::testing::PanicRemoteWatchlistRepository;
#[cfg(feature = "federation")]
use domain::testing::PanicSocialQueryPort;
use domain::{
ports::{
AuthService, DiaryExporter, DiaryRepository, DocumentParser, EventPublisher, ImageStorage,
@@ -144,6 +146,8 @@ impl TestContextBuilder {
config: self.config,
#[cfg(feature = "federation")]
remote_watchlist_repository: std::sync::Arc::new(PanicRemoteWatchlistRepository),
#[cfg(feature = "federation")]
social_query: std::sync::Arc::new(PanicSocialQueryPort),
}
}
}

48
crates/application/src/use_cases/get_activity_feed.rs
View File

@@ -5,6 +5,7 @@ use domain::{
FeedEntry,
collections::{PageParams, Paginated},
},
ports::FollowingFilter,
};
pub async fn execute(
@@ -12,12 +13,57 @@ pub async fn execute(
query: GetActivityFeedQuery,
) -> Result<Paginated<FeedEntry>, DomainError> {
let page = PageParams::new(Some(query.limit), Some(query.offset))?;
let following = build_following_filter(ctx, &query).await;
ctx.diary_repository
.query_activity_feed_filtered(
&page,
&query.sort_by,
query.search.as_deref(),
query.following.as_ref(),
following.as_ref(),
)
.await
}
async fn build_following_filter(
_ctx: &AppContext,
query: &GetActivityFeedQuery,
) -> Option<FollowingFilter> {
#[cfg(not(feature = "federation"))]
{
let _ = query;
return None;
}
#[cfg(feature = "federation")]
{
if !query.filter_following {
return None;
}
let viewer_id = match query.viewer_user_id {
Some(id) => id,
None => return None,
};
let urls = _ctx
.social_query
.get_accepted_following_urls(viewer_id)
.await
.unwrap_or_default();
let base_url = &_ctx.config.base_url;
let mut local_ids = vec![viewer_id];
let mut remote_urls = Vec::new();
for url in urls {
if let Some(suffix) = url.strip_prefix(&format!("{}/users/", base_url))
&& let Ok(parsed_id) = uuid::Uuid::parse_str(suffix)
{
local_ids.push(parsed_id);
continue;
}
remote_urls.push(url);
}
Some(FollowingFilter {
local_user_ids: local_ids,
remote_actor_urls: remote_urls,
})
}
}

31
crates/application/src/use_cases/get_current_profile.rs Normal file
View File

@@ -0,0 +1,31 @@
use domain::errors::DomainError;
use crate::{context::AppContext, queries::GetCurrentProfileQuery};
pub struct CurrentProfileData {
pub username: String,
pub bio: Option<String>,
pub avatar_url: Option<String>,
}
pub async fn execute(
ctx: &AppContext,
query: GetCurrentProfileQuery,
) -> Result<CurrentProfileData, DomainError> {
let user_id = domain::value_objects::UserId::from_uuid(query.user_id);
let user = ctx
.user_repository
.find_by_id(&user_id)
.await?
.ok_or_else(|| DomainError::NotFound("User not found".into()))?;
let avatar_url = user
.avatar_path()
.map(|path| format!("{}/images/{}", ctx.config.base_url, path));
Ok(CurrentProfileData {
username: user.username().value().to_string(),
bio: user.bio().map(|s| s.to_string()),
avatar_url,
})
}

102
crates/application/src/use_cases/get_user_profile.rs
View File

@@ -13,11 +13,21 @@ use domain::{
value_objects::UserId,
};
pub struct PendingFollowerView {
pub url: String,
pub handle: String,
pub display_name: Option<String>,
pub avatar_url: Option<String>,
}
pub struct UserProfileData {
pub stats: UserStats,
pub entries: Option<Paginated<DiaryEntry>>,
pub history: Option<Vec<MonthActivity>>,
pub trends: Option<UserTrends>,
pub following_count: usize,
pub followers_count: usize,
pub pending_followers: Vec<PendingFollowerView>,
}
pub async fn execute(
@@ -27,27 +37,30 @@ pub async fn execute(
let user_id = UserId::from_uuid(query.user_id);
let stats = ctx.stats_repository.get_user_stats(&user_id).await?;
let (following_count, followers_count, pending_followers) =
load_social_counts(ctx, query.user_id, query.is_own_profile).await;
let base = |entries, history, trends| UserProfileData {
stats,
entries,
history,
trends,
following_count,
followers_count,
pending_followers,
};
match query.view {
ProfileView::History => {
let all_entries = ctx.diary_repository.get_user_history(&user_id).await?;
let history = group_by_month(all_entries);
Ok(UserProfileData {
stats,
entries: None,
history: Some(history),
trends: None,
})
Ok(base(None, Some(history), None))
}
ProfileView::Trends => {
let trends = ctx.stats_repository.get_user_trends(&user_id).await?;
Ok(UserProfileData {
stats,
entries: None,
history: None,
trends: Some(trends),
})
Ok(base(None, None, Some(trends)))
}
ProfileView::Ratings => {
ProfileView::Ratings | ProfileView::Recent => {
let sort_direction = feed_sort_to_direction(query.sort_by);
let filter = paged_user_filter(
user_id,
@@ -57,30 +70,49 @@ pub async fn execute(
query.search.clone(),
)?;
let entries = ctx.diary_repository.query_diary(&filter).await?;
Ok(UserProfileData {
stats,
entries: Some(entries),
history: None,
trends: None,
})
Ok(base(Some(entries), None, None))
}
ProfileView::Recent => {
let sort_direction = feed_sort_to_direction(query.sort_by);
let filter = paged_user_filter(
user_id,
sort_direction,
query.limit,
query.offset,
query.search.clone(),
)?;
let entries = ctx.diary_repository.query_diary(&filter).await?;
Ok(UserProfileData {
stats,
entries: Some(entries),
history: None,
trends: None,
})
}
}
async fn load_social_counts(
_ctx: &AppContext,
_user_id: uuid::Uuid,
_is_own_profile: bool,
) -> (usize, usize, Vec<PendingFollowerView>) {
#[cfg(not(feature = "federation"))]
{
(0, 0, vec![])
}
#[cfg(feature = "federation")]
{
if !_is_own_profile {
return (0, 0, vec![]);
}
let following = _ctx
.social_query
.count_following(_user_id)
.await
.unwrap_or(0);
let followers = _ctx
.social_query
.count_accepted_followers(_user_id)
.await
.unwrap_or(0);
let pending = _ctx
.social_query
.get_pending_followers(_user_id)
.await
.unwrap_or_default()
.into_iter()
.map(|p| PendingFollowerView {
url: p.url,
handle: p.handle,
display_name: p.display_name,
avatar_url: p.avatar_url,
})
.collect();
(following, followers, pending)
}
}

25
crates/application/src/use_cases/get_users.rs
View File

@@ -1,9 +1,28 @@
use crate::{context::AppContext, queries::GetUsersQuery};
use domain::{errors::DomainError, models::UserSummary};
use domain::{errors::DomainError, models::UserSummary, ports::RemoteActorInfo};
pub struct UsersListData {
pub users: Vec<UserSummary>,
pub remote_actors: Vec<RemoteActorInfo>,
}
pub async fn execute(
ctx: &AppContext,
_query: GetUsersQuery,
) -> Result<Vec<UserSummary>, DomainError> {
ctx.user_repository.list_with_stats().await
) -> Result<UsersListData, DomainError> {
#[cfg(feature = "federation")]
let (users_result, actors_result) = tokio::join!(
ctx.user_repository.list_with_stats(),
ctx.social_query.list_all_followed_remote_actors()
);
#[cfg(not(feature = "federation"))]
let (users_result, actors_result) = (
ctx.user_repository.list_with_stats().await,
Ok::<Vec<RemoteActorInfo>, DomainError>(vec![]),
);
Ok(UsersListData {
users: users_result?,
remote_actors: actors_result?,
})
}

95
crates/application/src/use_cases/get_watchlist_page.rs Normal file
View File

@@ -0,0 +1,95 @@
use domain::{errors::DomainError, value_objects::UserId};
use crate::{context::AppContext, ports::WatchlistDisplayEntry, queries::GetWatchlistQuery};
pub struct WatchlistPageResult {
pub display_entries: Vec<WatchlistDisplayEntry>,
pub has_more: bool,
pub current_offset: u32,
pub limit: u32,
}
pub async fn execute(
ctx: &AppContext,
query: GetWatchlistQuery,
is_owner: bool,
) -> Result<WatchlistPageResult, DomainError> {
let user_id = UserId::from_uuid(query.user_id);
let is_local = ctx.user_repository.find_by_id(&user_id).await?.is_some();
if is_local {
let page = super::get_watchlist::execute(ctx, query).await?;
let has_more = page.offset + page.limit < page.total_count as u32;
let display_entries = page
.items
.iter()
.map(|w| {
let remove_url = if is_owner {
Some(format!("/watchlist/{}/remove", w.movie.id().value()))
} else {
None
};
WatchlistDisplayEntry {
poster_url: w
.movie
.poster_path()
.map(|p| format!("/images/{}", p.value())),
movie_title: w.movie.title().value().to_string(),
release_year: w.movie.release_year().value(),
movie_url: Some(format!("/movies/{}", w.movie.id().value())),
added_at: w.entry.added_at.format("%b %-d, %Y").to_string(),
remove_url,
}
})
.collect();
Ok(WatchlistPageResult {
display_entries,
has_more,
current_offset: page.offset,
limit: page.limit,
})
} else {
load_remote_watchlist(ctx, query.user_id).await
}
}
#[cfg(not(feature = "federation"))]
async fn load_remote_watchlist(
_ctx: &AppContext,
_user_id: uuid::Uuid,
) -> Result<WatchlistPageResult, DomainError> {
Ok(WatchlistPageResult {
display_entries: vec![],
has_more: false,
current_offset: 0,
limit: 0,
})
}
#[cfg(feature = "federation")]
async fn load_remote_watchlist(
ctx: &AppContext,
user_id: uuid::Uuid,
) -> Result<WatchlistPageResult, DomainError> {
let remote_entries = super::get_remote_watchlist::execute(ctx, user_id)
.await
.unwrap_or_default();
let len = remote_entries.len() as u32;
let display_entries = remote_entries
.into_iter()
.map(|e| WatchlistDisplayEntry {
poster_url: e.poster_url,
movie_title: e.movie_title,
release_year: e.release_year,
movie_url: None,
added_at: e.added_at.format("%b %-d, %Y").to_string(),
remove_url: None,
})
.collect();
Ok(WatchlistPageResult {
display_entries,
has_more: false,
current_offset: 0,
limit: len,
})
}

3
crates/application/src/use_cases/mod.rs
View File

@@ -9,6 +9,7 @@ pub mod enrich_movie;
pub mod execute_import;
pub mod export_diary;
pub mod get_activity_feed;
pub mod get_current_profile;
pub mod get_diary;
pub mod get_movie_social_page;
pub mod get_movies;
@@ -20,11 +21,13 @@ pub mod get_review_history;
pub mod get_user_profile;
pub mod get_users;
pub mod get_watchlist;
pub mod get_watchlist_page;
pub mod is_on_watchlist;
pub mod list_import_profiles;
pub mod log_review;
pub mod login;
pub mod register;
pub mod register_and_login;
pub mod remove_from_watchlist;
pub mod save_import_profile;
pub mod search;

32
crates/application/src/use_cases/register_and_login.rs Normal file
View File

@@ -0,0 +1,32 @@
use domain::errors::DomainError;
use crate::{
commands::RegisterAndLoginCommand,
context::AppContext,
use_cases::{login, register},
};
pub async fn execute(
ctx: &AppContext,
cmd: RegisterAndLoginCommand,
) -> Result<login::LoginResult, DomainError> {
register::execute(
ctx,
crate::commands::RegisterCommand {
email: cmd.email.clone(),
username: cmd.username,
password: cmd.password.clone(),
role: domain::models::UserRole::Standard,
},
)
.await?;
login::execute(
ctx,
crate::queries::LoginQuery {
email: cmd.email,
password: cmd.password,
},
)
.await
}

12
crates/application/src/use_cases/sync_poster.rs
View File

@@ -2,14 +2,13 @@ use domain::{
errors::DomainError,
events::DomainEvent,
models::IndexableDocument,
value_objects::{ExternalMetadataId, MovieId, PosterPath},
value_objects::{MovieId, PosterPath},
};
use crate::{commands::SyncPosterCommand, context::AppContext};
pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), DomainError> {
let movie_id = MovieId::from_uuid(cmd.movie_id);
let external_metadata_id = ExternalMetadataId::new(cmd.external_metadata_id)?;
let mut movie = match ctx.movie_repository.get_movie_by_id(&movie_id).await? {
Some(m) => m,
@@ -22,6 +21,15 @@ pub async fn execute(ctx: &AppContext, cmd: SyncPosterCommand) -> Result<(), Dom
}
};
let external_metadata_id = movie
.external_metadata_id()
.ok_or_else(|| {
DomainError::ValidationError(
"Movie has no external metadata ID, cannot sync poster".into(),
)
})?
.clone();
let poster_url = match ctx
.metadata_client
.get_poster_url(&external_metadata_id)

20
crates/domain/src/ports.rs
View File

@@ -58,17 +58,31 @@ pub struct RemoteActorInfo {
pub display_name: Option<String>,
}
/// New trait for social/federation read queries
#[derive(Debug, Clone)]
pub struct PendingFollowerInfo {
pub url: String,
pub handle: String,
pub display_name: Option<String>,
pub avatar_url: Option<String>,
}
#[async_trait]
pub trait SocialQueryPort: Send + Sync {
/// Returns all accepted remote_actor_urls followed by `user_id`.
async fn get_accepted_following_urls(
&self,
user_id: uuid::Uuid,
) -> Result<Vec<String>, DomainError>;
/// Returns all distinct remote actors followed by any local user on this instance.
async fn list_all_followed_remote_actors(&self) -> Result<Vec<RemoteActorInfo>, DomainError>;
async fn count_following(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
async fn count_accepted_followers(&self, user_id: uuid::Uuid) -> Result<usize, DomainError>;
async fn get_pending_followers(
&self,
user_id: uuid::Uuid,
) -> Result<Vec<PendingFollowerInfo>, DomainError>;
}
#[async_trait]

52
crates/domain/src/testing.rs
View File

@@ -787,3 +787,55 @@ impl UserProfileFieldsRepository for PanicProfileFieldsRepo {
panic!("PanicProfileFieldsRepo called")
}
}
pub struct PanicSocialQueryPort;
#[async_trait]
impl crate::ports::SocialQueryPort for PanicSocialQueryPort {
async fn get_accepted_following_urls(&self, _: uuid::Uuid) -> Result<Vec<String>, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn list_all_followed_remote_actors(
&self,
) -> Result<Vec<crate::ports::RemoteActorInfo>, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn count_following(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn count_accepted_followers(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!("PanicSocialQueryPort called")
}
async fn get_pending_followers(
&self,
_: uuid::Uuid,
) -> Result<Vec<crate::ports::PendingFollowerInfo>, DomainError> {
panic!("PanicSocialQueryPort called")
}
}
pub struct NoopSocialQueryPort;
#[async_trait]
impl crate::ports::SocialQueryPort for NoopSocialQueryPort {
async fn get_accepted_following_urls(&self, _: uuid::Uuid) -> Result<Vec<String>, DomainError> {
Ok(vec![])
}
async fn list_all_followed_remote_actors(
&self,
) -> Result<Vec<crate::ports::RemoteActorInfo>, DomainError> {
Ok(vec![])
}
async fn count_following(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
Ok(0)
}
async fn count_accepted_followers(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
Ok(0)
}
async fn get_pending_followers(
&self,
_: uuid::Uuid,
) -> Result<Vec<crate::ports::PendingFollowerInfo>, DomainError> {
Ok(vec![])
}
}

92
crates/presentation/src/handlers/api.rs
View File

@@ -33,7 +33,7 @@ use domain::{
DiaryEntry, ExportFormat, Movie, MovieSummary, PersonId, Review, collections::PageParams,
},
services::review_history::Trend,
value_objects::{MovieId, UserId},
value_objects::UserId,
};
use crate::{
@@ -178,32 +178,7 @@ pub async fn sync_poster(
_user: AuthenticatedUser,
Path(movie_id): Path<Uuid>,
) -> Result<impl IntoResponse, ApiError> {
let movie = state
.app_ctx
.movie_repository
.get_movie_by_id(&MovieId::from_uuid(movie_id))
.await?
.ok_or_else(|| ApiError(DomainError::NotFound(format!("Movie {movie_id}"))))?;
let external_id = movie
.external_metadata_id()
.ok_or_else(|| {
ApiError(DomainError::ValidationError(
"Movie has no external metadata ID, cannot sync poster".into(),
))
})?
.value()
.to_string();
sync_poster::execute(
&state.app_ctx,
SyncPosterCommand {
movie_id,
external_metadata_id: external_id,
},
)
.await?;
sync_poster::execute(&state.app_ctx, SyncPosterCommand { movie_id }).await?;
Ok(StatusCode::NO_CONTENT)
}
@@ -438,26 +413,26 @@ pub async fn get_profile(
State(state): State<AppState>,
AuthenticatedUser(user_id): AuthenticatedUser,
) -> impl IntoResponse {
let user = match state.app_ctx.user_repository.find_by_id(&user_id).await {
Ok(Some(u)) => u,
Ok(None) => return StatusCode::NOT_FOUND.into_response(),
match application::use_cases::get_current_profile::execute(
&state.app_ctx,
application::queries::GetCurrentProfileQuery {
user_id: user_id.value(),
},
)
.await
{
Ok(profile) => Json(ProfileResponse {
username: profile.username,
bio: profile.bio,
avatar_url: profile.avatar_url,
})
.into_response(),
Err(DomainError::NotFound(_)) => StatusCode::NOT_FOUND.into_response(),
Err(e) => {
tracing::error!("get_profile user lookup: {:?}", e);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
tracing::error!("get_profile error: {:?}", e);
StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
};
let base_url = &state.app_ctx.config.base_url;
let avatar_url = user
.avatar_path()
.map(|path| format!("{}/images/{}", base_url, path));
Json(ProfileResponse {
username: user.username().value().to_string(),
bio: user.bio().map(|s| s.to_string()),
avatar_url,
})
.into_response()
}
}
#[utoipa::path(
@@ -1032,7 +1007,8 @@ pub async fn get_activity_feed(
offset: params.offset.unwrap_or(0),
sort_by: domain::ports::FeedSortBy::Date,
search: None,
following: None,
viewer_user_id: None,
filter_following: false,
},
)
.await?;
@@ -1058,9 +1034,10 @@ pub async fn get_activity_feed(
responses((status = 200, body = UsersResponse)),
)]
pub async fn list_users(State(state): State<AppState>) -> Result<Json<UsersResponse>, ApiError> {
let users = get_users::execute(&state.app_ctx, GetUsersQuery).await?;
let result = get_users::execute(&state.app_ctx, GetUsersQuery).await?;
Ok(Json(UsersResponse {
users: users
users: result
.users
.iter()
.map(|u| UserSummaryDto {
id: u.user_id.value(),
@@ -1117,6 +1094,7 @@ pub async fn get_user_profile(
offset: params.offset,
sort_by: domain::ports::FeedSortBy::Date,
search: None,
is_own_profile: false,
},
)
.await
@@ -1128,20 +1106,6 @@ pub async fn get_user_profile(
}
};
#[cfg(feature = "federation")]
let following_count = state.ap_service.count_following(user_id).await.unwrap_or(0);
#[cfg(not(feature = "federation"))]
let following_count = 0usize;
#[cfg(feature = "federation")]
let followers_count = state
.ap_service
.count_accepted_followers(user_id)
.await
.unwrap_or(0);
#[cfg(not(feature = "federation"))]
let followers_count = 0usize;
let entries = profile.entries.map(|p| DiaryResponse {
items: p.items.iter().map(entry_to_dto).collect(),
total_count: p.total_count,
@@ -1192,8 +1156,8 @@ pub async fn get_user_profile(
favorite_director: profile.stats.favorite_director,
most_active_month: profile.stats.most_active_month,
},
following_count,
followers_count,
following_count: profile.following_count,
followers_count: profile.followers_count,
entries,
history,
trends,

307
crates/presentation/src/handlers/html.rs
View File

@@ -9,33 +9,26 @@ use axum::{
use chrono::Utc;
use uuid::Uuid;
#[cfg(feature = "federation")]
use application::ports::{
BlockedActorEntry, BlockedActorsPageData, BlockedDomainEntry, BlockedDomainsPageData,
FollowersPageData, FollowingPageData,
};
use application::{
commands::{
AddToWatchlistCommand, DeleteReviewCommand, MovieInput, RegisterCommand,
RemoveFromWatchlistCommand,
AddToWatchlistCommand, DeleteReviewCommand, MovieInput, RemoveFromWatchlistCommand,
},
ports::{
HtmlPageContext, LoginPageData, MovieDetailPageData, NewReviewPageData,
ProfileSettingsPageData, RegisterPageData, RemoteActorView, WatchlistDisplayEntry,
WatchlistPageData,
},
queries::{
ExportQuery, GetMovieSocialPageQuery, GetWatchlistQuery, IsOnWatchlistQuery, LoginQuery,
ProfileSettingsPageData, RegisterPageData, RemoteActorView, WatchlistPageData,
},
queries::{ExportQuery, GetMovieSocialPageQuery, IsOnWatchlistQuery, LoginQuery},
use_cases::{
add_to_watchlist, delete_review, export_diary as export_diary_uc, get_movie_social_page,
get_watchlist, is_on_watchlist, log_review, login as login_uc, register as register_uc,
remove_from_watchlist, update_profile, update_profile_fields,
is_on_watchlist, log_review, login as login_uc, remove_from_watchlist, update_profile,
update_profile_fields,
},
};
#[cfg(feature = "federation")]
use application::{
ports::{
BlockedActorEntry, BlockedActorsPageData, BlockedDomainEntry, BlockedDomainsPageData,
FollowersPageData, FollowingPageData,
},
use_cases::get_remote_watchlist,
};
use domain::models::ExportFormat;
use domain::{errors::DomainError, value_objects::UserId};
@@ -216,27 +209,21 @@ pub async fn post_register(
if crate::csrf::mismatch(&csrf, &form.csrf_token) {
return StatusCode::FORBIDDEN.into_response();
}
let email = form.email.clone();
let password = form.password.clone();
match register_uc::execute(
match application::use_cases::register_and_login::execute(
&state.app_ctx,
RegisterCommand {
application::commands::RegisterAndLoginCommand {
email: form.email,
username: form.username,
password: form.password,
role: domain::models::UserRole::Standard,
},
)
.await
{
Ok(_) => match login_uc::execute(&state.app_ctx, LoginQuery { email, password }).await {
Ok(result) => {
let max_age = (result.expires_at - Utc::now()).num_seconds().max(0);
let cookie = set_cookie_header(&result.token, max_age);
([cookie], Redirect::to("/")).into_response()
}
Err(_) => Redirect::to("/login").into_response(),
},
Ok(result) => {
let max_age = (result.expires_at - Utc::now()).num_seconds().max(0);
let cookie = set_cookie_header(&result.token, max_age);
([cookie], Redirect::to("/")).into_response()
}
Err(_) => {
Redirect::to("/register?error=Registration+failed.+Please+try+again.").into_response()
}
@@ -369,14 +356,9 @@ pub async fn get_activity_feed(
let limit = params.limit.unwrap_or(20);
let offset = params.offset.unwrap_or(0);
#[cfg(feature = "federation")]
let filter_str = if params.filter == "following" && user_id.is_some() {
"following"
} else {
"all"
};
#[cfg(not(feature = "federation"))]
let filter_str = "all";
let filter_following =
cfg!(feature = "federation") && params.filter == "following" && user_id.is_some();
let filter_str = if filter_following { "following" } else { "all" };
let sort_by_str = match params.sort_by.as_str() {
"date_asc" => "date_asc",
@@ -385,52 +367,17 @@ pub async fn get_activity_feed(
_ => "date",
};
#[cfg(feature = "federation")]
let following = if filter_str == "following" {
if let Some(uid) = user_id {
let urls = state
.social_query
.get_accepted_following_urls(uid.value())
.await
.unwrap_or_default();
let base_url = &state.app_ctx.config.base_url;
let mut local_ids = vec![uid.value()];
let mut remote_urls = Vec::new();
for url in urls {
if let Some(suffix) = url.strip_prefix(&format!("{}/users/", base_url))
&& let Ok(parsed_id) = uuid::Uuid::parse_str(suffix)
{
local_ids.push(parsed_id);
continue;
}
remote_urls.push(url);
}
Some(domain::ports::FollowingFilter {
local_user_ids: local_ids,
remote_actor_urls: remote_urls,
})
} else {
None
}
} else {
None
};
#[cfg(not(feature = "federation"))]
let following: Option<domain::ports::FollowingFilter> = None;
let search_opt = if params.search.is_empty() {
None
} else {
Some(params.search.clone())
};
let query = application::queries::GetActivityFeedQuery {
limit,
offset,
sort_by: sort_by_str.parse().unwrap_or_default(),
search: search_opt,
following,
search: if params.search.is_empty() {
None
} else {
Some(params.search.clone())
},
viewer_user_id: user_id.map(|u| u.value()),
filter_following,
};
match application::use_cases::get_activity_feed::execute(&state.app_ctx, query).await {
@@ -467,27 +414,15 @@ pub async fn get_users_list(
ctx.page_title = "Members — Movies Diary".to_string();
ctx.canonical_url = format!("{}/users", state.app_ctx.config.base_url);
#[cfg(feature = "federation")]
let (users_result, actors_result) = tokio::join!(
application::use_cases::get_users::execute(
&state.app_ctx,
application::queries::GetUsersQuery,
),
state.social_query.list_all_followed_remote_actors()
);
#[cfg(not(feature = "federation"))]
let (users_result, actors_result) = (
application::use_cases::get_users::execute(
&state.app_ctx,
application::queries::GetUsersQuery,
)
.await,
Ok::<Vec<domain::ports::RemoteActorInfo>, domain::errors::DomainError>(vec![]),
);
match (users_result, actors_result) {
(Ok(users), Ok(remote_actors)) => {
let actor_views = remote_actors
match application::use_cases::get_users::execute(
&state.app_ctx,
application::queries::GetUsersQuery,
)
.await
{
Ok(result) => {
let actor_views = result
.remote_actors
.into_iter()
.map(|a| application::ports::RemoteActorView {
handle: a.handle,
@@ -498,7 +433,7 @@ pub async fn get_users_list(
.collect();
let data = application::ports::UsersPageData {
ctx,
users,
users: result.users,
remote_actors: actor_views,
};
match state.html_renderer.render_users_page(data) {
@@ -506,8 +441,7 @@ pub async fn get_users_list(
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e).into_response(),
}
}
(Err(e), _) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
(_, Err(e)) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(),
}
}
@@ -605,57 +539,6 @@ pub async fn get_user_profile(
.map(|u| u.value() == profile_user_uuid)
.unwrap_or(false);
#[cfg(feature = "federation")]
let following_count = if is_own_profile {
if let Some(ref uid) = user_id {
state
.ap_service
.count_following(uid.value())
.await
.unwrap_or(0)
} else {
0
}
} else {
0
};
#[cfg(not(feature = "federation"))]
let following_count = 0usize;
#[cfg(feature = "federation")]
let followers_count = if is_own_profile {
state
.ap_service
.count_accepted_followers(profile_user_uuid)
.await
.unwrap_or(0)
} else {
0
};
#[cfg(not(feature = "federation"))]
let followers_count = 0usize;
#[cfg(feature = "federation")]
let pending_followers: Vec<application::ports::RemoteActorView> = if is_own_profile {
state
.ap_service
.get_pending_followers(profile_user_uuid)
.await
.unwrap_or_default()
.into_iter()
.map(|a| application::ports::RemoteActorView {
handle: a.handle,
url: a.url,
display_name: a.display_name,
avatar_url: a.avatar_url.clone(),
})
.collect()
} else {
vec![]
};
#[cfg(not(feature = "federation"))]
let pending_followers: Vec<application::ports::RemoteActorView> = vec![];
let query = application::queries::GetUserProfileQuery {
user_id: profile_user_uuid,
view: profile_view,
@@ -667,6 +550,7 @@ pub async fn get_user_profile(
} else {
Some(params.search.clone())
},
is_own_profile,
};
match application::use_cases::get_user_profile::execute(&state.app_ctx, query).await {
@@ -682,6 +566,16 @@ pub async fn get_user_profile(
if !is_own_profile {
ctx.page_rss_url = Some(format!("/users/{}/feed.rss", profile_user_uuid));
}
let pending_followers: Vec<application::ports::RemoteActorView> = profile
.pending_followers
.into_iter()
.map(|p| application::ports::RemoteActorView {
handle: p.handle,
url: p.url,
display_name: p.display_name,
avatar_url: p.avatar_url,
})
.collect();
let data = application::ports::ProfilePageData {
ctx,
profile_user_id: profile_user_uuid,
@@ -696,8 +590,8 @@ pub async fn get_user_profile(
trends: profile.trends,
is_own_profile,
error: params.error,
following_count,
followers_count,
following_count: profile.following_count,
followers_count: profile.followers_count,
pending_followers,
sort_by: sort_by_str.to_string(),
search: params.search.clone(),
@@ -1115,94 +1009,33 @@ pub async fn get_watchlist_page(
Extension(csrf): Extension<CsrfToken>,
) -> impl IntoResponse {
let ctx = build_page_context(&state, viewer_id.clone(), csrf.0).await;
let limit = params.limit.unwrap_or(20);
let offset = params.offset.unwrap_or(0);
let is_owner = viewer_id.map(|u| u.value() == owner_id).unwrap_or(false);
// Try local user first
let local_user = state
.app_ctx
.user_repository
.find_by_id(&domain::value_objects::UserId::from_uuid(owner_id))
.await
.ok()
.flatten();
let (display_entries, has_more, current_offset, page_limit) = if local_user.is_some() {
match get_watchlist::execute(
&state.app_ctx,
GetWatchlistQuery {
user_id: owner_id,
limit: Some(limit),
offset: Some(offset),
},
)
.await
{
Err(e) => {
tracing::error!("watchlist error: {:?}", e);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
Ok(entries) => {
let has_more = entries.offset + entries.limit < entries.total_count as u32;
let display: Vec<WatchlistDisplayEntry> = entries
.items
.iter()
.map(|w| {
let remove_url = if is_owner {
Some(format!("/watchlist/{}/remove", w.movie.id().value()))
} else {
None
};
WatchlistDisplayEntry {
poster_url: w
.movie
.poster_path()
.map(|p| format!("/images/{}", p.value())),
movie_title: w.movie.title().value().to_string(),
release_year: w.movie.release_year().value(),
movie_url: Some(format!("/movies/{}", w.movie.id().value())),
added_at: w.entry.added_at.format("%b %-d, %Y").to_string(),
remove_url,
}
})
.collect();
(display, has_more, entries.offset, entries.limit)
}
}
} else {
#[cfg(feature = "federation")]
{
let remote_entries = get_remote_watchlist::execute(&state.app_ctx, owner_id)
.await
.unwrap_or_default();
let display: Vec<WatchlistDisplayEntry> = remote_entries
.into_iter()
.map(|e| WatchlistDisplayEntry {
poster_url: e.poster_url,
movie_title: e.movie_title,
release_year: e.release_year,
movie_url: None,
added_at: e.added_at.format("%b %-d, %Y").to_string(),
remove_url: None,
})
.collect();
let len = display.len() as u32;
(display, false, 0u32, len)
}
#[cfg(not(feature = "federation"))]
{
(vec![], false, 0u32, 0u32)
let result = match application::use_cases::get_watchlist_page::execute(
&state.app_ctx,
application::queries::GetWatchlistQuery {
user_id: owner_id,
limit: params.limit.or(Some(20)),
offset: params.offset.or(Some(0)),
},
is_owner,
)
.await
{
Ok(r) => r,
Err(e) => {
tracing::error!("watchlist error: {:?}", e);
return StatusCode::INTERNAL_SERVER_ERROR.into_response();
}
};
let data = WatchlistPageData {
ctx,
owner_id,
display_entries,
current_offset,
has_more,
limit: page_limit,
display_entries: result.display_entries,
current_offset: result.current_offset,
has_more: result.has_more,
limit: result.limit,
is_owner,
error: params.error,
};

2
crates/presentation/src/main.rs
View File

@@ -202,6 +202,8 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> {
profile_fields_repository: profile_fields_repo,
#[cfg(feature = "federation")]
remote_watchlist_repository: remote_watchlist_repo,
#[cfg(feature = "federation")]
social_query: social_query.clone(),
person_command,
person_query,
search_port,

14
crates/presentation/src/tests/extractors.rs
View File

@@ -135,6 +135,18 @@ impl domain::ports::SocialQueryPort for Panic {
) -> Result<Vec<domain::ports::RemoteActorInfo>, DomainError> {
panic!()
}
async fn count_following(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!()
}
async fn count_accepted_followers(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!()
}
async fn get_pending_followers(
&self,
_: uuid::Uuid,
) -> Result<Vec<domain::ports::PendingFollowerInfo>, DomainError> {
panic!()
}
}
#[async_trait::async_trait]
impl StatsRepository for Panic {
@@ -584,6 +596,8 @@ pub fn make_test_state(auth_service: Arc<dyn AuthService>) -> crate::state::AppS
profile_fields_repository: Arc::clone(&repo) as _,
#[cfg(feature = "federation")]
remote_watchlist_repository: Arc::clone(&repo) as _,
#[cfg(feature = "federation")]
social_query: Arc::clone(&repo) as _,
person_command: Arc::clone(&repo) as _,
person_query: Arc::clone(&repo) as _,
search_port: Arc::clone(&repo) as _,

14
crates/presentation/tests/api_test.rs
View File

@@ -370,6 +370,18 @@ impl domain::ports::SocialQueryPort for PanicSocialQuery {
) -> Result<Vec<domain::ports::RemoteActorInfo>, DomainError> {
panic!()
}
async fn count_following(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!()
}
async fn count_accepted_followers(&self, _: uuid::Uuid) -> Result<usize, DomainError> {
panic!()
}
async fn get_pending_followers(
&self,
_: uuid::Uuid,
) -> Result<Vec<domain::ports::PendingFollowerInfo>, DomainError> {
panic!()
}
}
async fn test_app() -> Router {
@@ -402,6 +414,8 @@ async fn test_app() -> Router {
profile_fields_repository: Arc::new(PanicProfileFields),
#[cfg(feature = "federation")]
remote_watchlist_repository: Arc::new(PanicRemoteWatchlist),
#[cfg(feature = "federation")]
social_query: Arc::new(PanicSocialQuery),
person_command: Arc::new(PanicPersonCommand),
person_query: Arc::new(PanicPersonQuery),
search_port: Arc::new(PanicSearchPort),

4
crates/worker/src/main.rs
View File

@@ -60,7 +60,7 @@ async fn main() -> anyhow::Result<()> {
fed_follow_repo,
fed_actor_repo,
fed_blocklist_repo,
_fed_social_query,
fed_social_query,
fed_review_store,
fed_remote_watchlist_repo,
) = match &db_pool {
@@ -91,6 +91,8 @@ async fn main() -> anyhow::Result<()> {
profile_fields_repository: Arc::clone(&profile_fields_repo),
#[cfg(feature = "federation")]
remote_watchlist_repository: fed_remote_watchlist_repo.clone(),
#[cfg(feature = "federation")]
social_query: fed_social_query,
person_command: Arc::clone(&person_command),
person_query: Arc::clone(&person_query),
search_port: Arc::clone(&search_port),