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

refactor: bootstrap crate, thin server/worker binaries

Browse Source
This commit is contained in:
Gabriel Kaszewski
2026-10-25 21:16:38 +01:00
parent 04f9c13625
commit 9bf9342fc0
39 changed files with 939 additions and 905 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
crates/worker/Cargo.toml
View File

@@ -5,43 +5,16 @@ edition.workspace = true
[features]
default = ["federation"]
federation = [
"dep:activitypub",
"dep:sqlite-federation",
"application/federation",
"composition/federation",
]
nats = ["dep:nats", "infra-wiring/nats", "composition/nats"]
federation = ["bootstrap/federation"]
nats = ["bootstrap/nats"]
# deprecated aliases kept so existing deploy commands keep working
sqlite = []
sqlite-federation = ["federation"]
[dependencies]
bootstrap = { workspace = true }
config = { workspace = true }
domain = { workspace = true }
application = { workspace = true }
composition = { workspace = true }
tokio = { workspace = true, features = ["signal"] }
anyhow = { workspace = true }
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
dotenvy = { workspace = true }
auth = { workspace = true }
metadata = { workspace = true }
poster-fetcher = { workspace = true }
object-storage = { workspace = true }
poster-sync = { workspace = true }
export = { workspace = true }
tmdb-enrichment = { workspace = true }
importer = { workspace = true }
image-converter = { workspace = true }
nats = { workspace = true, optional = true }
sqlx = { workspace = true }
infra-wiring = { workspace = true }
async-trait = { workspace = true }
sqlite = { workspace = true }
# Optional — federation
activitypub = { workspace = true, optional = true }
sqlite-federation = { workspace = true, optional = true }
tracing = { workspace = true }

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

@@ -1,32 +0,0 @@
use std::sync::Arc;
#[cfg(feature = "nats")]
use anyhow::Context;
use domain::ports::{EventConsumer, EventPublisher};
use infra_wiring::{DbPool, EventBusBackend};
pub async fn create(
db_pool: &DbPool,
config: &config::AppConfig,
) -> anyhow::Result<(Arc<dyn EventPublisher>, Arc<dyn EventConsumer>)> {
match EventBusBackend::from_config(&config.events)? {
EventBusBackend::Db => {
tracing::info!("event bus: DB queue");
match db_pool {
DbPool::Sqlite(pool) => Ok(sqlite::SqliteEventQueue::create_channel(
pool.clone(),
&config.events,
)
.await?),
}
}
#[cfg(feature = "nats")]
EventBusBackend::Nats => {
let cfg = nats::NatsConfig::from_config(&config.nats)
.context("EVENT_BUS_BACKEND=nats requires NATS_URL to be set")?;
tracing::info!("event bus: NATS ({})", cfg.url);
Ok(nats::create_channel(cfg).await?)
}
}
}

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

@@ -1,49 +0,0 @@
use std::sync::Arc;
use async_trait::async_trait;
use domain::{
errors::DomainError,
events::DomainEvent,
ports::{ApBackfillPort, EventHandler},
value_objects::SocialIdentity,
};
pub struct FollowBackfillHandler {
pub backfill: Arc<dyn ApBackfillPort>,
}
#[async_trait]
impl EventHandler for FollowBackfillHandler {
async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
match event {
DomainEvent::FollowAccepted {
owner,
requester: SocialIdentity::Remote { actor_url },
} => {
tracing::info!(actor = %actor_url, "follow accepted — looking up outbox for import");
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.backfill
.import_remote_outbox(outbox_url, actor_url)
.await?;
} else {
tracing::warn!(actor = %actor_url, "no outbox URL for accepted follow — skipping import");
}
}
Ok(())
}
DomainEvent::BackfillFollower {
owner_user_id,
follower_inbox_url,
} => {
tracing::info!(owner = %owner_user_id.value(), inbox = %follower_inbox_url, "backfilling local content to new follower");
self.backfill
.run_backfill_for_follower(owner_user_id.value(), follower_inbox_url.clone())
.await
}
_ => Ok(()),
}
}
}

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

