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

structural refactor and codebase improvements

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-08-09 14:58:14 +02:00
parent 22b1dd3f56
commit c9715baab8
247 changed files with 11515 additions and 3063 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
crates/worker/Cargo.toml
View File

@@ -5,17 +5,30 @@ edition = "2024"
[features]
default = ["sqlite", "sqlite-federation"]
sqlite = ["dep:sqlite", "dep:sqlite-event-queue", "dep:sqlite-search", "infra-wiring/sqlite"]
postgres = ["dep:postgres", "dep:postgres-event-queue", "dep:postgres-search", "infra-wiring/postgres"]
nats = ["dep:nats", "infra-wiring/nats"]
sqlite = ["dep:sqlite", "dep:sqlite-event-queue", "dep:sqlite-search", "infra-wiring/sqlite", "composition/sqlite"]
postgres = ["dep:postgres", "dep:postgres-event-queue", "dep:postgres-search", "infra-wiring/postgres", "composition/postgres"]
nats = ["dep:nats", "infra-wiring/nats", "composition/nats"]
federation = ["application/federation"]
sqlite-federation = ["sqlite", "dep:sqlite-federation", "dep:activitypub", "federation"]
postgres-federation = ["postgres", "dep:postgres-federation", "dep:activitypub", "federation"]
sqlite-federation = [
"sqlite",
"dep:sqlite-federation",
"dep:activitypub",
"federation",
"composition/sqlite-federation",
]
postgres-federation = [
"postgres",
"dep:postgres-federation",
"dep:activitypub",
"federation",
"composition/postgres-federation",
]
[dependencies]
domain = { workspace = true }
application = { workspace = true }
tokio = { workspace = true }
composition = { workspace = true }
tokio = { workspace = true, features = ["signal"] }
anyhow = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }

125
crates/worker/src/db.rs
View File

