From e461c689d9df4222cd531c08a7785d5e441df2f0 Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Sun, 10 May 2026 18:58:41 +0200 Subject: [PATCH] feat: add axum dependency and implement wiring functions for federation repositories --- Cargo.lock | 2 + crates/adapters/activitypub/Cargo.toml | 1 + crates/adapters/activitypub/src/lib.rs | 46 ++++++++++ crates/adapters/auth/src/lib.rs | 11 +++ crates/adapters/metadata/Cargo.toml | 1 + crates/adapters/metadata/src/lib.rs | 11 +++ crates/adapters/poster-fetcher/src/lib.rs | 4 + crates/adapters/poster-storage/src/lib.rs | 4 + .../adapters/postgres-federation/src/lib.rs | 13 +++ crates/adapters/sqlite-federation/src/lib.rs | 13 +++ crates/presentation/src/main.rs | 84 ++++--------------- crates/worker/src/main.rs | 78 ++++------------- 12 files changed, 138 insertions(+), 130 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3b8a889..fe3b58f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10,6 +10,7 @@ dependencies = [ "activitypub_federation", "anyhow", "async-trait", + "axum", "chrono", "domain", "serde", @@ -2761,6 +2762,7 @@ dependencies = [ name = "metadata" version = "0.1.0" dependencies = [ + "anyhow", "async-trait", "domain", "reqwest 0.13.3", diff --git a/crates/adapters/activitypub/Cargo.toml b/crates/adapters/activitypub/Cargo.toml index 5620fd5..533586f 100644 --- a/crates/adapters/activitypub/Cargo.toml +++ b/crates/adapters/activitypub/Cargo.toml @@ -6,6 +6,7 @@ edition = "2024" [dependencies] activitypub-base = { workspace = true } domain = { workspace = true } +axum = { workspace = true } tokio = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index 0139a3e..fcc2fae 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -17,3 +17,49 @@ pub use port::{ActivityPubPort, NoopActivityPubService}; pub use remote_review_repository::RemoteReviewRepository; pub use review_handler::ReviewObjectHandler; pub use user_adapter::DomainUserRepoAdapter; + +pub struct ActivityPubWire { + pub service: std::sync::Arc, + pub router: axum::Router, + pub event_handler: std::sync::Arc, +} + +pub async fn wire( + federation_repo: std::sync::Arc, + review_store: std::sync::Arc, + user_repo: std::sync::Arc, + movie_repo: std::sync::Arc, + review_repo: std::sync::Arc, + diary_repo: std::sync::Arc, + base_url: String, +) -> anyhow::Result { + let concrete = std::sync::Arc::new( + ActivityPubService::new( + federation_repo, + std::sync::Arc::new(DomainUserRepoAdapter(user_repo)), + std::sync::Arc::new(ReviewObjectHandler { + movie_repository: std::sync::Arc::clone(&movie_repo), + diary_repository: diary_repo, + review_store, + base_url: base_url.clone(), + }), + base_url.clone(), + cfg!(debug_assertions), + ) + .await?, + ); + + let router = concrete.router(); + let event_handler = std::sync::Arc::new(ActivityPubEventHandler::new( + std::sync::Arc::clone(&concrete), + movie_repo, + review_repo, + base_url, + )) as std::sync::Arc; + + Ok(ActivityPubWire { + service: concrete as std::sync::Arc, + router, + event_handler, + }) +} diff --git a/crates/adapters/auth/src/lib.rs b/crates/adapters/auth/src/lib.rs index 1805254..93ae497 100644 --- a/crates/adapters/auth/src/lib.rs +++ b/crates/adapters/auth/src/lib.rs @@ -105,3 +105,14 @@ impl PasswordHasher for Argon2PasswordHasher { .is_ok()) } } + +pub fn create() -> anyhow::Result<( + std::sync::Arc, + std::sync::Arc, +)> { + let config = AuthConfig::from_env()?; + Ok(( + std::sync::Arc::new(JwtAuthService::new(config)), + std::sync::Arc::new(Argon2PasswordHasher), + )) +} diff --git a/crates/adapters/metadata/Cargo.toml b/crates/adapters/metadata/Cargo.toml index 0e5a596..3ae1851 100644 --- a/crates/adapters/metadata/Cargo.toml +++ b/crates/adapters/metadata/Cargo.toml @@ -4,6 +4,7 @@ version = "0.1.0" edition = "2024" [dependencies] +anyhow = { workspace = true } async-trait = { workspace = true } reqwest = { workspace = true } serde = { workspace = true } diff --git a/crates/adapters/metadata/src/lib.rs b/crates/adapters/metadata/src/lib.rs index 0a2a612..7a66eda 100644 --- a/crates/adapters/metadata/src/lib.rs +++ b/crates/adapters/metadata/src/lib.rs @@ -65,3 +65,14 @@ impl MetadataClient for MetadataClientImpl { Ok(pm.poster_url) } } + +pub fn create() -> anyhow::Result> { + use anyhow::Context; + if let Ok(key) = std::env::var("TMDB_API_KEY") { + Ok(std::sync::Arc::new(MetadataClientImpl::new_tmdb(key))) + } else { + let key = std::env::var("OMDB_API_KEY") + .context("either TMDB_API_KEY or OMDB_API_KEY must be set")?; + Ok(std::sync::Arc::new(MetadataClientImpl::new_omdb(key))) + } +} diff --git a/crates/adapters/poster-fetcher/src/lib.rs b/crates/adapters/poster-fetcher/src/lib.rs index 571619c..7cd5463 100644 --- a/crates/adapters/poster-fetcher/src/lib.rs +++ b/crates/adapters/poster-fetcher/src/lib.rs @@ -36,3 +36,7 @@ impl PosterFetcherClient for ReqwestPosterFetcher { Ok(bytes.to_vec()) } } + +pub fn create() -> anyhow::Result> { + Ok(std::sync::Arc::new(ReqwestPosterFetcher::new(PosterFetcherConfig::from_env())?)) +} diff --git a/crates/adapters/poster-storage/src/lib.rs b/crates/adapters/poster-storage/src/lib.rs index 6910c27..745561f 100644 --- a/crates/adapters/poster-storage/src/lib.rs +++ b/crates/adapters/poster-storage/src/lib.rs @@ -68,6 +68,10 @@ impl PosterStorage for PosterStorageAdapter { } } +pub fn create() -> anyhow::Result> { + Ok(std::sync::Arc::new(PosterStorageAdapter::from_config(StorageConfig::from_env()?))) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/adapters/postgres-federation/src/lib.rs b/crates/adapters/postgres-federation/src/lib.rs index 7aa5f72..2a7d104 100644 --- a/crates/adapters/postgres-federation/src/lib.rs +++ b/crates/adapters/postgres-federation/src/lib.rs @@ -474,3 +474,16 @@ impl domain::ports::SocialQueryPort for PostgresFederationRepository { Ok(rows.into_iter().map(|(url, handle, display_name)| domain::ports::RemoteActorInfo { url, handle, display_name }).collect()) } } + +pub fn wire(pool: sqlx::PgPool) -> ( + std::sync::Arc, + std::sync::Arc, + std::sync::Arc, +) { + let fed = std::sync::Arc::new(PostgresFederationRepository::new(pool)); + ( + std::sync::Arc::clone(&fed) as _, + std::sync::Arc::clone(&fed) as _, + fed as _, + ) +} diff --git a/crates/adapters/sqlite-federation/src/lib.rs b/crates/adapters/sqlite-federation/src/lib.rs index a4933a6..a23354d 100644 --- a/crates/adapters/sqlite-federation/src/lib.rs +++ b/crates/adapters/sqlite-federation/src/lib.rs @@ -537,6 +537,19 @@ impl domain::ports::SocialQueryPort for SqliteFederationRepository { } } +pub fn wire(pool: sqlx::SqlitePool) -> ( + std::sync::Arc, + std::sync::Arc, + std::sync::Arc, +) { + let fed = std::sync::Arc::new(SqliteFederationRepository::new(pool)); + ( + std::sync::Arc::clone(&fed) as _, + std::sync::Arc::clone(&fed) as _, + fed as _, + ) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/presentation/src/main.rs b/crates/presentation/src/main.rs index 5784597..e76b10f 100644 --- a/crates/presentation/src/main.rs +++ b/crates/presentation/src/main.rs @@ -5,23 +5,8 @@ use anyhow::Context; use tokio::net::TcpListener; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -#[cfg(feature = "sqlite-federation")] -use sqlite_federation::SqliteFederationRepository; -#[cfg(feature = "postgres-federation")] -use postgres_federation::PostgresFederationRepository; - -#[cfg(feature = "federation")] -use activitypub::{ - ActivityPubPort, ActivityPubService, DomainUserRepoAdapter, - ReviewObjectHandler, -}; - use application::{config::AppConfig, context::AppContext}; -use auth::{Argon2PasswordHasher, AuthConfig, JwtAuthService}; use export::ExportAdapter; -use metadata::MetadataClientImpl; -use poster_fetcher::{PosterFetcherConfig, ReqwestPosterFetcher}; -use poster_storage::{PosterStorageAdapter, StorageConfig}; use rss::RssAdapter; use template_askama::AskamaHtmlRenderer; @@ -29,10 +14,7 @@ use doc::ApiDocExt; use presentation::{openapi::ApiDoc, routes, state::AppState}; use utoipa::OpenApi as _; -use domain::ports::{ - AuthService, DiaryExporter, EventPublisher, MetadataClient, - PasswordHasher, PosterFetcherClient, PosterStorage, -}; +use domain::ports::{DiaryExporter, EventPublisher}; #[cfg(not(any(feature = "sqlite", feature = "postgres")))] compile_error!("At least one database backend must be enabled. Use --features sqlite or --features postgres"); @@ -59,26 +41,14 @@ async fn main() -> anyhow::Result<()> { } async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { - let auth_config = AuthConfig::from_env()?; - let storage_config = StorageConfig::from_env()?; let app_config = AppConfig::from_env(); let database_url = std::env::var("DATABASE_URL").context("DATABASE_URL must be set")?; let backend = std::env::var("DATABASE_BACKEND").unwrap_or_else(|_| "sqlite".to_string()); - let metadata_client: Arc = - if let Ok(tmdb_key) = std::env::var("TMDB_API_KEY") { - Arc::new(MetadataClientImpl::new_tmdb(tmdb_key)) - } else { - let omdb_key = std::env::var("OMDB_API_KEY") - .context("Either TMDB_API_KEY or OMDB_API_KEY must be set")?; - Arc::new(MetadataClientImpl::new_omdb(omdb_key)) - }; - let poster_fetcher: Arc = - Arc::new(ReqwestPosterFetcher::new(PosterFetcherConfig::from_env())?); - let poster_storage: Arc = - Arc::new(PosterStorageAdapter::from_config(storage_config)); - let auth_service: Arc = Arc::new(JwtAuthService::new(auth_config)); - let password_hasher: Arc = Arc::new(Argon2PasswordHasher); + let (auth_service, password_hasher) = auth::create()?; + let metadata_client = metadata::create()?; + let poster_fetcher = poster_fetcher::create()?; + let poster_storage = poster_storage::create()?; let (movie_repository, review_repository, diary_repository, stats_repository, user_repository, db_pool) = match backend.as_str() { @@ -101,44 +71,26 @@ async fn wire_dependencies() -> anyhow::Result<(AppState, axum::Router)> { #[cfg(feature = "federation")] let (event_publisher_arc, ap_router, ap_service, social_query) = { - let (federation_repo, social_query_arc, review_store): ( - Arc, - Arc, - Arc, - ) = match &db_pool { + let (federation_repo, social_query_arc, review_store) = match &db_pool { #[cfg(feature = "postgres-federation")] - DbPool::Postgres(pool) => { - let fed = Arc::new(PostgresFederationRepository::new(pool.clone())); - (Arc::clone(&fed) as _, Arc::clone(&fed) as _, fed as _) - } + DbPool::Postgres(pool) => postgres_federation::wire(pool.clone()), #[cfg(feature = "sqlite-federation")] - DbPool::Sqlite(pool) => { - let fed = Arc::new(SqliteFederationRepository::new(pool.clone())); - (Arc::clone(&fed) as _, Arc::clone(&fed) as _, fed as _) - } + DbPool::Sqlite(pool) => sqlite_federation::wire(pool.clone()), #[cfg(not(feature = "sqlite-federation"))] _ => anyhow::bail!("DATABASE_BACKEND={backend} federation is not supported by this build"), }; - let user_repo_adapter = Arc::new(DomainUserRepoAdapter(Arc::clone(&user_repository))); - let review_handler = Arc::new(ReviewObjectHandler { - movie_repository: Arc::clone(&movie_repository), - diary_repository: Arc::clone(&diary_repository), + let ap = activitypub::wire( + federation_repo, review_store, - base_url: app_config.base_url.clone(), - }); - let concrete_ap_service = Arc::new( - ActivityPubService::new( - federation_repo, - user_repo_adapter, - review_handler, - app_config.base_url.clone(), - cfg!(debug_assertions), - ) - .await?, - ); - let ap_router = concrete_ap_service.router(); - let ap_service_arc: Arc = concrete_ap_service; + Arc::clone(&user_repository), + Arc::clone(&movie_repository), + Arc::clone(&review_repository), + Arc::clone(&diary_repository), + app_config.base_url.clone(), + ).await?; + let ap_router = ap.router; + let ap_service_arc = ap.service; let ep: Arc = match event_bus { EventBusBackend::Db => { diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index b9fa7bf..03752d1 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -2,27 +2,11 @@ use std::sync::Arc; use anyhow::Context; use application::{config::AppConfig, context::AppContext, event_handlers::PosterSyncHandler, worker::WorkerService}; -use auth::{Argon2PasswordHasher, AuthConfig, JwtAuthService}; use export::ExportAdapter; -use metadata::MetadataClientImpl; -use poster_fetcher::{PosterFetcherConfig, ReqwestPosterFetcher}; -use poster_storage::{PosterStorageAdapter, StorageConfig}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -#[cfg(feature = "sqlite-federation")] -use sqlite_federation::SqliteFederationRepository; -#[cfg(feature = "postgres-federation")] -use postgres_federation::PostgresFederationRepository; -#[cfg(feature = "federation")] -use activitypub::{ - ActivityPubEventHandler, ActivityPubService, DomainUserRepoAdapter, ReviewObjectHandler, -}; - -use domain::ports::{ - AuthService, DiaryExporter, EventHandler, MetadataClient, - PasswordHasher, PosterFetcherClient, PosterStorage, -}; +use domain::ports::{DiaryExporter, EventHandler}; #[cfg(not(any(feature = "sqlite", feature = "postgres")))] compile_error!("At least one database backend must be enabled. Use --features sqlite or --features postgres"); @@ -34,24 +18,12 @@ async fn main() -> anyhow::Result<()> { let database_url = std::env::var("DATABASE_URL").context("DATABASE_URL must be set")?; let backend = std::env::var("DATABASE_BACKEND").unwrap_or_else(|_| "sqlite".to_string()); - let auth_config = AuthConfig::from_env()?; - let storage_config = StorageConfig::from_env()?; let app_config = AppConfig::from_env(); - let metadata_client: Arc = - if let Ok(tmdb_key) = std::env::var("TMDB_API_KEY") { - Arc::new(MetadataClientImpl::new_tmdb(tmdb_key)) - } else { - let omdb_key = std::env::var("OMDB_API_KEY") - .context("Either TMDB_API_KEY or OMDB_API_KEY must be set")?; - Arc::new(MetadataClientImpl::new_omdb(omdb_key)) - }; - let poster_fetcher: Arc = - Arc::new(ReqwestPosterFetcher::new(PosterFetcherConfig::from_env())?); - let poster_storage: Arc = - Arc::new(PosterStorageAdapter::from_config(storage_config)); - let auth_service: Arc = Arc::new(JwtAuthService::new(auth_config)); - let password_hasher: Arc = Arc::new(Argon2PasswordHasher); + let (auth_service, password_hasher) = auth::create()?; + let metadata_client = metadata::create()?; + let poster_fetcher = poster_fetcher::create()?; + let poster_storage = poster_storage::create()?; let (movie_repository, review_repository, diary_repository, stats_repository, user_repository, db_pool) = match backend.as_str() { @@ -125,44 +97,22 @@ async fn main() -> anyhow::Result<()> { #[cfg(feature = "federation")] { - let (federation_repo, review_store): ( - Arc, - Arc, - ) = match &db_pool { + let (federation_repo, _social_query, review_store) = match &db_pool { #[cfg(feature = "sqlite-federation")] - DbPool::Sqlite(pool) => { - let fed = Arc::new(SqliteFederationRepository::new(pool.clone())); - (Arc::clone(&fed) as _, fed as _) - } + DbPool::Sqlite(pool) => sqlite_federation::wire(pool.clone()), #[cfg(feature = "postgres-federation")] - DbPool::Postgres(pool) => { - let fed = Arc::new(PostgresFederationRepository::new(pool.clone())); - (Arc::clone(&fed) as _, fed as _) - } + DbPool::Postgres(pool) => postgres_federation::wire(pool.clone()), }; - let ap_service = Arc::new( - ActivityPubService::new( - federation_repo, - Arc::new(DomainUserRepoAdapter(fed_user_repo)), - Arc::new(ReviewObjectHandler { - movie_repository: Arc::clone(&fed_movie_repo), - diary_repository: fed_diary_repo, - review_store, - base_url: base_url.clone(), - }), - base_url.clone(), - cfg!(debug_assertions), - ) - .await?, - ); - - let ap = Arc::new(ActivityPubEventHandler::new( - ap_service, + let ap = activitypub::wire( + federation_repo, + review_store, + fed_user_repo, fed_movie_repo, fed_review_repo, + fed_diary_repo, base_url, - )) as Arc; + ).await?.event_handler; tracing::info!("federation event handler registered"); vec![poster, ap]