refactor: drop postgres leftovers, narrow bootstrap api
This commit is contained in:
@@ -17,7 +17,7 @@ federation = [
|
||||
config = { workspace = true }
|
||||
domain = { workspace = true }
|
||||
application = { workspace = true }
|
||||
http-axum = { path = "../adapters/http-axum", default-features = false }
|
||||
http-axum = { workspace = true, default-features = false }
|
||||
auth = { workspace = true }
|
||||
metadata = { workspace = true }
|
||||
poster-fetcher = { workspace = true }
|
||||
|
||||
@@ -44,7 +44,7 @@ use application::wrapup::deps::{
|
||||
};
|
||||
use domain::value_objects::InstanceIdentity;
|
||||
|
||||
use crate::{DatabaseOutput, Repositories};
|
||||
use crate::{Repositories, database::DatabaseOutput};
|
||||
|
||||
/// The composition root's single assembly point: every one of the server-facing
|
||||
/// deps structs, built once from the already-constructed `Repositories` and
|
||||
|
||||
@@ -2,16 +2,11 @@ use std::sync::Arc;
|
||||
|
||||
use anyhow::Context;
|
||||
use domain::ports::{
|
||||
AuthService, ImageRefCommand, ImageRefQuery, LocalApContentQuery, MetadataClient,
|
||||
MovieDeduplicator, ObjectStorage, PasswordHasher, PosterFetcherClient,
|
||||
RefreshSessionRepository, UserProfileFieldsRepository, WatchEventCommand, WatchEventQuery,
|
||||
WebhookTokenRepository,
|
||||
AuthService, ImageRefCommand, ImageRefQuery, MetadataClient, MovieDeduplicator, ObjectStorage,
|
||||
PasswordHasher, PosterFetcherClient, RefreshSessionRepository, UserProfileFieldsRepository,
|
||||
WatchEventCommand, WatchEventQuery, WebhookTokenRepository,
|
||||
};
|
||||
|
||||
pub enum DbPool {
|
||||
Sqlite(sqlx::SqlitePool),
|
||||
}
|
||||
|
||||
pub struct DatabaseOutput {
|
||||
pub movie_command: Arc<dyn domain::ports::MovieCommand>,
|
||||
pub movie_query: Arc<dyn domain::ports::MovieQuery>,
|
||||
@@ -31,13 +26,15 @@ pub struct DatabaseOutput {
|
||||
pub search_port: Arc<dyn domain::ports::SearchPort>,
|
||||
pub search_command: Arc<dyn domain::ports::SearchCommand>,
|
||||
pub profile_fields: Arc<dyn UserProfileFieldsRepository>,
|
||||
pub ap_content: Arc<dyn LocalApContentQuery>,
|
||||
#[cfg(feature = "federation")]
|
||||
pub ap_content: Arc<dyn domain::ports::LocalApContentQuery>,
|
||||
pub wrapup_stats: Arc<dyn domain::ports::WrapUpStatsQuery>,
|
||||
pub wrapup_repo: Arc<dyn domain::ports::WrapUpRepository>,
|
||||
pub insights: Arc<dyn domain::ports::InsightsQuery>,
|
||||
pub goal_command: Arc<dyn domain::ports::GoalCommand>,
|
||||
pub goal_query: Arc<dyn domain::ports::GoalQuery>,
|
||||
pub user_settings: Arc<dyn domain::ports::UserSettingsRepository>,
|
||||
#[cfg(feature = "federation")]
|
||||
pub federation_settings: std::sync::Arc<dyn domain::ports::UserFederationSettingsQuery>,
|
||||
pub remote_goal: Arc<dyn domain::ports::RemoteGoalRepository>,
|
||||
pub refresh_session: Arc<dyn RefreshSessionRepository>,
|
||||
@@ -46,7 +43,7 @@ pub struct DatabaseOutput {
|
||||
pub image_ref_query: Arc<dyn ImageRefQuery>,
|
||||
pub follow_command: Arc<dyn domain::ports::FollowCommand>,
|
||||
pub follow_query: Arc<dyn domain::ports::FollowQuery>,
|
||||
pub db_pool: DbPool,
|
||||
pub db_pool: sqlx::SqlitePool,
|
||||
}
|
||||
|
||||
pub async fn build_database_adapters(
|
||||
@@ -105,6 +102,7 @@ fn assemble(
|
||||
search_port: sp,
|
||||
search_command: sc,
|
||||
profile_fields: pf,
|
||||
#[cfg(feature = "federation")]
|
||||
ap_content: w.ap_content,
|
||||
wrapup_stats: w.wrapup_stats,
|
||||
wrapup_repo: w.wrapup_repo,
|
||||
@@ -112,6 +110,7 @@ fn assemble(
|
||||
goal_command: w.goal_command,
|
||||
goal_query: w.goal_query,
|
||||
user_settings: w.user_settings,
|
||||
#[cfg(feature = "federation")]
|
||||
federation_settings: w.federation_settings,
|
||||
remote_goal: w.remote_goal,
|
||||
refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone())) as _,
|
||||
@@ -120,7 +119,7 @@ fn assemble(
|
||||
image_ref_query,
|
||||
follow_command: Arc::clone(&social) as _,
|
||||
follow_query: social as _,
|
||||
db_pool: DbPool::Sqlite(w.pool),
|
||||
db_pool: w.pool,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -148,14 +147,6 @@ pub fn build_object_storage(
|
||||
object_storage::create(config)
|
||||
}
|
||||
|
||||
pub fn build_profile_fields_repo(
|
||||
pool: &DbPool,
|
||||
) -> anyhow::Result<Arc<dyn UserProfileFieldsRepository>> {
|
||||
match pool {
|
||||
DbPool::Sqlite(pool) => Ok(sqlite::create_profile_fields_repo(pool.clone())),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn warn_if_database_backend_set() {
|
||||
if let Ok(v) = std::env::var("DATABASE_BACKEND")
|
||||
&& v != "sqlite"
|
||||
|
||||
@@ -4,8 +4,6 @@ use std::sync::Arc;
|
||||
use anyhow::Context;
|
||||
use domain::ports::{EventConsumer, EventPublisher};
|
||||
|
||||
use crate::database::DbPool;
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
pub enum EventBusBackend {
|
||||
Db,
|
||||
@@ -29,17 +27,13 @@ impl EventBusBackend {
|
||||
}
|
||||
|
||||
pub async fn create_publisher(
|
||||
db_pool: &DbPool,
|
||||
db_pool: &sqlx::SqlitePool,
|
||||
config: &config::AppConfig,
|
||||
) -> anyhow::Result<Arc<dyn EventPublisher>> {
|
||||
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?
|
||||
}
|
||||
})
|
||||
Ok(sqlite::SqliteEventQueue::create_publisher(db_pool.clone(), &config.events).await?)
|
||||
}
|
||||
#[cfg(feature = "nats")]
|
||||
EventBusBackend::Nats => {
|
||||
@@ -52,19 +46,13 @@ pub async fn create_publisher(
|
||||
}
|
||||
|
||||
pub async fn create_channel(
|
||||
db_pool: &DbPool,
|
||||
db_pool: &sqlx::SqlitePool,
|
||||
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?),
|
||||
}
|
||||
Ok(sqlite::SqliteEventQueue::create_channel(db_pool.clone(), &config.events).await?)
|
||||
}
|
||||
#[cfg(feature = "nats")]
|
||||
EventBusBackend::Nats => {
|
||||
|
||||
@@ -64,8 +64,7 @@ pub async fn wire_server_social(
|
||||
config: &config::AppConfig,
|
||||
event_publisher: &Arc<dyn EventPublisher>,
|
||||
) -> anyhow::Result<ServerSocial> {
|
||||
let crate::database::DbPool::Sqlite(pool) = &db.db_pool;
|
||||
let fed_repos = sqlite_federation::wire(pool.clone(), instance.clone());
|
||||
let fed_repos = sqlite_federation::wire(db.db_pool.clone(), instance.clone());
|
||||
|
||||
let ap = activitypub::wire(activitypub::ActivityPubDeps {
|
||||
activity_repo: fed_repos.activity,
|
||||
@@ -117,7 +116,7 @@ pub async fn wire_server_social(
|
||||
remote_watchlist: fed_repos.remote_watchlist,
|
||||
federation_admin: fed_repos.admin_query,
|
||||
federated_profile: Some(sqlite::create_federated_profile_query(
|
||||
pool.clone(),
|
||||
db.db_pool.clone(),
|
||||
instance.clone(),
|
||||
)),
|
||||
})
|
||||
@@ -132,8 +131,7 @@ pub async fn wire_worker_handlers(
|
||||
config: &config::AppConfig,
|
||||
event_publisher: &Arc<dyn EventPublisher>,
|
||||
) -> anyhow::Result<Vec<Arc<dyn domain::ports::EventHandler>>> {
|
||||
let crate::database::DbPool::Sqlite(pool) = &db.db_pool;
|
||||
let fed_repos = sqlite_federation::wire(pool.clone(), instance.clone());
|
||||
let fed_repos = sqlite_federation::wire(db.db_pool.clone(), instance.clone());
|
||||
|
||||
let ap_wire = activitypub::wire(activitypub::ActivityPubDeps {
|
||||
activity_repo: fed_repos.activity,
|
||||
|
||||
@@ -4,8 +4,8 @@
|
||||
//! [`build_worker`]; everything between configuration and a running service is
|
||||
//! assembled here (ADR-0005, ADR-0013).
|
||||
|
||||
pub mod build;
|
||||
pub mod database;
|
||||
mod build;
|
||||
mod database;
|
||||
mod event_bus;
|
||||
mod federation;
|
||||
#[cfg(feature = "federation")]
|
||||
@@ -15,9 +15,7 @@ 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 build::build_deps;
|
||||
pub use repositories::Repositories;
|
||||
pub use server::{ServerApp, build_server, build_server_on};
|
||||
pub use tracing_setup::init_tracing;
|
||||
|
||||
@@ -66,7 +66,7 @@ async fn assemble(
|
||||
event_publisher: Arc::clone(&event_publisher_arc),
|
||||
person_enrichment,
|
||||
};
|
||||
let worker_deps = crate::build_worker_deps(&db, &worker_services);
|
||||
let worker_deps = crate::build::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);
|
||||
|
||||
103
crates/bootstrap/tests/config_propagation_test.rs
Normal file
103
crates/bootstrap/tests/config_propagation_test.rs
Normal file
@@ -0,0 +1,103 @@
|
||||
//! Configuration values reach the running app through the composition root.
|
||||
|
||||
mod support;
|
||||
|
||||
use axum::{
|
||||
body::Body,
|
||||
http::{Method, Request, header},
|
||||
};
|
||||
|
||||
async fn server_with(config: config::AppConfig) -> axum::Router {
|
||||
bootstrap::build_server_on(&config, support::memory_pool().await)
|
||||
.await
|
||||
.expect("server builds")
|
||||
}
|
||||
|
||||
fn set_cookies(headers: &axum::http::HeaderMap) -> Vec<String> {
|
||||
headers
|
||||
.get_all(header::SET_COOKIE)
|
||||
.iter()
|
||||
.map(|v| v.to_str().unwrap().to_string())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn registration_is_refused_when_allow_registration_is_false() {
|
||||
let mut config = support::test_config();
|
||||
config.server.allow_registration = false;
|
||||
let router = server_with(config).await;
|
||||
|
||||
let form = Request::builder()
|
||||
.method(Method::POST)
|
||||
.uri("/register")
|
||||
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
|
||||
.body(Body::from(format!(
|
||||
"email=a%40example.com&username=alice&password={}&_csrf=x",
|
||||
support::PASSWORD
|
||||
)))
|
||||
.unwrap();
|
||||
let (status, headers, _) = support::send(&router, form).await;
|
||||
assert_eq!(status, 303, "classic form redirects away");
|
||||
assert_eq!(headers.get(header::LOCATION).unwrap(), "/");
|
||||
assert!(
|
||||
support::cookie_value(&headers, "token").is_none(),
|
||||
"no session issued"
|
||||
);
|
||||
|
||||
let (status, _, _) = support::send(
|
||||
&router,
|
||||
support::json_request(
|
||||
Method::POST,
|
||||
"/api/v1/auth/register",
|
||||
None,
|
||||
serde_json::json!({
|
||||
"email": "a@example.com",
|
||||
"username": "alice",
|
||||
"password": support::PASSWORD,
|
||||
}),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, 401, "JSON register is refused");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn secure_cookies_setting_marks_cookies_secure() {
|
||||
for secure in [true, false] {
|
||||
let mut config = support::test_config();
|
||||
config.server.secure_cookies = secure;
|
||||
let router = server_with(config).await;
|
||||
|
||||
let (status, headers, _) = support::send(&router, support::get("/register", None)).await;
|
||||
assert_eq!(status, 200);
|
||||
let cookies = set_cookies(&headers);
|
||||
assert!(!cookies.is_empty(), "csrf cookie is set");
|
||||
assert!(
|
||||
cookies.iter().all(|c| c.contains("Secure") == secure),
|
||||
"secure_cookies={secure}: {cookies:?}"
|
||||
);
|
||||
|
||||
let csrf = support::cookie_value(&headers, "csrf").expect("csrf cookie");
|
||||
let form = Request::builder()
|
||||
.method(Method::POST)
|
||||
.uri("/register")
|
||||
.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
|
||||
.header(header::COOKIE, format!("csrf={csrf}"))
|
||||
.body(Body::from(format!(
|
||||
"email=a%40example.com&username=alice&password={}&_csrf={csrf}",
|
||||
support::PASSWORD
|
||||
)))
|
||||
.unwrap();
|
||||
let (status, headers, _) = support::send(&router, form).await;
|
||||
assert_eq!(status, 303);
|
||||
let token_cookie = set_cookies(&headers)
|
||||
.into_iter()
|
||||
.find(|c| c.starts_with("token="))
|
||||
.expect("session cookie");
|
||||
assert_eq!(
|
||||
token_cookie.contains("Secure"),
|
||||
secure,
|
||||
"session cookie, secure_cookies={secure}: {token_cookie}"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -359,3 +359,49 @@ async fn admin_user_surfaces_a_repository_error_as_500() {
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
}
|
||||
|
||||
async fn path_id_handler(
|
||||
http_axum::extractors::PathId(id): http_axum::extractors::PathId<u32>,
|
||||
) -> String {
|
||||
id.to_string()
|
||||
}
|
||||
|
||||
async fn refusal_code(resp: axum::response::Response) -> String {
|
||||
let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
|
||||
.await
|
||||
.unwrap();
|
||||
let v: serde_json::Value = serde_json::from_slice(&bytes).expect("refusal body is JSON");
|
||||
v["code"].as_str().unwrap().to_string()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn path_extractor_on_a_route_without_params_is_refused_as_internal() {
|
||||
let app = Router::new().route("/no-params", get(path_id_handler));
|
||||
let resp = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/no-params")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
|
||||
assert_eq!(refusal_code(resp).await, "INTERNAL");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn path_extractor_with_an_unparseable_param_is_refused_as_invalid_input() {
|
||||
let app = Router::new().route("/items/{id}", get(path_id_handler));
|
||||
let resp = app
|
||||
.oneshot(
|
||||
Request::builder()
|
||||
.uri("/items/abc")
|
||||
.body(Body::empty())
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
|
||||
assert_eq!(refusal_code(resp).await, "INVALID_INPUT");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user