@@ -1,125 +0,0 @@
use std::sync::Arc;
use anyhow::Context;
use domain::ports::{
DiaryQuery, GoalCommand, GoalQuery, ImageRefCommand, ImageRefQuery, ImportSessionRepository,
LocalApContentQuery, MovieCommand, MovieDeduplicator, MovieProfileRepository, MovieQuery,
PersonCommand, PersonQuery, ReviewRepository, SearchCommand, StatsRepository, UserRepository,
WatchEventCommand, WatchEventQuery,
};
pub use infra_wiring::DbPool;
pub struct WorkerDbOutput {
pub movie_command: Arc<dyn MovieCommand>,
pub movie_query: Arc<dyn MovieQuery>,
pub review: Arc<dyn ReviewRepository>,
pub diary: Arc<dyn DiaryQuery>,
pub stats: Arc<dyn StatsRepository>,
pub _goal_command: Arc<dyn GoalCommand>,
pub goal_query: Arc<dyn GoalQuery>,
pub user: Arc<dyn UserRepository>,
pub import_session: Arc<dyn ImportSessionRepository>,
pub movie_profile: Arc<dyn MovieProfileRepository>,
pub watch_event_command: Arc<dyn WatchEventCommand>,
pub _watch_event_query: Arc<dyn WatchEventQuery>,
pub person_command: Arc<dyn PersonCommand>,
pub person_query: Arc<dyn PersonQuery>,
pub search_command: Arc<dyn SearchCommand>,
pub ap_content: Arc<dyn LocalApContentQuery>,
pub image_ref_command: Arc<dyn ImageRefCommand>,
pub image_ref_query: Arc<dyn ImageRefQuery>,
pub wrapup_stats: Arc<dyn domain::ports::WrapUpStatsQuery>,
pub wrapup_repo: Arc<dyn domain::ports::WrapUpRepository>,
pub remote_goal: Arc<dyn domain::ports::RemoteGoalRepository>,
pub refresh_session: Arc<dyn domain::ports::RefreshSessionRepository>,
pub federation_settings: Arc<dyn domain::ports::UserFederationSettingsQuery>,
pub deduplicator: Arc<dyn MovieDeduplicator>,
pub db_pool: DbPool,
}
pub async fn connect(database_url: &str, backend: &str) -> anyhow::Result<WorkerDbOutput> {
match backend {
#[cfg(feature = "postgres")]
"postgres" => {
let w = postgres::wire(database_url)
.await
.context("PostgreSQL connection failed")?;
let (image_ref_command, image_ref_query) = postgres::create_image_ref(w.pool.clone());
let (person_command, person_query) = postgres::create_person_adapter(w.pool.clone());
let (search_command, _search_port) =
postgres_search::create_search_adapter(w.pool.clone());
let we = Arc::new(postgres::PostgresWatchEventRepository::new(w.pool.clone()));
Ok(WorkerDbOutput {
movie_command: w.movie_command,
movie_query: w.movie_query,
review: w.review,
diary: w.diary,
stats: w.stats,
_goal_command: w.goal_command,
goal_query: w.goal_query,
user: w.user,
import_session: w.import_session,
movie_profile: w.movie_profile,
watch_event_command: we.clone() as _,
_watch_event_query: we as _,
person_command,
person_query,
search_command,
ap_content: w.ap_content,
image_ref_command,
image_ref_query,
wrapup_stats: w.wrapup_stats,
wrapup_repo: w.wrapup_repo,
remote_goal: w.remote_goal,
refresh_session: Arc::new(postgres::PostgresRefreshSessionAdapter::new(
w.pool.clone(),
)) as _,
federation_settings: w.federation_settings,
deduplicator: w.deduplicator,
db_pool: DbPool::Postgres(w.pool),
})
}
#[cfg(feature = "sqlite")]
_ => {
let w = sqlite::wire(database_url)
.await
.context("SQLite connection failed")?;
let (image_ref_command, image_ref_query) = sqlite::create_image_ref(w.pool.clone());
let (person_command, person_query) = sqlite::create_person_adapter(w.pool.clone());
let (search_command, _search_port) =
sqlite_search::create_search_adapter(w.pool.clone());
let we = Arc::new(sqlite::SqliteWatchEventRepository::new(w.pool.clone()));
Ok(WorkerDbOutput {
movie_command: w.movie_command,
movie_query: w.movie_query,
review: w.review,
diary: w.diary,
stats: w.stats,
_goal_command: w.goal_command,
goal_query: w.goal_query,
user: w.user,
import_session: w.import_session,
movie_profile: w.movie_profile,
watch_event_command: we.clone() as _,
_watch_event_query: we as _,
person_command,
person_query,
search_command,
ap_content: w.ap_content,
image_ref_command,
image_ref_query,
wrapup_stats: w.wrapup_stats,
wrapup_repo: w.wrapup_repo,
remote_goal: w.remote_goal,
refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone()))
as _,
federation_settings: w.federation_settings,
deduplicator: w.deduplicator,
db_pool: DbPool::Sqlite(w.pool),
})
}
#[cfg(not(feature = "sqlite"))]
_ => anyhow::bail!("DATABASE_BACKEND={backend} is not supported by this build"),
}
}

3
crates/worker/src/event_bus.rs
View File

@@ -4,8 +4,7 @@ use std::sync::Arc;
use anyhow::Context;
use domain::ports::{EventConsumer, EventPublisher};
use crate::db::DbPool;
use infra_wiring::EventBusBackend;
use infra_wiring::{DbPool, EventBusBackend};
pub async fn create(
db_pool: &DbPool,

21
crates/worker/src/follow_backfill_handler.rs
View File

@@ -2,11 +2,14 @@ use std::sync::Arc;
use async_trait::async_trait;
use domain::{
errors::DomainError, events::DomainEvent, ports::EventHandler, value_objects::SocialIdentity,
errors::DomainError,
events::DomainEvent,
ports::{ApBackfillPort, EventHandler},
value_objects::SocialIdentity,
};
pub struct FollowBackfillHandler {
pub ap_service: Arc<dyn activitypub::ActivityPubPort>,
pub backfill: Arc<dyn ApBackfillPort>,
}
#[async_trait]
@@ -18,18 +21,13 @@ impl EventHandler for FollowBackfillHandler {
requester: SocialIdentity::Remote { actor_url },
} => {
tracing::info!(actor = %actor_url, "follow accepted — looking up outbox for import");
let following = self
.ap_service
.get_following(owner.value())
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
let following = self.backfill.get_following(owner.value()).await?;
if let Some(actor) = following.iter().find(|a| a.url == *actor_url) {
if let Some(outbox_url) = &actor.outbox_url {
tracing::info!(outbox = %outbox_url, actor = %actor_url, "importing remote outbox");
self.ap_service
self.backfill
.import_remote_outbox(outbox_url, actor_url)
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))?;
.await?;
} else {
tracing::warn!(actor = %actor_url, "no outbox URL for accepted follow — skipping import");
}
@@ -41,10 +39,9 @@ impl EventHandler for FollowBackfillHandler {
follower_inbox_url,
} => {
tracing::info!(owner = %owner_user_id.value(), inbox = %follower_inbox_url, "backfilling local content to new follower");
self.ap_service
self.backfill
.run_backfill_for_follower(owner_user_id.value(), follower_inbox_url.clone())
.await
.map_err(|e| DomainError::InfrastructureError(e.to_string()))
}
_ => Ok(()),
}

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

