diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index 96fdd1f0..fa63712f 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -44,4 +44,4 @@ jobs: # Layering guards. These live in the Makefile and were previously local-only, # so a PR could go green in CI while violating them. - name: guards - run: make check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-sqlite-is-ap-free + run: make check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-binaries-are-thin check-sqlite-is-ap-free diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 91f1eef6..1716751e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -44,7 +44,7 @@ jobs: # Layering guards. These live in the Makefile and were previously local-only, # so a PR could go green in CI while violating them. - name: guards - run: make check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-sqlite-is-ap-free + run: make check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-binaries-are-thin check-sqlite-is-ap-free docker: name: Build & Push Docker Image diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 2aa9e755..47d7df88 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -44,7 +44,7 @@ The project follows hexagonal (ports & adapters) architecture. See `architecture **Key rules:** - Presentation handlers never touch repositories directly — all domain logic goes through use cases in the `application` crate - Application use cases return raw domain data — URL formatting, date display, and view model assembly belong in presentation mappers (`presentation/src/mappers/`) -- Use cases called from presentation handlers take a `&FooDeps` struct (registered in `application::Deps`, built by `composition::build_deps`) — never `&AppContext` itself, which `application` cannot even depend on. A few keep individual `Arc` params instead: `enrich_movie` and `request_enrichment` because they are called from adapters rather than handlers, and `diary::log_review` because its one extra dependency comes from `Services` rather than a repository. `wrapup::compute`, `import::cleanup` and `integrations::cleanup` also take individual params, but they are only ever called from jobs. See ADR-0007 for the exact list and why each is legitimate +- Use cases called from presentation handlers take a `&FooDeps` struct (registered in `application::Deps`, built by `bootstrap::build_deps`) — never `&AppContext` itself, which `application` cannot even depend on. A few keep individual `Arc` params instead: `enrich_movie` and `request_enrichment` because they are called from adapters rather than handlers, and `diary::log_review` because its one extra dependency comes from `Services` rather than a repository. `wrapup::compute`, `import::cleanup` and `integrations::cleanup` also take individual params, but they are only ever called from jobs. See ADR-0007 for the exact list and why each is legitimate ``` domain → pure types, traits (ports), zero deps diff --git a/Cargo.lock b/Cargo.lock index d4de1912..f6d8cc76 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -319,10 +319,10 @@ dependencies = [ "async-trait", "bytes", "chrono", + "config", "domain", "futures", "hex", - "infra-wiring", "rand 0.9.4", "serde_json", "sha2", @@ -905,6 +905,41 @@ dependencies = [ "piper", ] +[[package]] +name = "bootstrap" +version = "0.1.0" +dependencies = [ + "activitypub", + "anyhow", + "application", + "async-trait", + "auth", + "axum", + "chrono", + "config", + "domain", + "export", + "image-converter", + "importer", + "jellyfin", + "metadata", + "nats", + "object-storage", + "plex", + "poster-fetcher", + "poster-sync", + "presentation", + "rss 0.1.0", + "sqlite", + "sqlite-federation", + "sqlx", + "tmdb-enrichment", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "built" version = "0.7.7" @@ -1120,33 +1155,6 @@ dependencies = [ "static_assertions", ] -[[package]] -name = "composition" -version = "0.1.0" -dependencies = [ - "activitypub", - "anyhow", - "application", - "async-trait", - "auth", - "chrono", - "config", - "domain", - "infra-wiring", - "jellyfin", - "metadata", - "nats", - "object-storage", - "plex", - "poster-fetcher", - "sqlite", - "sqlite-federation", - "sqlx", - "tokio", - "tracing", - "uuid", -] - [[package]] name = "concurrent-queue" version = "2.5.0" @@ -2771,15 +2779,6 @@ dependencies = [ "cfb", ] -[[package]] -name = "infra-wiring" -version = "0.1.0" -dependencies = [ - "anyhow", - "config", - "sqlx", -] - [[package]] name = "inout" version = "0.1.4" @@ -4013,10 +4012,10 @@ dependencies = [ "async-trait", "axum", "axum-governor", + "bootstrap", "bytes", "chrono", "chrono-tz", - "composition", "domain", "dotenvy", "futures", @@ -4948,36 +4947,25 @@ dependencies = [ name = "server" version = "0.1.0" dependencies = [ - "activitypub", "anyhow", "application", "async-trait", - "auth", "axum", + "bootstrap", "bytes", - "composition", "config", "domain", "dotenvy", - "export", "futures", "http-body-util", - "importer", - "infra-wiring", - "metadata", - "nats", - "object-storage", - "poster-fetcher", "presentation", "rss 0.1.0", "serde_json", "sqlite", - "sqlite-federation", "sqlx", "tokio", "tower", "tracing", - "tracing-subscriber", "uuid", ] @@ -7182,31 +7170,12 @@ dependencies = [ name = "worker" version = "0.1.0" dependencies = [ - "activitypub", "anyhow", - "application", - "async-trait", - "auth", - "composition", + "bootstrap", "config", - "domain", "dotenvy", - "export", - "image-converter", - "importer", - "infra-wiring", - "metadata", - "nats", - "object-storage", - "poster-fetcher", - "poster-sync", - "sqlite", - "sqlite-federation", - "sqlx", - "tmdb-enrichment", "tokio", "tracing", - "tracing-subscriber", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 49ec33ba..68eec11e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,7 @@ members = [ "crates/adapters/tmdb-enrichment", "crates/adapters/image-converter", "crates/domain", - "crates/composition", + "crates/bootstrap", "crates/presentation", "crates/server", "crates/tui", @@ -27,7 +27,6 @@ members = [ "crates/adapters/importer", "crates/adapters/jellyfin", "crates/adapters/plex", - "crates/infra-wiring", "crates/config", ] resolver = "2" @@ -70,7 +69,7 @@ api-types = { path = "crates/api-types" } domain = { path = "crates/domain" } tmdb-enrichment = { path = "crates/adapters/tmdb-enrichment" } application = { path = "crates/application" } -composition = { path = "crates/composition" } +bootstrap = { path = "crates/bootstrap" } presentation = { path = "crates/presentation" } auth = { path = "crates/adapters/auth" } metadata = { path = "crates/adapters/metadata" } @@ -90,7 +89,6 @@ importer = { path = "crates/adapters/importer" } jellyfin = { path = "crates/adapters/jellyfin" } plex = { path = "crates/adapters/plex" } image-converter = { path = "crates/adapters/image-converter" } -infra-wiring = { path = "crates/infra-wiring" } config = { path = "crates/config" } [profile.dev] diff --git a/Dockerfile b/Dockerfile index a737d7be..27437609 100644 --- a/Dockerfile +++ b/Dockerfile @@ -39,13 +39,12 @@ COPY crates/api-types/Cargo.toml crates/api-types/Cargo.toml COPY crates/application/Cargo.toml crates/application/Cargo.toml COPY crates/adapters/tmdb-enrichment/Cargo.toml crates/adapters/tmdb-enrichment/Cargo.toml COPY crates/domain/Cargo.toml crates/domain/Cargo.toml -COPY crates/composition/Cargo.toml crates/composition/Cargo.toml +COPY crates/bootstrap/Cargo.toml crates/bootstrap/Cargo.toml COPY crates/presentation/Cargo.toml crates/presentation/Cargo.toml COPY crates/server/Cargo.toml crates/server/Cargo.toml COPY crates/tui/Cargo.toml crates/tui/Cargo.toml COPY crates/adapters/image-converter/Cargo.toml crates/adapters/image-converter/Cargo.toml COPY crates/worker/Cargo.toml crates/worker/Cargo.toml -COPY crates/infra-wiring/Cargo.toml crates/infra-wiring/Cargo.toml COPY crates/config/Cargo.toml crates/config/Cargo.toml # Stub every crate so cargo can resolve and fetch deps diff --git a/Makefile b/Makefile index 82b878ba..d5126704 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ .DEFAULT_GOAL := check -check: fmt-check clippy test check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-sqlite-is-ap-free +check: fmt-check clippy test check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-binaries-are-thin check-sqlite-is-ap-free @echo "✅ All checks passed" check-appcontext: @@ -47,16 +47,36 @@ check-presentation-adapters: check-federation-off-social: - @if grep -n "NoopSocialCommand\|NoopSocialQuery" crates/server/src/main.rs | grep -q .; then \ - echo "❌ server wires social noops — federation-off would report an empty follow graph:"; \ - grep -n "NoopSocialCommand\|NoopSocialQuery" crates/server/src/main.rs; \ + @if grep -n "NoopSocialCommand\|NoopSocialQuery" crates/bootstrap/src/federation.rs | grep -q .; then \ + echo "❌ bootstrap wires social noops — federation-off would report an empty follow graph:"; \ + grep -n "NoopSocialCommand\|NoopSocialQuery" crates/bootstrap/src/federation.rs; \ exit 1; \ fi - @if ! awk '/cfg\(not\(feature = "federation"\)\)/ && !started {started=1} started && /cfg\(feature = "federation"\)/ {exit} started {print}' crates/server/src/main.rs | grep -q "LocalSocialService"; then \ - echo "❌ LocalSocialService not wired in the federation-off region of crates/server/src/main.rs — federation-off has no social behavior wired at all, not even noops"; \ + @if ! awk '/cfg\(not\(feature = "federation"\)\)/ && !started {started=1} started && /cfg\(feature = "federation"\)/ {exit} started {print}' crates/bootstrap/src/federation.rs | grep -q "LocalSocialService"; then \ + echo "❌ LocalSocialService not wired in the federation-off region of crates/bootstrap/src/federation.rs — federation-off has no social behavior wired at all, not even noops"; \ exit 1; \ fi - @echo "✅ server wires real social behavior in every configuration" + @echo "✅ bootstrap wires real social behavior in every configuration" + + +BINARIES_ALLOWED_DEPS := bootstrap config tokio axum anyhow dotenvy tracing + +check-binaries-are-thin: + @bad=""; \ + for c in server worker; do \ + deps=$$(awk '/^\[dependencies\]/{f=1;next} /^\[/{f=0} f' crates/$$c/Cargo.toml \ + | sed -n 's/^\([A-Za-z0-9_-]\{1,\}\)[[:space:]]*=.*/\1/p'); \ + for d in $$deps; do \ + case " $(BINARIES_ALLOWED_DEPS) " in *" $$d "*) continue;; esac; \ + bad="$$bad $$c:$$d"; \ + done; \ + done; \ + if [ -n "$$bad" ]; then \ + echo "❌ binary [dependencies] outside {$(BINARIES_ALLOWED_DEPS)}:$$bad"; \ + echo " wiring belongs in crates/bootstrap — see ADR-0013."; \ + exit 1; \ + fi; \ + echo "✅ server and worker depend only on {$(BINARIES_ALLOWED_DEPS)}" check-sqlite-is-ap-free: @@ -106,4 +126,4 @@ dev-spa: spa-dev: @npm --prefix spa run dev -.PHONY: check fmt fmt-check clippy test fix dev dev-stop dev-worker dev-spa spa-dev check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-sqlite-is-ap-free +.PHONY: check fmt fmt-check clippy test fix dev dev-stop dev-worker dev-spa spa-dev check-appcontext check-handler-deps check-handler-repos check-presentation-adapters check-federation-off-social check-binaries-are-thin check-sqlite-is-ap-free diff --git a/README.md b/README.md index 9a316e2c..12be0709 100644 --- a/README.md +++ b/README.md @@ -91,12 +91,13 @@ Hexagonal (Ports & Adapters) with Domain-Driven Design: ``` api-types — shared REST API request/response DTOs (Serialize/Deserialize + utoipa schemas) + HtmlPageContext; used by presentation, tui, and template adapters -infra-wiring — shared infrastructure types (DbPool, EventBusBackend, AppConfig) used by both server and worker binaries +config — typed configuration (defaults, config.toml, legacy env names); loaded once by the binaries +bootstrap — composition root: the only crate that names adapters; builds the server router and the worker service domain — pure types and CQRS port traits (MovieCommand/MovieQuery, WatchEventCommand/WatchEventQuery, GoalCommand/GoalQuery, DiaryQuery, PersonCommand/PersonQuery, SearchCommand/SearchPort, SocialCommand/SocialQuery, ImageFetcher, RssFeedRenderer), no external deps except serde application — use cases (commands + queries), business logic orchestration; handlers delegate here for all domain logic; modules: auth, diary, goals, import, integrations, movies, person, search, social, users, watchlist, wrapup presentation — Axum HTTP router, OpenAPI spec assembly, Swagger UI + Scalar serving; library crate, no binary of its own -server — owns the HTTP binary, features (federation, nats), boots presentation's router -worker — standalone worker binary (event consumer, poster sync, federation) +server — thin HTTP binary (dotenv, config, bootstrap::build_server, axum::serve); features (federation, nats) forward to bootstrap +worker — thin worker binary (bootstrap::build_worker, run until ctrl-c) adapters/ auth — JWT issuance and validation (Argon2 passwords) sqlite — SQLite repositories, search (FTS5), social graph, durable event queue, connection factory + migrations diff --git a/architecture.mmd b/architecture.mmd index 3e6a8c04..4e7cc9bd 100644 --- a/architecture.mmd +++ b/architecture.mmd @@ -77,8 +77,8 @@ graph TB DTO["DTOs
MovieDto, ReviewDto,
FeedEntryDto, UserSummaryDto,
HtmlPageContext, ..."] end - subgraph InfraWiring["infra-wiring"] - IW["DbPool, EventBusBackend,
AppConfig
Shared infra types"] + subgraph Bootstrap["bootstrap"] + IW["build_server, build_worker,
DbPool, EventBusBackend
Composition root"] end subgraph Adapters["Adapters (implement Port Traits)"] diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index f25126a2..34bb266b 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -75,6 +75,13 @@ pub async fn wire(database_url: &str) -> anyhow::Result { .await .context("Failed to connect to SQLite database")?; + wire_pool(pool).await +} + +/// Migrates an already-connected pool and builds every repository on it. +pub async fn wire_pool(pool: SqlitePool) -> anyhow::Result { + use anyhow::Context; + migrate(&pool) .await .map_err(|e| anyhow::anyhow!("{e}")) diff --git a/crates/application/Cargo.toml b/crates/application/Cargo.toml index 2ff9fe03..f6366262 100644 --- a/crates/application/Cargo.toml +++ b/crates/application/Cargo.toml @@ -6,7 +6,7 @@ edition.workspace = true [dependencies] async-trait = { workspace = true } domain = { workspace = true } -infra-wiring = { workspace = true } +config = { workspace = true } uuid = { workspace = true } chrono = { workspace = true } tracing = { workspace = true } diff --git a/crates/application/src/config.rs b/crates/application/src/config.rs index 38eb0b15..101cad4b 100644 --- a/crates/application/src/config.rs +++ b/crates/application/src/config.rs @@ -1 +1,41 @@ -pub use infra_wiring::{AppConfig, WrapUpConfig}; +#[derive(Clone)] +pub struct AppConfig { + pub allow_registration: bool, + pub base_url: String, + pub rate_limit: u64, + pub refresh_ttl_seconds: u64, + pub secure_cookies: bool, + pub cors_origins: String, + pub wrapup: WrapUpConfig, +} + +#[derive(Clone)] +pub struct WrapUpConfig { + pub font_path: Option, + pub logo_path: Option, + pub bg_dir: Option, +} + +impl From<&::config::AppConfig> for AppConfig { + fn from(config: &::config::AppConfig) -> Self { + Self { + allow_registration: config.server.allow_registration, + base_url: config.server.base_url.clone(), + rate_limit: config.server.rate_limit, + refresh_ttl_seconds: config.auth.refresh_ttl_seconds, + secure_cookies: config.server.secure_cookies, + cors_origins: config.server.cors_origins.clone(), + wrapup: WrapUpConfig::from(&config.wrapup), + } + } +} + +impl From<&::config::WrapUpConfig> for WrapUpConfig { + fn from(config: &::config::WrapUpConfig) -> Self { + Self { + font_path: config.font_path.clone(), + logo_path: config.logo_path.clone(), + bg_dir: config.bg_dir.clone(), + } + } +} diff --git a/crates/application/src/deps.rs b/crates/application/src/deps.rs index ae943fd6..969e3f8c 100644 --- a/crates/application/src/deps.rs +++ b/crates/application/src/deps.rs @@ -153,10 +153,10 @@ pub struct WrapupGroup { /// Every deps struct a handler can need, built once by the composition root. /// Use cases still receive only their own narrow struct — nothing takes `&Deps`. /// -/// `composition::build_deps` is called only from `crates/presentation/src/main.rs`. +/// `bootstrap::build_deps` is called only from `bootstrap::build_server`. /// The worker-only groups this used to also carry now live in `WorkerDeps`, built by -/// `composition::build_worker_deps` and consumed by `crates/worker/src/main.rs` — -/// `crates/worker` no longer wires its own deps (its former `db.rs` is gone). Nothing +/// `bootstrap::build_worker_deps` and consumed by `bootstrap::build_worker` — +/// the worker binary no longer wires anything itself. Nothing /// in `Deps` below is worker-only anymore, except one field with no consumer anywhere /// in the workspace — see its comment. pub struct Deps { @@ -188,7 +188,7 @@ pub struct WorkerServices { pub person_enrichment: Option>, } -/// The deps structs the worker binary needs, built by `composition::build_worker_deps`. +/// The deps structs the worker binary needs, built by `bootstrap::build_worker_deps`. /// These are the five groups that moved out of `Deps` during worker unification — /// they have no consumer the server binary can ever reach. pub struct WorkerDeps { diff --git a/crates/composition/Cargo.toml b/crates/bootstrap/Cargo.toml similarity index 62% rename from crates/composition/Cargo.toml rename to crates/bootstrap/Cargo.toml index 292ae32d..3632f8cd 100644 --- a/crates/composition/Cargo.toml +++ b/crates/bootstrap/Cargo.toml @@ -1,36 +1,49 @@ [package] -name = "composition" +name = "bootstrap" version = "0.1.0" edition.workspace = true [features] default = [] -nats = ["dep:nats", "infra-wiring/nats"] -federation = ["application/federation", "dep:sqlite-federation", "dep:activitypub"] +nats = ["dep:nats"] +federation = [ + "application/federation", + "presentation/federation", + "dep:sqlite-federation", + "dep:activitypub", +] [dependencies] config = { workspace = true } domain = { workspace = true } application = { workspace = true } -infra-wiring = { workspace = true } +presentation = { path = "../presentation", default-features = false } auth = { workspace = true } metadata = { workspace = true } poster-fetcher = { workspace = true } object-storage = { workspace = true } +poster-sync = { workspace = true } +tmdb-enrichment = { workspace = true } +image-converter = { workspace = true } +export = { workspace = true } +importer = { workspace = true } +rss = { workspace = true } jellyfin = { workspace = true } plex = { workspace = true } -anyhow = { workspace = true } -tracing = { workspace = true } - sqlite = { workspace = true } +sqlx = { workspace = true } +anyhow = { workspace = true } +async-trait = { workspace = true } +axum = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } + nats = { workspace = true, optional = true } activitypub = { workspace = true, optional = true } sqlite-federation = { workspace = true, optional = true } [dev-dependencies] -async-trait = { workspace = true } chrono = { workspace = true } domain = { workspace = true, features = ["test-helpers"] } -sqlx = { workspace = true } -tokio = { workspace = true } uuid = { workspace = true } diff --git a/crates/composition/src/build.rs b/crates/bootstrap/src/build.rs similarity index 100% rename from crates/composition/src/build.rs rename to crates/bootstrap/src/build.rs diff --git a/crates/composition/src/factory.rs b/crates/bootstrap/src/database.rs similarity index 89% rename from crates/composition/src/factory.rs rename to crates/bootstrap/src/database.rs index 1a15c430..bad5da5e 100644 --- a/crates/composition/src/factory.rs +++ b/crates/bootstrap/src/database.rs @@ -8,7 +8,9 @@ use domain::ports::{ WebhookTokenRepository, }; -pub use infra_wiring::DbPool; +pub enum DbPool { + Sqlite(sqlx::SqlitePool), +} pub struct DatabaseOutput { pub movie_command: Arc, @@ -54,6 +56,25 @@ pub async fn build_database_adapters( let w = sqlite::wire(url) .await .context("SQLite connection failed")?; + assemble(w, instance) +} + +/// Like [`build_database_adapters`], on a pool the caller already connected +/// (migrations still run). Lets tests drive the real wiring on an in-memory pool. +pub async fn build_database_adapters_on( + pool: sqlx::SqlitePool, + instance: &domain::value_objects::InstanceIdentity, +) -> anyhow::Result { + let w = sqlite::wire_pool(pool) + .await + .context("SQLite wiring failed")?; + assemble(w, instance) +} + +fn assemble( + w: sqlite::SqliteWireOutput, + instance: &domain::value_objects::InstanceIdentity, +) -> anyhow::Result { let (image_ref_command, image_ref_query) = sqlite::create_image_ref(w.pool.clone()); let (pc, pq) = sqlite::create_person_adapter(w.pool.clone()); let (sc, sp) = sqlite::create_search_adapter(w.pool.clone()); diff --git a/crates/bootstrap/src/event_bus.rs b/crates/bootstrap/src/event_bus.rs new file mode 100644 index 00000000..b4ace78a --- /dev/null +++ b/crates/bootstrap/src/event_bus.rs @@ -0,0 +1,77 @@ +use std::sync::Arc; + +#[cfg(feature = "nats")] +use anyhow::Context; +use domain::ports::{EventConsumer, EventPublisher}; + +use crate::database::DbPool; + +#[derive(Clone, Copy)] +pub enum EventBusBackend { + Db, + #[cfg(feature = "nats")] + Nats, +} + +impl EventBusBackend { + pub fn from_config(config: &config::EventsConfig) -> anyhow::Result { + match config.bus_backend.as_str() { + "db" => Ok(Self::Db), + #[cfg(feature = "nats")] + "nats" => Ok(Self::Nats), + #[cfg(not(feature = "nats"))] + "nats" => { + anyhow::bail!("EVENT_BUS_BACKEND=nats requires the nats feature to be compiled in") + } + other => anyhow::bail!("unknown EVENT_BUS_BACKEND={other}, expected 'db' or 'nats'"), + } + } +} + +pub async fn create_publisher( + db_pool: &DbPool, + config: &config::AppConfig, +) -> anyhow::Result> { + match EventBusBackend::from_config(&config.events)? { + EventBusBackend::Db => { + tracing::info!("event bus: DB queue"); + Ok(match db_pool { + DbPool::Sqlite(pool) => { + sqlite::SqliteEventQueue::create_publisher(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_publisher(cfg).await?) + } + } +} + +pub async fn create_channel( + db_pool: &DbPool, + config: &config::AppConfig, +) -> anyhow::Result<(Arc, Arc)> { + 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?) + } + } +} diff --git a/crates/bootstrap/src/federation.rs b/crates/bootstrap/src/federation.rs new file mode 100644 index 00000000..941b9c9d --- /dev/null +++ b/crates/bootstrap/src/federation.rs @@ -0,0 +1,170 @@ +//! Social and federation wiring for the server and worker. +//! +//! Federation off still wires real local social behaviour (`LocalSocialService`): +//! the follow graph and block queries must never silently report empty. + +use std::sync::Arc; + +use domain::ports::{ + BlockQuery, EventPublisher, FederatedProfileQuery, FederationAdminQuery, FollowGraphQuery, + RemoteWatchlistRepository, SocialCommand, +}; +use domain::value_objects::InstanceIdentity; + +use crate::database::DatabaseOutput; + +pub struct ServerSocial { + pub ap_router: axum::Router, + #[cfg(feature = "federation")] + pub ap_document: Arc, + #[cfg(feature = "federation")] + pub ap_blocklist: Arc, + pub social_command: Arc, + pub follow_graph: Arc, + pub block_query: Arc, + pub remote_watchlist: Arc, + pub federation_admin: Arc, + pub federated_profile: Option>, +} + +#[cfg(not(feature = "federation"))] +pub async fn wire_server_social( + db: &DatabaseOutput, + instance: &InstanceIdentity, + _config: &config::AppConfig, + _event_publisher: &Arc, +) -> anyhow::Result { + let local = Arc::new(application::social::local_service::LocalSocialService::new( + Arc::clone(&db.user), + Arc::clone(&db.follow_command), + Arc::clone(&db.follow_query), + instance.clone(), + )); + Ok(ServerSocial { + ap_router: axum::Router::new(), + social_command: Arc::clone(&local) as Arc, + follow_graph: Arc::clone(&local) as Arc, + block_query: local as Arc, + remote_watchlist: Arc::new(domain::ports::noop::NoopRemoteWatchlistRepository), + // The empty list this noop returns is an invariant, not an approximation: + // federation-off, `LocalSocialService::follow_resolved` hard-errors on any + // non-Local target and block/unblock always error, so no remote follow can + // ever be persisted through this wiring. (Stale rows from an instance + // previously built federation-ON are the one exception — display staleness + // only; see ADR-0009.) + federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery), + federated_profile: None, + }) +} + +#[cfg(feature = "federation")] +pub async fn wire_server_social( + db: &DatabaseOutput, + instance: &InstanceIdentity, + config: &config::AppConfig, + event_publisher: &Arc, +) -> anyhow::Result { + let crate::database::DbPool::Sqlite(pool) = &db.db_pool; + let fed_repos = sqlite_federation::wire(pool.clone(), instance.clone()); + + let ap = 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.clone(), + remote_goal_repo: Arc::clone(&db.remote_goal), + local_ap_content: Arc::clone(&db.ap_content), + movie_repo: Arc::clone(&db.movie_query), + review_repo: Arc::clone(&db.review), + diary_repo: Arc::clone(&db.diary), + goal_repo: Arc::clone(&db.goal_query), + stats_repo: Arc::clone(&db.stats), + user_repo: Arc::clone(&db.user), + federation_settings: Arc::clone(&db.federation_settings), + follow_command: Arc::clone(&fed_repos.follow_command), + follow_query: Arc::clone(&fed_repos.follow_query), + instance: instance.clone(), + allow_registration: config.server.allow_registration, + federation: config.federation.clone(), + event_publisher: Arc::clone(event_publisher), + }) + .await?; + + let local_social: Arc = + Arc::new(application::social::local_service::LocalSocialService::new( + Arc::clone(&db.user), + Arc::clone(&db.follow_command), + Arc::clone(&db.follow_query), + instance.clone(), + )); + + let composite_social = Arc::new(activitypub::CompositeSocialAdapter::new( + local_social, + ap.service, + Arc::clone(&db.user), + instance.clone(), + )); + + Ok(ServerSocial { + ap_router: ap.router, + ap_document: ap.document, + ap_blocklist: ap.blocklist, + social_command: composite_social.clone() as Arc, + follow_graph: composite_social.clone() as Arc, + block_query: composite_social as Arc, + remote_watchlist: fed_repos.remote_watchlist, + federation_admin: fed_repos.admin_query, + federated_profile: Some(sqlite::create_federated_profile_query( + pool.clone(), + instance.clone(), + )), + }) +} + +/// The worker's federation event handlers: the ActivityPub delivery handler and +/// the follow-accepted outbox backfill. +#[cfg(feature = "federation")] +pub async fn wire_worker_handlers( + db: &DatabaseOutput, + instance: &InstanceIdentity, + config: &config::AppConfig, + event_publisher: &Arc, +) -> anyhow::Result>> { + let crate::database::DbPool::Sqlite(pool) = &db.db_pool; + let fed_repos = sqlite_federation::wire(pool.clone(), instance.clone()); + + 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(&db.remote_goal), + local_ap_content: Arc::clone(&db.ap_content), + movie_repo: Arc::clone(&db.movie_query), + review_repo: Arc::clone(&db.review), + diary_repo: Arc::clone(&db.diary), + goal_repo: Arc::clone(&db.goal_query), + stats_repo: Arc::clone(&db.stats), + user_repo: Arc::clone(&db.user), + follow_command: fed_repos.follow_command, + follow_query: fed_repos.follow_query, + instance: instance.clone(), + allow_registration: config.server.allow_registration, + federation: config.federation.clone(), + event_publisher: Arc::clone(event_publisher), + federation_settings: Arc::clone(&db.federation_settings), + }) + .await?; + + tracing::info!("federation event handler registered"); + Ok(vec![ + ap_wire.event_handler, + Arc::new(crate::follow_backfill_handler::FollowBackfillHandler { + backfill: ap_wire.backfill, + }), + ]) +} diff --git a/crates/worker/src/follow_backfill_handler.rs b/crates/bootstrap/src/follow_backfill_handler.rs similarity index 100% rename from crates/worker/src/follow_backfill_handler.rs rename to crates/bootstrap/src/follow_backfill_handler.rs diff --git a/crates/bootstrap/src/lib.rs b/crates/bootstrap/src/lib.rs new file mode 100644 index 00000000..12533284 --- /dev/null +++ b/crates/bootstrap/src/lib.rs @@ -0,0 +1,28 @@ +//! The composition root: the only crate that names concrete adapters. +//! +//! `server` and `worker` are thin binaries over [`build_server`] and +//! [`build_worker`]; everything between configuration and a running service is +//! assembled here (ADR-0005, ADR-0013). + +pub mod build; +pub mod database; +mod event_bus; +mod federation; +#[cfg(feature = "federation")] +mod follow_backfill_handler; +pub mod repositories; +mod server; +mod tracing_setup; +mod worker; + +pub use build::{build_deps, build_worker_deps}; +pub use database::{DatabaseOutput, DbPool}; +pub use event_bus::EventBusBackend; +pub use repositories::Repositories; +pub use server::{ServerApp, build_server, build_server_on}; +pub use tracing_setup::init_tracing; +pub use worker::{WorkerApp, build_worker}; + +#[cfg(test)] +#[path = "tests/build.rs"] +mod build_tests; diff --git a/crates/composition/src/repositories.rs b/crates/bootstrap/src/repositories.rs similarity index 100% rename from crates/composition/src/repositories.rs rename to crates/bootstrap/src/repositories.rs diff --git a/crates/bootstrap/src/server.rs b/crates/bootstrap/src/server.rs new file mode 100644 index 00000000..f430fd7a --- /dev/null +++ b/crates/bootstrap/src/server.rs @@ -0,0 +1,147 @@ +use std::net::SocketAddr; +use std::sync::Arc; + +use anyhow::Context; +use application::config::AppConfig as AppSettings; +use domain::ports::{DiaryExporter, DocumentParser, EventPublisher}; +use domain::value_objects::InstanceIdentity; +use export::ExportAdapter; +use importer::ImporterDocumentParser; +use presentation::context::AppContext; +use presentation::{openapi, routes, state::AppState}; +use rss::RssAdapter; + +use crate::database::{self, DatabaseOutput}; +use crate::{Repositories, event_bus, federation}; + +pub struct ServerApp { + pub router: axum::Router, + pub addr: SocketAddr, +} + +/// Wires the whole HTTP application from configuration. +pub async fn build_server(config: &config::AppConfig) -> anyhow::Result { + database::warn_if_database_backend_set(); + let instance = InstanceIdentity::new(config.server.base_url.clone()); + let db = database::build_database_adapters(&config.database.url, &instance).await?; + let router = assemble(config, db, instance) + .await + .context("Failed to wire dependencies")?; + + // Resolved like `TcpListener::bind(&str)` would, so a hostname HOST keeps working. + let bind = format!("{}:{}", config.server.host, config.server.port); + let addr = tokio::net::lookup_host(&bind) + .await + .with_context(|| format!("Invalid HOST/PORT {bind}"))? + .next() + .with_context(|| format!("HOST/PORT {bind} resolved to no address"))?; + Ok(ServerApp { router, addr }) +} + +/// Builds the same router as [`build_server`] on a caller-provided pool. +/// Migrations run on it exactly as they do for the normal path. +pub async fn build_server_on( + config: &config::AppConfig, + pool: sqlx::SqlitePool, +) -> anyhow::Result { + let instance = InstanceIdentity::new(config.server.base_url.clone()); + let db = database::build_database_adapters_on(pool, &instance).await?; + assemble(config, db, instance).await +} + +async fn assemble( + config: &config::AppConfig, + db: DatabaseOutput, + instance: InstanceIdentity, +) -> anyhow::Result { + let app_config = AppSettings::from(config); + + let (auth_service, password_hasher) = database::build_auth_adapters(&config.auth)?; + let metadata_client = database::build_metadata_client(&config.metadata)?; + let poster_fetcher = database::build_poster_fetcher(&config.metadata)?; + let object_storage = database::build_object_storage(&config.storage)?; + + let event_publisher: Arc = + event_bus::create_publisher(&db.db_pool, config).await?; + + let social = federation::wire_server_social(&db, &instance, config, &event_publisher).await?; + + let review_logger = Arc::new(application::diary::review_logger::DefaultReviewLogger::new( + Arc::clone(&db.movie_command), + Arc::clone(&db.movie_query), + Arc::clone(&db.review), + Arc::clone(&db.watchlist), + Arc::clone(&metadata_client), + Arc::clone(&event_publisher), + )); + + let repos = Repositories { + movie_command: db.movie_command, + movie_query: db.movie_query, + review: db.review, + diary: db.diary, + stats: db.stats, + user: db.user, + import_session: db.import_session, + import_profile: db.import_profile, + movie_profile: db.movie_profile, + watchlist: db.watchlist, + watch_event_command: db.watch_event_command, + watch_event_query: db.watch_event_query, + webhook_token: db.webhook_token, + person_command: db.person_command, + person_query: db.person_query, + search_port: db.search_port, + search_command: db.search_command, + profile_fields: db.profile_fields, + remote_watchlist: social.remote_watchlist, + social_command: social.social_command, + follow_graph: social.follow_graph, + block_query: social.block_query, + federation_admin: social.federation_admin, + wrapup_stats: db.wrapup_stats, + wrapup_repo: db.wrapup_repo, + insights: db.insights, + goal_command: db.goal_command, + goal_query: db.goal_query, + user_settings: db.user_settings, + remote_goal: db.remote_goal, + refresh_session: db.refresh_session, + federated_profile: social.federated_profile, + }; + + let services = application::Services { + auth: auth_service, + password_hasher, + metadata: metadata_client, + poster_fetcher, + object_storage, + event_publisher, + diary_exporter: Arc::new(ExportAdapter) as Arc, + document_parser: Arc::new(ImporterDocumentParser) as Arc, + review_logger, + person_enrichment: None, + }; + + let deps = Arc::new(crate::build_deps(&repos, &services, &app_config, &instance)); + + let app_ctx = AppContext { + deps, + services, + config: app_config, + instance, + #[cfg(feature = "federation")] + ap_document: social.ap_document, + #[cfg(feature = "federation")] + ap_blocklist: social.ap_blocklist, + }; + + let state = AppState { + app_ctx, + rss_renderer: Arc::new(RssAdapter::new(config.server.base_url.clone())), + }; + Ok(openapi::serve(routes::build_router( + state, + social.ap_router, + ))) +} diff --git a/crates/composition/src/tests/build.rs b/crates/bootstrap/src/tests/build.rs similarity index 99% rename from crates/composition/src/tests/build.rs rename to crates/bootstrap/src/tests/build.rs index 4cfc62d3..c3c506f9 100644 --- a/crates/composition/src/tests/build.rs +++ b/crates/bootstrap/src/tests/build.rs @@ -266,11 +266,11 @@ fn test_repositories() -> Repositories { /// A lazily-connected in-memory sqlite pool — never actually opened in this test, /// just a value to satisfy `DatabaseOutput::db_pool`. -fn test_db_pool() -> infra_wiring::DbPool { +fn test_db_pool() -> crate::DbPool { let pool = sqlx::sqlite::SqlitePoolOptions::new() .connect_lazy("sqlite::memory:") .expect("lazy sqlite pool"); - infra_wiring::DbPool::Sqlite(pool) + crate::DbPool::Sqlite(pool) } fn test_database_output() -> DatabaseOutput { @@ -348,7 +348,7 @@ fn test_config() -> AppConfig { refresh_ttl_seconds: 2_592_000, secure_cookies: false, cors_origins: String::new(), - wrapup: infra_wiring::WrapUpConfig { + wrapup: application::config::WrapUpConfig { font_path: None, logo_path: None, bg_dir: None, diff --git a/crates/bootstrap/src/tracing_setup.rs b/crates/bootstrap/src/tracing_setup.rs new file mode 100644 index 00000000..786a3e4d --- /dev/null +++ b/crates/bootstrap/src/tracing_setup.rs @@ -0,0 +1,25 @@ +use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; + +/// Installs the global subscriber. `RUST_LOG` overrides `default_directives`. +pub fn init_tracing(default_directives: &str) { + tracing_subscriber::registry() + .with(log_filter(default_directives)) + .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}")) + } +} diff --git a/crates/bootstrap/src/worker.rs b/crates/bootstrap/src/worker.rs new file mode 100644 index 00000000..5bcf7e60 --- /dev/null +++ b/crates/bootstrap/src/worker.rs @@ -0,0 +1,231 @@ +use std::sync::Arc; + +use application::config::AppConfig as AppSettings; +use application::{ + MovieDiscoveryIndexer, SearchCleanupHandler, SearchReindexHandler, worker::WorkerService, +}; +use domain::ports::{EventHandler, MovieEnrichmentClient, PeriodicJob, PersonEnrichmentClient}; +use domain::value_objects::InstanceIdentity; + +use crate::{database, event_bus}; + +pub struct WorkerApp { + pub service: WorkerService, +} + +/// Wires the background worker: event handlers, periodic jobs (already +/// spawned) and the consumer. Run it with `service.run(shutdown)`. +pub async fn build_worker(config: &config::AppConfig) -> anyhow::Result { + database::warn_if_database_backend_set(); + let app_config = AppSettings::from(config); + let instance = 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> = tmdb_client + .clone() + .map(|c| c as Arc); + + let db = database::build_database_adapters(&config.database.url, &instance).await?; + let (event_publisher_arc, consumer_arc) = + event_bus::create_channel(&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 = crate::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 federation_handlers = + crate::federation::wire_worker_handlers(&db, &instance, config, &event_publisher_arc) + .await?; + + 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; + let refresh_session = db.refresh_session; + + let event_publisher = event_publisher_arc; + 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>, + Option>, + Option>, + ); + 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, + 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; + 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; + let job = Arc::new(application::jobs::EnrichmentStalenessJob::new( + Arc::clone(&movie_profile), + Arc::clone(&event_publisher), + )) as Arc; + (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> = 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> = { + 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; + + let cleanup = Arc::new(object_storage::ImageCleanupHandler::new(Arc::clone( + &object_storage, + ))) as Arc; + + let search_cleanup = Arc::new(SearchCleanupHandler::new( + Arc::clone(&search_command), + Arc::clone(&person_query), + )) as Arc; + + let discovery_indexer = Arc::new(MovieDiscoveryIndexer::new( + Arc::clone(&movie_query), + Arc::clone(&search_command), + )) as Arc; + + 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; + + let reindex_handler = Arc::new(SearchReindexHandler::new(worker_deps.reindex_search)) + as Arc; + + let mut h = vec![ + poster, + cleanup, + search_cleanup, + discovery_indexer, + wrapup_handler, + reindex_handler, + ]; + + #[cfg(feature = "federation")] + h.extend(federation_handlers); + + 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 + }; + + Ok(WorkerApp { + service: WorkerService::new(consumer_arc, handlers), + }) +} diff --git a/crates/composition/src/lib.rs b/crates/composition/src/lib.rs deleted file mode 100644 index 88e56cb4..00000000 --- a/crates/composition/src/lib.rs +++ /dev/null @@ -1,11 +0,0 @@ -pub mod build; -pub mod factory; -pub mod repositories; - -pub use build::{build_deps, build_worker_deps}; -pub use factory::{DatabaseOutput, DbPool}; -pub use repositories::Repositories; - -#[cfg(test)] -#[path = "tests/build.rs"] -mod build_tests; diff --git a/crates/infra-wiring/Cargo.toml b/crates/infra-wiring/Cargo.toml deleted file mode 100644 index 8a96e9b5..00000000 --- a/crates/infra-wiring/Cargo.toml +++ /dev/null @@ -1,13 +0,0 @@ -[package] -name = "infra-wiring" -version = "0.1.0" -edition = "2021" - -[dependencies] -config = { workspace = true } -sqlx = { workspace = true } -anyhow = { workspace = true } - -[features] -default = [] -nats = [] diff --git a/crates/infra-wiring/src/config.rs b/crates/infra-wiring/src/config.rs deleted file mode 100644 index 101cad4b..00000000 --- a/crates/infra-wiring/src/config.rs +++ /dev/null @@ -1,41 +0,0 @@ -#[derive(Clone)] -pub struct AppConfig { - pub allow_registration: bool, - pub base_url: String, - pub rate_limit: u64, - pub refresh_ttl_seconds: u64, - pub secure_cookies: bool, - pub cors_origins: String, - pub wrapup: WrapUpConfig, -} - -#[derive(Clone)] -pub struct WrapUpConfig { - pub font_path: Option, - pub logo_path: Option, - pub bg_dir: Option, -} - -impl From<&::config::AppConfig> for AppConfig { - fn from(config: &::config::AppConfig) -> Self { - Self { - allow_registration: config.server.allow_registration, - base_url: config.server.base_url.clone(), - rate_limit: config.server.rate_limit, - refresh_ttl_seconds: config.auth.refresh_ttl_seconds, - secure_cookies: config.server.secure_cookies, - cors_origins: config.server.cors_origins.clone(), - wrapup: WrapUpConfig::from(&config.wrapup), - } - } -} - -impl From<&::config::WrapUpConfig> for WrapUpConfig { - fn from(config: &::config::WrapUpConfig) -> Self { - Self { - font_path: config.font_path.clone(), - logo_path: config.logo_path.clone(), - bg_dir: config.bg_dir.clone(), - } - } -} diff --git a/crates/infra-wiring/src/lib.rs b/crates/infra-wiring/src/lib.rs deleted file mode 100644 index a0205078..00000000 --- a/crates/infra-wiring/src/lib.rs +++ /dev/null @@ -1,29 +0,0 @@ -pub mod config; - -pub use config::{AppConfig, WrapUpConfig}; - -pub enum DbPool { - Sqlite(sqlx::SqlitePool), -} - -#[derive(Clone, Copy)] -pub enum EventBusBackend { - Db, - #[cfg(feature = "nats")] - Nats, -} - -impl EventBusBackend { - pub fn from_config(config: &::config::EventsConfig) -> anyhow::Result { - match config.bus_backend.as_str() { - "db" => Ok(Self::Db), - #[cfg(feature = "nats")] - "nats" => Ok(Self::Nats), - #[cfg(not(feature = "nats"))] - "nats" => { - anyhow::bail!("EVENT_BUS_BACKEND=nats requires the nats feature to be compiled in") - } - other => anyhow::bail!("unknown EVENT_BUS_BACKEND={other}, expected 'db' or 'nats'"), - } - } -} diff --git a/crates/presentation/Cargo.toml b/crates/presentation/Cargo.toml index cc0952f0..bd25a45a 100644 --- a/crates/presentation/Cargo.toml +++ b/crates/presentation/Cargo.toml @@ -43,4 +43,4 @@ bytes = { workspace = true } futures = { workspace = true } tower = { version = "0.5", features = ["util"] } domain = { workspace = true, features = ["test-helpers"] } -composition = { workspace = true } +bootstrap = { workspace = true } diff --git a/crates/presentation/src/tests/extractors.rs b/crates/presentation/src/tests/extractors.rs index 152c72f3..71dd3670 100644 --- a/crates/presentation/src/tests/extractors.rs +++ b/crates/presentation/src/tests/extractors.rs @@ -7,7 +7,7 @@ use axum::{ http::{Request, StatusCode}, routing::get, }; -use composition::Repositories; +use bootstrap::Repositories; use domain::{ errors::DomainError, events::DomainEvent, @@ -816,7 +816,7 @@ pub(crate) fn test_repositories() -> Repositories { // --- Single state factory — builds `deps` from whatever `repos` it is given --- -/// Builds an `AppState` from caller-supplied `repos`, running `composition::build_deps` +/// Builds an `AppState` from caller-supplied `repos`, running `bootstrap::build_deps` /// AFTER any overrides the caller already applied to `repos`. This is the only place /// `deps` gets built for tests, so `repos` and `deps` can never disagree. pub(crate) fn build_test_state_from( @@ -854,9 +854,7 @@ pub(crate) fn build_test_state_from( let instance = domain::value_objects::InstanceIdentity::new("http://localhost:3000"); - let deps = Arc::new(composition::build_deps( - &repos, &services, &config, &instance, - )); + let deps = Arc::new(bootstrap::build_deps(&repos, &services, &config, &instance)); crate::state::AppState { app_ctx: AppContext { diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 3004e56b..eff91cd0 100644 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -7,54 +7,32 @@ license = "MIT" [features] default = ["federation"] -federation = [ - "dep:activitypub", - "dep:sqlite-federation", - "application/federation", - "presentation/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 } axum = { workspace = true } tokio = { workspace = true } anyhow = { workspace = true } -tracing = { workspace = true } -tracing-subscriber = { workspace = true } dotenvy = { workspace = true } - -domain = { workspace = true } -application = { workspace = true } -composition = { workspace = true } -presentation = { path = "../presentation", default-features = false } -auth = { workspace = true } -metadata = { workspace = true } -poster-fetcher = { workspace = true } -object-storage = { workspace = true } -rss = { workspace = true } -export = { workspace = true } -importer = { 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 } [dev-dependencies] +domain = { workspace = true, features = ["test-helpers"] } +application = { workspace = true } +presentation = { path = "../presentation", default-features = false } +rss = { workspace = true } +sqlite = { workspace = true } +sqlx = { workspace = true } +async-trait = { workspace = true } bytes = { workspace = true } futures = { workspace = true } serde_json = { workspace = true } uuid = { workspace = true } tower = { version = "0.5", features = ["util"] } http-body-util = "0.1" -domain = { workspace = true, features = ["test-helpers"] } diff --git a/crates/server/src/main.rs b/crates/server/src/main.rs index fc39b4bc..09fe19f2 100644 --- a/crates/server/src/main.rs +++ b/crates/server/src/main.rs @@ -1,305 +1,22 @@ -use std::sync::Arc; - use anyhow::Context; - use tokio::net::TcpListener; -use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; - -use composition::Repositories; -use export::ExportAdapter; -use importer::ImporterDocumentParser; -use presentation::context::AppContext; -use presentation::{openapi, routes, state::AppState}; -use rss::RssAdapter; - -use domain::ports::{DiaryExporter, DocumentParser, EventPublisher}; -use infra_wiring::EventBusBackend; #[tokio::main] async fn main() -> anyhow::Result<()> { dotenvy::dotenv().ok(); - init_tracing(); + bootstrap::init_tracing("server=debug,tower_http=debug"); let config = config::load().context("Failed to load configuration")?; + let app = bootstrap::build_server(&config).await?; - let (state, ap_router) = wire_dependencies(&config) - .await - .context("Failed to wire dependencies")?; - - let app = openapi::serve(routes::build_router(state, ap_router)); - - let addr = format!("{}:{}", config.server.host, config.server.port); - let listener = TcpListener::bind(&addr).await?; - tracing::info!("Listening on {}", addr); + let listener = TcpListener::bind(app.addr).await?; + tracing::info!("Listening on {}", app.addr); axum::serve( listener, - app.into_make_service_with_connect_info::(), + app.router + .into_make_service_with_connect_info::(), ) .await?; Ok(()) } - -async fn wire_dependencies(config: &config::AppConfig) -> anyhow::Result<(AppState, axum::Router)> { - let app_config = application::config::AppConfig::from(config); - let instance = domain::value_objects::InstanceIdentity::new(app_config.base_url.clone()); - let database_url = config.database.url.as_str(); - composition::factory::warn_if_database_backend_set(); - - let (auth_service, password_hasher) = composition::factory::build_auth_adapters(&config.auth)?; - let metadata_client = composition::factory::build_metadata_client(&config.metadata)?; - let poster_fetcher = composition::factory::build_poster_fetcher(&config.metadata)?; - let object_storage = composition::factory::build_object_storage(&config.storage)?; - - let db = composition::factory::build_database_adapters(database_url, &instance).await?; - #[cfg(feature = "federation")] - let ap_content_repo = db.ap_content; - let db_pool = db.db_pool; - - // Wire up event channel, federation service, and ap_router - let event_bus = EventBusBackend::from_config(&config.events)?; - - #[cfg(feature = "federation")] - let ( - event_publisher_arc, - ap_router, - ap_document, - ap_blocklist, - social_query, - remote_watchlist_repo, - social_command_arc, - follow_graph_arc, - block_query_arc, - ) = { - let fed_repos = match &db_pool { - composition::DbPool::Sqlite(pool) => { - sqlite_federation::wire(pool.clone(), instance.clone()) - } - }; - - let ep = create_event_publisher(event_bus, &db_pool, config).await?; - - let ap = 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.clone(), - remote_goal_repo: Arc::clone(&db.remote_goal), - local_ap_content: Arc::clone(&ap_content_repo), - movie_repo: Arc::clone(&db.movie_query), - review_repo: Arc::clone(&db.review), - diary_repo: Arc::clone(&db.diary), - goal_repo: Arc::clone(&db.goal_query), - stats_repo: Arc::clone(&db.stats), - user_repo: Arc::clone(&db.user), - federation_settings: std::sync::Arc::clone(&db.federation_settings), - follow_command: Arc::clone(&fed_repos.follow_command), - follow_query: Arc::clone(&fed_repos.follow_query), - instance: instance.clone(), - allow_registration: app_config.allow_registration, - federation: config.federation.clone(), - event_publisher: Arc::clone(&ep), - }) - .await?; - let ap_router = ap.router; - let ap_document = ap.document; - let ap_blocklist = ap.blocklist; - - let local_social: Arc = - Arc::new(application::social::local_service::LocalSocialService::new( - Arc::clone(&db.user), - Arc::clone(&db.follow_command), - Arc::clone(&db.follow_query), - instance.clone(), - )); - - let composite_social = Arc::new(activitypub::CompositeSocialAdapter::new( - local_social, - ap.service, - Arc::clone(&db.user), - instance.clone(), - )); - - ( - ep, - ap_router, - ap_document, - ap_blocklist, - fed_repos.admin_query, - fed_repos.remote_watchlist, - composite_social.clone() as Arc, - composite_social.clone() as Arc, - composite_social as Arc, - ) - }; - - #[cfg(not(feature = "federation"))] - let event_publisher_arc = create_event_publisher(event_bus, &db_pool, config).await?; - #[cfg(not(feature = "federation"))] - let ap_router = axum::Router::new(); - #[cfg(not(feature = "federation"))] - let (social_command_arc, follow_graph_arc, block_query_arc) = { - let local = Arc::new(application::social::local_service::LocalSocialService::new( - Arc::clone(&db.user), - Arc::clone(&db.follow_command), - Arc::clone(&db.follow_query), - instance.clone(), - )); - ( - Arc::clone(&local) as Arc, - Arc::clone(&local) as Arc, - local as Arc, - ) - }; - - let review_logger = Arc::new(application::diary::review_logger::DefaultReviewLogger::new( - Arc::clone(&db.movie_command), - Arc::clone(&db.movie_query), - Arc::clone(&db.review), - Arc::clone(&db.watchlist), - Arc::clone(&metadata_client), - Arc::clone(&event_publisher_arc), - )); - - let repos = Repositories { - movie_command: db.movie_command, - movie_query: db.movie_query, - review: db.review, - diary: db.diary, - stats: db.stats, - user: db.user, - import_session: db.import_session, - import_profile: db.import_profile, - movie_profile: db.movie_profile, - watchlist: db.watchlist, - watch_event_command: db.watch_event_command, - watch_event_query: db.watch_event_query, - webhook_token: db.webhook_token, - person_command: db.person_command, - person_query: db.person_query, - search_port: db.search_port, - search_command: db.search_command, - profile_fields: db.profile_fields, - #[cfg(feature = "federation")] - remote_watchlist: remote_watchlist_repo, - #[cfg(not(feature = "federation"))] - remote_watchlist: Arc::new(domain::ports::noop::NoopRemoteWatchlistRepository), - social_command: social_command_arc, - follow_graph: follow_graph_arc, - block_query: block_query_arc, - #[cfg(feature = "federation")] - federation_admin: social_query.clone(), - // The empty list this noop returns is an invariant, not an approximation: - // federation-off, `LocalSocialService::follow_resolved` hard-errors on any - // non-Local target and block/unblock always error, so no remote follow can - // ever be persisted through this wiring. (Stale rows from an instance - // previously built federation-ON are the one exception — display staleness - // only; see ADR-0009.) - #[cfg(not(feature = "federation"))] - federation_admin: Arc::new(domain::ports::noop::NoopFederationAdminQuery), - wrapup_stats: db.wrapup_stats, - wrapup_repo: db.wrapup_repo, - insights: db.insights, - goal_command: db.goal_command, - goal_query: db.goal_query, - user_settings: db.user_settings, - remote_goal: db.remote_goal, - refresh_session: db.refresh_session, - #[cfg(feature = "federation")] - federated_profile: Some({ - match &db_pool { - composition::DbPool::Sqlite(pool) => { - sqlite::create_federated_profile_query(pool.clone(), instance.clone()) - } - } - }), - #[cfg(not(feature = "federation"))] - federated_profile: None, - }; - - let services = application::Services { - auth: auth_service, - password_hasher, - metadata: metadata_client, - poster_fetcher, - object_storage, - event_publisher: event_publisher_arc, - diary_exporter: Arc::new(ExportAdapter) as Arc, - document_parser: Arc::new(ImporterDocumentParser) as Arc, - review_logger, - person_enrichment: None, - }; - - let deps = Arc::new(composition::build_deps( - &repos, - &services, - &app_config, - &instance, - )); - - let app_ctx = AppContext { - deps, - services, - config: app_config, - instance, - #[cfg(feature = "federation")] - ap_document, - #[cfg(feature = "federation")] - ap_blocklist, - }; - - let state = AppState { - app_ctx, - rss_renderer: Arc::new(RssAdapter::new(config.server.base_url.clone())), - }; - Ok((state, ap_router)) -} - -async fn create_event_publisher( - event_bus: EventBusBackend, - db_pool: &composition::DbPool, - config: &config::AppConfig, -) -> anyhow::Result> { - match event_bus { - EventBusBackend::Db => { - tracing::info!("event bus: DB queue"); - Ok(match db_pool { - composition::DbPool::Sqlite(pool) => { - sqlite::SqliteEventQueue::create_publisher(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_publisher(cfg).await?) - } - } -} - -fn init_tracing() { - tracing_subscriber::registry() - .with(log_filter("server=debug,tower_http=debug")) - .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}")) - } -} diff --git a/crates/server/tests/api_test.rs b/crates/server/tests/api_test.rs index de982935..68c4c41e 100644 --- a/crates/server/tests/api_test.rs +++ b/crates/server/tests/api_test.rs @@ -7,7 +7,7 @@ use axum::{ body::Body, http::{Request, StatusCode}, }; -use composition::Repositories; +use bootstrap::Repositories; use domain::{ errors::DomainError, events::DomainEvent, @@ -461,9 +461,7 @@ async fn test_app_with(auth: impl AuthService + 'static) -> (Router, SqlitePool) let instance = domain::value_objects::InstanceIdentity::new("http://localhost:3000"); - let deps = Arc::new(composition::build_deps( - &repos, &services, &config, &instance, - )); + let deps = Arc::new(bootstrap::build_deps(&repos, &services, &config, &instance)); let state = AppState { app_ctx: AppContext { diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 6a1f7ab7..de32a83c 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -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 } diff --git a/crates/worker/src/event_bus.rs b/crates/worker/src/event_bus.rs deleted file mode 100644 index 244c5cd3..00000000 --- a/crates/worker/src/event_bus.rs +++ /dev/null @@ -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, Arc)> { - 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?) - } - } -} diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index f1027431..908c9623 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -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> = tmdb_client - .clone() - .map(|c| c as Arc); - - 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>, - Option>, - Option>, - ); - 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, - 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; - 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; - let job = Arc::new(application::jobs::EnrichmentStalenessJob::new( - Arc::clone(&movie_profile), - Arc::clone(&event_publisher), - )) as Arc; - (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> = 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> = { - 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; - - let cleanup = Arc::new(object_storage::ImageCleanupHandler::new(Arc::clone( - &object_storage, - ))) as Arc; - - let search_cleanup = Arc::new(SearchCleanupHandler::new( - Arc::clone(&search_command), - Arc::clone(&person_query), - )) as Arc; - - let discovery_indexer = Arc::new(MovieDiscoveryIndexer::new( - Arc::clone(&movie_query), - Arc::clone(&search_command), - )) as Arc; - - 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; - - let reindex_handler = Arc::new(SearchReindexHandler::new(worker_deps.reindex_search)) - as Arc; - - 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); - } - - 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}")) - } -} diff --git a/docs/adr/0005-single-composition-root.md b/docs/adr/0005-single-composition-root.md index c7fb962c..d76f1948 100644 --- a/docs/adr/0005-single-composition-root.md +++ b/docs/adr/0005-single-composition-root.md @@ -75,3 +75,10 @@ unconditionally, with no `#[cfg(feature = "federation")]` gate. `cargo build -p --no-default-features --features sqlite` (federation off) fails to compile because of it. This predates worker unification — confirmed byte-identical at the commit before this plan started — and is unrelated to the composition-root split; it needs its own fix. + +## Realised by `bootstrap` (ADR-0013) + +`crates/composition` is now `crates/bootstrap`, and `crates/infra-wiring` is absorbed into it. The +server and worker `main.rs` files no longer wire anything: `bootstrap::build_server` and +`bootstrap::build_worker` own the whole assembly, so the single composition root is now the only +place adapters are named. See ADR-0013. diff --git a/docs/adr/0013-config-and-bootstrap.md b/docs/adr/0013-config-and-bootstrap.md new file mode 100644 index 00000000..2af5842a --- /dev/null +++ b/docs/adr/0013-config-and-bootstrap.md @@ -0,0 +1,44 @@ +# Typed configuration crate and a bootstrap crate that owns all wiring + +Configuration was read ad hoc from `std::env` in many adapters, and the two binaries each carried +hundreds of lines of adapter wiring (`wire_dependencies`, `create_event_publisher`, the worker's +handler and job assembly). ADR-0005 made `composition` the single root for deps, but the binaries +still named adapters themselves, and `infra-wiring` existed only so `application` could re-export an +`AppConfig`. + +## Decision + +**`config` crate.** One typed `config::AppConfig` with sections (`server`, `database`, `auth`, +`events`, `nats`, `storage`, `image_conversion`, `metadata`, `federation`, `wrapup`), loaded once by +the binary via `config::load()`. Layering, lowest to highest precedence: + +1. built-in defaults; +2. `config.toml`, or the file named by `CONFIG_FILE` (a named file that is missing is an error); +3. environment variables, using the legacy names (`DATABASE_URL`, `PORT`, `JWT_SECRET`, ...) with no + prefix. Every variable that ever worked keeps its name, meaning and default. + +A malformed value fails startup with an error naming the variable (`PORT=abc`). An empty or +whitespace-only value is treated as unset, so the default applies (or `None` for optional fields) +rather than being an error. Adapters take their section (or a typed config built from it), never +raw environment strings. + +**`bootstrap` crate.** Replaces `composition` and `infra-wiring`. It is the only crate that names +concrete adapters. It exposes `build_server(&AppConfig) -> ServerApp { router, addr }`, +`build_worker(&AppConfig) -> WorkerApp { service }`, `build_server_on(&AppConfig, SqlitePool)` (the +same router on a caller-provided pool, for tests) and `init_tracing`. `DbPool` and +`EventBusBackend` live here. `application` keeps a small application-owned `AppConfig` / +`WrapUpConfig` built from the `config` sections, so use-case deps structs and their test fixtures +are unchanged, and `application` no longer pulls `sqlx` or `anyhow`. + +**Binaries are thin.** `server` and `worker` do dotenv, `config::load`, `init_tracing`, one +`bootstrap::build_*` call and `axum::serve` / `service.run`. Their `[dependencies]` are limited to +`bootstrap`, `config`, `tokio`, `axum`, `anyhow`, `dotenvy` and `tracing`; `make +check-binaries-are-thin` enforces it. The `federation` and `nats` features forward to bootstrap. + +## Consequences + +- Adding or changing an adapter touches `bootstrap` only; the binaries do not change. +- The wiring is reachable from tests (`build_server_on`) without a process or a port. +- `check-federation-off-social` now inspects `crates/bootstrap/src/federation.rs`: federation-off + must still wire `LocalSocialService`, never social noops (ADR-0009). +- Operators can move settings into `config.toml` incrementally; existing `.env` files keep working.