@@ -1,289 +1,12 @@
mod event_bus;
#[cfg(feature = "federation")]
mod follow_backfill_handler;
use std::sync::Arc;
use anyhow::Context;
use application::{
MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, worker::WorkerService,
};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
use domain::ports::{EventHandler, MovieEnrichmentClient, PeriodicJob, PersonEnrichmentClient};
#[tokio::main]
async fn main() -> anyhow::Result<()> {
dotenvy::dotenv().ok();
init_tracing();
bootstrap::init_tracing("worker=info,application=info,k_ap=info");
let config = config::load().context("Failed to load configuration")?;
composition::factory::warn_if_database_backend_set();
let app_config = application::config::AppConfig::from(&config);
let instance = domain::value_objects::InstanceIdentity::new(app_config.base_url.clone());
let metadata_client = metadata::create(&config.metadata)?;
let poster_fetcher = poster_fetcher::create(&config.metadata)?;
let object_storage = object_storage::create(&config.storage)?;
// 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_config(&config.metadata) {
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(&config.database.url, &instance).await?;
let (event_publisher_arc, consumer_arc) = event_bus::create(&db.db_pool, &config).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);
#[cfg(feature = "federation")]
let (
fed_ap_content,
fed_movie_repo,
fed_review_repo,
fed_diary_repo,
fed_goal_repo,
fed_stats_repo,
fed_user_repo,
allow_registration,
) = (
Arc::clone(&db.ap_content),
Arc::clone(&db.movie_query),
Arc::clone(&db.review),
Arc::clone(&db.diary),
Arc::clone(&db.goal_query),
Arc::clone(&db.stats),
Arc::clone(&db.user),
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 {
composition::DbPool::Sqlite(pool) => {
sqlite_federation::wire(pool.clone(), instance.clone())
}
};
let movie_command = db.movie_command;
let movie_query = db.movie_query;
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_query = db.person_query;
let search_command = db.search_command;
let wrapup_repo = db.wrapup_repo;
#[cfg(feature = "federation")]
let remote_goal = db.remote_goal;
let refresh_session = db.refresh_session;
let event_publisher = event_publisher_arc;
let object_storage = object_storage;
let metadata = metadata_client;
// ── Enrichment ────────────────────────────────────────────────────────────
// Both the event handler and the staleness job are gated on TMDB_API_KEY.
// Without a key, no MovieEnrichmentRequested events are produced or handled.
type EnrichmentParts = (
Option<Arc<dyn EventHandler>>,
Option<Arc<dyn EventHandler>>,
Option<Arc<dyn PeriodicJob>>,
);
let (enrichment_handler, person_enrichment_handler, enrichment_job): EnrichmentParts =
match tmdb_client {
Some(client) => {
let image_fetcher = poster_fetcher::create_image_fetcher(&config.metadata)?;
let handler = Arc::new(application::movies::MovieEnrichmentHandler::new(
Arc::clone(&client) as Arc<dyn MovieEnrichmentClient>,
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_handler = Arc::new(application::person::PersonEnrichmentHandler::new(
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),
Arc::clone(&event_publisher),
)) as Arc<dyn PeriodicJob>;
(Some(handler), Some(person_handler), Some(job))
}
None => (None, None, None),
};
// ── Image conversion ──────────────────────────────────────────────────────
let conversion = image_converter::build(
&config.image_conversion,
Arc::clone(&object_storage),
image_ref_command,
image_ref_query,
Arc::clone(&event_publisher),
)?;
// ── Periodic jobs ─────────────────────────────────────────────────────────
let mut periodic_jobs: Vec<Arc<dyn PeriodicJob>> = vec![
Arc::new(application::jobs::MovieDeduplicationJob::new(
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(),
)),
Arc::new(application::jobs::WatchEventCleanupJob::new(
watch_event_command.clone(),
)),
Arc::new(application::jobs::WrapUpAutoGenerateJob::new(
Arc::clone(&user),
Arc::clone(&wrapup_repo),
Arc::clone(&event_publisher),
)),
Arc::new(application::jobs::WrapUpCleanupJob::new(Arc::clone(
&wrapup_repo,
))),
Arc::new(application::jobs::RefreshSessionCleanupJob::new(
Arc::clone(&refresh_session),
)),
];
if let Some(job) = enrichment_job {
periodic_jobs.push(job);
}
if let Some((_, ref conv_job)) = conversion {
periodic_jobs.push(Arc::clone(conv_job));
}
for job in periodic_jobs {
tokio::spawn(async move {
let mut tick = tokio::time::interval(job.interval());
loop {
tick.tick().await;
if let Err(e) = job.run().await {
tracing::error!("periodic job failed: {e}");
}
}
});
}
// ── Event handlers ────────────────────────────────────────────────────────
let handlers: Vec<Arc<dyn EventHandler>> = {
let poster = Arc::new(poster_sync::PosterSyncHandler::new(
Arc::clone(&movie_command),
Arc::clone(&movie_query),
Arc::clone(&metadata),
Arc::clone(&poster_fetcher),
Arc::clone(&object_storage),
Arc::clone(&event_publisher),
3,
)) as Arc<dyn EventHandler>;
let cleanup = Arc::new(object_storage::ImageCleanupHandler::new(Arc::clone(
&object_storage,
))) as Arc<dyn EventHandler>;
let search_cleanup = Arc::new(SearchCleanupHandler::new(
Arc::clone(&search_command),
Arc::clone(&person_query),
)) as Arc<dyn EventHandler>;
let discovery_indexer = Arc::new(MovieDiscoveryIndexer::new(
Arc::clone(&movie_query),
Arc::clone(&search_command),
)) as Arc<dyn EventHandler>;
let wrapup_handler = Arc::new(application::wrapup::event_handler::WrapUpEventHandler::new(
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(worker_deps.reindex_search))
as Arc<dyn EventHandler>;
let mut h = vec![
poster,
cleanup,
search_cleanup,
discovery_indexer,
wrapup_handler,
reindex_handler,
];
#[cfg(feature = "federation")]
{
let ap_wire = activitypub::wire(activitypub::ActivityPubDeps {
activity_repo: fed_repos.activity,
follow_repo: fed_repos.follow,
actor_repo: fed_repos.actor,
blocklist_repo: fed_repos.blocklist,
review_store: fed_repos.review_store,
remote_watchlist_repo: fed_repos.remote_watchlist,
remote_goal_repo: Arc::clone(&remote_goal),
local_ap_content: fed_ap_content,
movie_repo: fed_movie_repo,
review_repo: fed_review_repo,
diary_repo: fed_diary_repo,
goal_repo: fed_goal_repo,
stats_repo: fed_stats_repo,
user_repo: fed_user_repo,
follow_command: fed_repos.follow_command,
follow_query: fed_repos.follow_query,
instance: instance.clone(),
allow_registration,
federation: config.federation.clone(),
event_publisher: Arc::clone(&event_publisher),
federation_settings: std::sync::Arc::clone(&db.federation_settings),
})
.await?;
tracing::info!("federation event handler registered");
h.push(ap_wire.event_handler);
h.push(Arc::new(follow_backfill_handler::FollowBackfillHandler {
backfill: ap_wire.backfill,
}) as Arc<dyn EventHandler>);
}
if let Some(e) = enrichment_handler {
h.push(e);
}
if let Some(e) = person_enrichment_handler {
h.push(e);
}
if let Some((ref conv_handler, _)) = conversion {
h.push(Arc::clone(conv_handler));
}
h
};
// ── Run ───────────────────────────────────────────────────────────────────
let app = bootstrap::build_worker(&config).await?;
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
tokio::spawn(async move {
@@ -291,32 +14,8 @@ async fn main() -> anyhow::Result<()> {
let _ = shutdown_tx.send(true);
});
let worker = WorkerService::new(consumer_arc, handlers);
tracing::info!("worker started");
worker.run(shutdown_rx).await;
app.service.run(shutdown_rx).await;
Ok(())
}
fn init_tracing() {
tracing_subscriber::registry()
.with(log_filter("worker=info,application=info,k_ap=info"))
.with(tracing_subscriber::fmt::layer())
.init();
}
/// `RUST_LOG` names the crates to trace in detail; anything it doesn't name
/// still reports warnings and errors. Without a baseline, a filter like
/// `presentation=debug` silently drops adapter errors (e.g. the sqlx detail
/// behind "Database operation failed").
fn log_filter(default: &str) -> tracing_subscriber::EnvFilter {
let directives = std::env::var("RUST_LOG").unwrap_or_else(|_| default.to_string());
let has_baseline = directives
.split(',')
.any(|d| !d.trim().is_empty() && !d.contains('='));
if has_baseline {
tracing_subscriber::EnvFilter::new(directives)
} else {
tracing_subscriber::EnvFilter::new(format!("warn,{directives}"))
}
}