refactor: adapters configured from config crate
This commit is contained in:
@@ -17,6 +17,7 @@ sqlite = []
|
||||
sqlite-federation = ["federation"]
|
||||
|
||||
[dependencies]
|
||||
config = { workspace = true }
|
||||
domain = { workspace = true }
|
||||
application = { workspace = true }
|
||||
composition = { workspace = true }
|
||||
|
||||
@@ -8,19 +8,22 @@ 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_env()? {
|
||||
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()).await?)
|
||||
}
|
||||
DbPool::Sqlite(pool) => Ok(sqlite::SqliteEventQueue::create_channel(
|
||||
pool.clone(),
|
||||
&config.events,
|
||||
)
|
||||
.await?),
|
||||
}
|
||||
}
|
||||
#[cfg(feature = "nats")]
|
||||
EventBusBackend::Nats => {
|
||||
let cfg = nats::NatsConfig::from_env()
|
||||
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?)
|
||||
|
||||
@@ -6,8 +6,7 @@ use std::sync::Arc;
|
||||
|
||||
use anyhow::Context;
|
||||
use application::{
|
||||
MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, config::AppConfig,
|
||||
worker::WorkerService,
|
||||
MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, worker::WorkerService,
|
||||
};
|
||||
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
|
||||
|
||||
@@ -18,18 +17,18 @@ async fn main() -> anyhow::Result<()> {
|
||||
dotenvy::dotenv().ok();
|
||||
init_tracing();
|
||||
|
||||
let database_url = std::env::var("DATABASE_URL").context("DATABASE_URL must be set")?;
|
||||
let config = config::load().context("Failed to load configuration")?;
|
||||
composition::factory::warn_if_database_backend_set();
|
||||
let app_config = AppConfig::from_env();
|
||||
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()?;
|
||||
let poster_fetcher = poster_fetcher::create()?;
|
||||
let object_storage = object_storage::create()?;
|
||||
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_env() {
|
||||
let tmdb_client = match tmdb_enrichment::TmdbEnrichmentClient::from_config(&config.metadata) {
|
||||
Ok(client) => {
|
||||
tracing::info!("TMDb enrichment enabled");
|
||||
Some(Arc::new(client))
|
||||
@@ -43,8 +42,8 @@ async fn main() -> anyhow::Result<()> {
|
||||
.clone()
|
||||
.map(|c| c as Arc<dyn PersonEnrichmentClient>);
|
||||
|
||||
let db = composition::factory::build_database_adapters(&database_url, &instance).await?;
|
||||
let (event_publisher_arc, consumer_arc) = event_bus::create(&db.db_pool).await?;
|
||||
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),
|
||||
@@ -113,7 +112,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
let (enrichment_handler, person_enrichment_handler, enrichment_job): EnrichmentParts =
|
||||
match tmdb_client {
|
||||
Some(client) => {
|
||||
let image_fetcher = poster_fetcher::create_image_fetcher()?;
|
||||
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(),
|
||||
@@ -140,6 +139,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
// ── Image conversion ──────────────────────────────────────────────────────
|
||||
|
||||
let conversion = image_converter::build(
|
||||
&config.image_conversion,
|
||||
Arc::clone(&object_storage),
|
||||
image_ref_command,
|
||||
image_ref_query,
|
||||
@@ -257,6 +257,7 @@ async fn main() -> anyhow::Result<()> {
|
||||
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),
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user