@@ -1,4 +1,3 @@
mod db;
mod event_bus;
mod follow_backfill_handler;
@@ -7,7 +6,7 @@ use std::sync::Arc;
use anyhow::Context;
use application::{
MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, config::AppConfig,
movies::deps::ReindexSearchDeps, worker::WorkerService,
worker::WorkerService,
};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -26,14 +25,39 @@ 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 app_config = AppConfig::from_env();
let instance = domain::value_objects::InstanceIdentity::new(app_config.base_url.clone());
let metadata_client = metadata::create()?;
let poster_fetcher = poster_fetcher::create()?;
let object_storage = object_storage::create()?;
let db = db::connect(&database_url, &backend).await?;
// TMDb client detection happens before `build_worker_deps` so the optional
// person-enrichment port can be handed to `WorkerServices` up front.
let tmdb_client = match tmdb_enrichment::TmdbEnrichmentClient::from_env() {
Ok(client) => {
tracing::info!("TMDb enrichment enabled");
Some(Arc::new(client))
}
Err(e) => {
tracing::warn!("TMDb enrichment disabled: {e}");
None
}
};
let person_enrichment: Option<Arc<dyn PersonEnrichmentClient>> = tmdb_client
.clone()
.map(|c| c as Arc<dyn PersonEnrichmentClient>);
let db =
composition::factory::build_database_adapters(&backend, &database_url, &instance).await?;
let (event_publisher_arc, consumer_arc) = event_bus::create(&db.db_pool).await?;
let worker_services = application::WorkerServices {
object_storage: Arc::clone(&object_storage),
event_publisher: Arc::clone(&event_publisher_arc),
person_enrichment,
};
let worker_deps = composition::build_worker_deps(&db, &worker_services);
let image_ref_command = Arc::clone(&db.image_ref_command);
let image_ref_query = Arc::clone(&db.image_ref_query);
@@ -46,7 +70,6 @@ async fn main() -> anyhow::Result<()> {
fed_goal_repo,
fed_stats_repo,
fed_user_repo,
base_url,
allow_registration,
) = (
Arc::clone(&db.ap_content),
@@ -56,29 +79,29 @@ async fn main() -> anyhow::Result<()> {
Arc::clone(&db.goal_query),
Arc::clone(&db.stats),
Arc::clone(&db.user),
app_config.base_url.clone(),
app_config.allow_registration,
);
// Wire federation repos early to get remote_watchlist_repo for AppContext.
#[cfg(feature = "federation")]
let fed_repos = match &db.db_pool {
#[cfg(feature = "sqlite-federation")]
db::DbPool::Sqlite(pool) => sqlite_federation::wire(pool.clone()),
composition::DbPool::Sqlite(pool) => {
sqlite_federation::wire(pool.clone(), instance.clone())
}
#[cfg(feature = "postgres-federation")]
db::DbPool::Postgres(pool) => postgres_federation::wire(pool.clone()),
composition::DbPool::Postgres(pool) => {
postgres_federation::wire(pool.clone(), instance.clone())
}
};
let movie_command = db.movie_command;
let movie_query = db.movie_query;
let deduplicator = db.deduplicator;
let user = db.user;
let import_session = db.import_session;
let movie_profile = db.movie_profile;
let watch_event_command = db.watch_event_command;
let person_command = db.person_command;
let person_query = db.person_query;
let search_command = db.search_command;
let wrapup_stats = db.wrapup_stats;
let wrapup_repo = db.wrapup_repo;
let remote_goal = db.remote_goal;
let refresh_session = db.refresh_session;
@@ -97,25 +120,22 @@ async fn main() -> anyhow::Result<()> {
Option<Arc<dyn PeriodicJob>>,
);
let (enrichment_handler, person_enrichment_handler, enrichment_job): EnrichmentParts =
match tmdb_enrichment::TmdbEnrichmentClient::from_env() {
Ok(client) => {
tracing::info!("TMDb enrichment enabled");
let client = Arc::new(client);
match tmdb_client {
Some(client) => {
let image_fetcher = poster_fetcher::create_image_fetcher()?;
let handler = Arc::new(application::movies::MovieEnrichmentHandler::new(
Arc::clone(&client) as Arc<dyn MovieEnrichmentClient>,
Arc::clone(&movie_query),
Arc::clone(&movie_profile),
Arc::clone(&person_command),
Arc::clone(&search_command),
worker_deps.enrich_movie.movie_query.clone(),
worker_deps.enrich_movie.movie_profile.clone(),
worker_deps.enrich_movie.person_command.clone(),
worker_deps.enrich_movie.search_command.clone(),
Arc::clone(&object_storage),
image_fetcher,
)) as Arc<dyn EventHandler>;
let person_enrichment_arc = Arc::clone(&client) as Arc<dyn PersonEnrichmentClient>;
let person_handler = Arc::new(application::person::PersonEnrichmentHandler::new(
Arc::clone(&person_query),
Some(person_enrichment_arc),
Arc::clone(&person_command),
worker_deps.enrich_person.person_query.clone(),
worker_deps.enrich_person.person_enrichment.clone(),
worker_deps.enrich_person.person_command.clone(),
)) as Arc<dyn EventHandler>;
let job = Arc::new(application::jobs::EnrichmentStalenessJob::new(
Arc::clone(&movie_profile),
@@ -123,10 +143,7 @@ async fn main() -> anyhow::Result<()> {
)) as Arc<dyn PeriodicJob>;
(Some(handler), Some(person_handler), Some(job))
}
Err(e) => {
tracing::warn!("TMDb enrichment disabled: {e}");
(None, None, None)
}
None => (None, None, None),
};
// ── Image conversion ──────────────────────────────────────────────────────
@@ -142,9 +159,9 @@ async fn main() -> anyhow::Result<()> {
let mut periodic_jobs: Vec<Arc<dyn PeriodicJob>> = vec![
Arc::new(application::jobs::MovieDeduplicationJob::new(
Arc::clone(&movie_query),
Arc::clone(&deduplicator),
Arc::clone(&object_storage),
worker_deps.merge_duplicates.movie_query.clone(),
worker_deps.merge_duplicates.deduplicator.clone(),
worker_deps.merge_duplicates.object_storage.clone(),
)),
Arc::new(application::jobs::ImportSessionCleanupJob::new(
import_session.clone(),
@@ -211,18 +228,13 @@ async fn main() -> anyhow::Result<()> {
)) as Arc<dyn EventHandler>;
let wrapup_handler = Arc::new(application::wrapup::event_handler::WrapUpEventHandler::new(
Arc::clone(&wrapup_repo),
Arc::clone(&event_publisher),
Arc::clone(&wrapup_stats),
worker_deps.handle_requested.wrapup_repo.clone(),
worker_deps.handle_requested.event_publisher.clone(),
worker_deps.handle_requested.wrapup_stats.clone(),
)) as Arc<dyn EventHandler>;
let reindex_handler = Arc::new(SearchReindexHandler::new(ReindexSearchDeps {
movie_query: Arc::clone(&movie_query),
movie_profile: Arc::clone(&movie_profile),
search_command: Arc::clone(&search_command),
person_command: Arc::clone(&person_command),
person_query: Arc::clone(&person_query),
})) as Arc<dyn EventHandler>;
let reindex_handler = Arc::new(SearchReindexHandler::new(worker_deps.reindex_search))
as Arc<dyn EventHandler>;
let mut h = vec![
poster,
@@ -252,7 +264,7 @@ async fn main() -> anyhow::Result<()> {
user_repo: fed_user_repo,
follow_command: fed_repos.follow_command,
follow_query: fed_repos.follow_query,
base_url,
instance: instance.clone(),
allow_registration,
event_publisher: Arc::clone(&event_publisher),
federation_settings: std::sync::Arc::clone(&db.federation_settings),
@@ -262,7 +274,7 @@ async fn main() -> anyhow::Result<()> {
tracing::info!("federation event handler registered");
h.push(ap_wire.event_handler);
h.push(Arc::new(follow_backfill_handler::FollowBackfillHandler {
ap_service: ap_wire.service,
backfill: ap_wire.backfill,
}) as Arc<dyn EventHandler>);
}