use std::{
env,
path::Path,
time::Duration,
};
use buck2_resources::Buck2Resources;
use derive_builder::Builder;
use serde::{
Deserialize,
Serialize,
};
use si_crypto::{
SymmetricCryptoServiceConfig,
SymmetricCryptoServiceConfigFile,
VeritechCryptoConfig,
};
use si_data_nats::NatsConfig;
use si_data_pg::PgPoolConfig;
use si_layer_cache::db::LayerDbConfig;
use si_service_endpoints::ServiceEndpointsConfig;
pub(crate) use si_settings::StandardConfig;
pub use si_settings::StandardConfigFile;
use si_std::CanonicalFileError;
use si_tls::CertificateSource;
use telemetry::prelude::*;
use thiserror::Error;
use ulid::Ulid;
const DEFAULT_CONCURRENCY_LIMIT: Option<usize> = None;
const DEFAULT_PARALLEL_BUILD_LIMIT: usize = 50;
const DEFAULT_QUIESCENT_PERIOD_SECS: u64 = 60 * 10;
const DEFAULT_QUIESCENT_PERIOD: Duration = Duration::from_secs(DEFAULT_QUIESCENT_PERIOD_SECS);
#[remain::sorted]
#[derive(Debug, Error)]
pub enum ConfigError {
#[error("config builder")]
Builder(#[from] ConfigBuilderError),
#[error("canonical file error: {0}")]
CanonicalFile(#[from] CanonicalFileError),
#[error("error configuring for development")]
Development(#[source] Box<dyn std::error::Error + 'static + Sync + Send>),
#[error("layer cache error: {0}")]
LayerCache(#[from] si_layer_cache::LayerDbError),
#[error("si settings error: {0}")]
SiSettings(#[from] si_settings::SettingsError),
}
impl ConfigError {
fn development(err: impl std::error::Error + 'static + Sync + Send) -> Self {
Self::Development(Box::new(err))
}
}
type Result<T> = std::result::Result<T, ConfigError>;
/// The config for the forklift server.
#[derive(Debug, Builder, Serialize)]
pub struct Config {
#[builder(default = "random_instance_id()")]
instance_id: String,
#[builder(default = "default_concurrency_limit()")]
concurrency_limit: Option<usize>,
#[builder(default = "default_parallel_build_limit()")]
parallel_build_limit: usize,
#[builder(default = "default_streaming_patches()")]
streaming_patches: bool,
#[builder(default = "PgPoolConfig::default()")]
pg_pool: PgPoolConfig,
#[builder(default = "NatsConfig::default()")]
nats: NatsConfig,
#[builder(default = "VeritechCryptoConfig::default()")]
crypto: VeritechCryptoConfig,
#[builder(default = "SymmetricCryptoServiceConfig::default()")]
symmetric_crypto_service: SymmetricCryptoServiceConfig,
#[builder(default = "default_layer_db_config()")]
layer_db_config: LayerDbConfig,
#[builder(default = "default_quiescent_period()")]
quiescent_period: Duration,
#[builder(default = "default_service_endpoints_config()")]
service_endpoints: ServiceEndpointsConfig,
}
impl StandardConfig for Config {
type Builder = ConfigBuilder;
}
impl Config {
/// Gets the config's concurrency limit.
pub fn concurrency_limit(&self) -> Option<usize> {
self.concurrency_limit
}
/// Gets the config's parallel build limit.
pub fn parallel_build_limit(&self) -> usize {
self.parallel_build_limit
}
/// Gets whether edda should stream patches, or send as a single batch.
pub fn streaming_patches(&self) -> bool {
self.streaming_patches
}
/// Gets the config's instance ID.
pub fn instance_id(&self) -> &str {
self.instance_id.as_ref()
}
/// Gets a reference to the config's pg pool.
#[must_use]
pub fn pg_pool(&self) -> &PgPoolConfig {
&self.pg_pool
}
/// Gets a reference to the config's nats.
#[must_use]
pub fn nats(&self) -> &NatsConfig {
&self.nats
}
/// Gets a reference to the config's subject prefix.
pub fn subject_prefix(&self) -> Option<&str> {
self.nats.subject_prefix.as_deref()
}
/// Gets a reference to the config's crypto config.
#[must_use]
pub fn crypto(&self) -> &VeritechCryptoConfig {
&self.crypto
}
/// Gets a reference to the symmetric crypto service.
pub fn symmetric_crypto_service(&self) -> &SymmetricCryptoServiceConfig {
&self.symmetric_crypto_service
}
/// Gets a reference to the layderdb config
#[must_use]
pub fn layer_db_config(&self) -> &LayerDbConfig {
&self.layer_db_config
}
/// Gets the period of inactivity before a change set consuming stream will shut down
pub fn quiescent_period(&self) -> Duration {
self.quiescent_period
}
/// Gets a reference to the config's service endpoints configuration.
#[must_use]
pub fn service_endpoints(&self) -> &ServiceEndpointsConfig {
&self.service_endpoints
}
}
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct ConfigFile {
#[serde(default = "random_instance_id")]
instance_id: String,
// TODO(fnichol): this field is prefixed with `edda_` due to our current rendering of systemd
// units (i.e. our global `service.toml`) which will only render *one* concurrency limit for
// all service types
#[serde(default = "default_concurrency_limit")]
edda_concurrency_limit: Option<usize>,
#[serde(default = "default_parallel_build_limit")]
edda_parallel_build_limit: usize,
#[serde(default = "default_streaming_patches")]
streaming_patches: bool,
#[serde(default)]
pg: PgPoolConfig,
#[serde(default)]
nats: NatsConfig,
#[serde(default)]
crypto: VeritechCryptoConfig,
#[serde(default = "default_symmetric_crypto_config")]
symmetric_crypto_service: SymmetricCryptoServiceConfigFile,
#[serde(default = "default_layer_db_config")]
layer_db_config: LayerDbConfig,
#[serde(default = "default_quiescent_period_secs")]
quiescent_period_secs: u64,
#[serde(default = "default_service_endpoints_config")]
service_endpoints: ServiceEndpointsConfig,
}
impl Default for ConfigFile {
fn default() -> Self {
Self {
instance_id: random_instance_id(),
edda_concurrency_limit: default_concurrency_limit(),
edda_parallel_build_limit: default_parallel_build_limit(),
streaming_patches: default_streaming_patches(),
pg: Default::default(),
nats: Default::default(),
crypto: Default::default(),
symmetric_crypto_service: default_symmetric_crypto_config(),
layer_db_config: default_layer_db_config(),
quiescent_period_secs: default_quiescent_period_secs(),
service_endpoints: default_service_endpoints_config(),
}
}
}
impl StandardConfigFile for ConfigFile {
type Error = ConfigError;
}
impl TryFrom<ConfigFile> for Config {
type Error = ConfigError;
fn try_from(mut value: ConfigFile) -> Result<Self> {
detect_and_configure_development(&mut value)?;
let mut config = Config::builder();
config.pg_pool(value.pg);
config.nats(value.nats);
config.crypto(value.crypto);
config.symmetric_crypto_service(value.symmetric_crypto_service.try_into()?);
config.layer_db_config(value.layer_db_config);
config.concurrency_limit(value.edda_concurrency_limit);
config.streaming_patches(value.streaming_patches);
config.parallel_build_limit(value.edda_parallel_build_limit);
config.instance_id(value.instance_id);
config.quiescent_period(Duration::from_secs(value.quiescent_period_secs));
config.service_endpoints(value.service_endpoints);
config.build().map_err(Into::into)
}
}
fn random_instance_id() -> String {
Ulid::new().to_string()
}
fn default_concurrency_limit() -> Option<usize> {
DEFAULT_CONCURRENCY_LIMIT
}
fn default_parallel_build_limit() -> usize {
DEFAULT_PARALLEL_BUILD_LIMIT
}
fn default_streaming_patches() -> bool {
false
}
fn default_symmetric_crypto_config() -> SymmetricCryptoServiceConfigFile {
SymmetricCryptoServiceConfigFile {
active_key: None,
active_key_base64: None,
extra_keys: vec![],
}
}
fn default_layer_db_config() -> LayerDbConfig {
LayerDbConfig::default()
}
fn default_quiescent_period() -> Duration {
DEFAULT_QUIESCENT_PERIOD
}
fn default_quiescent_period_secs() -> u64 {
DEFAULT_QUIESCENT_PERIOD_SECS
}
fn default_service_endpoints_config() -> ServiceEndpointsConfig {
ServiceEndpointsConfig::new(0)
}
#[allow(clippy::disallowed_methods)] // Used to determine if running in development
pub fn detect_and_configure_development(config: &mut ConfigFile) -> Result<()> {
if env::var("BUCK_RUN_BUILD_ID").is_ok() || env::var("BUCK_BUILD_ID").is_ok() {
buck2_development(config)
} else if let Ok(dir) = env::var("CARGO_MANIFEST_DIR") {
cargo_development(dir, config)
} else {
Ok(())
}
}
fn buck2_development(config: &mut ConfigFile) -> Result<()> {
let resources = Buck2Resources::read().map_err(ConfigError::development)?;
let veritech_encryption_key_path = resources
.get_ends_with("dev.encryption.key")
.map_err(ConfigError::development)?
.to_string_lossy()
.to_string();
let symmetric_crypto_service_key = resources
.get_ends_with("dev.donkey.key")
.map_err(ConfigError::development)?
.to_string_lossy()
.to_string();
let postgres_cert = resources
.get_ends_with("dev.postgres.root.crt")
.map_err(ConfigError::development)?
.to_string_lossy()
.to_string();
warn!(
veritech_encryption_key_path = veritech_encryption_key_path.as_str(),
symmetric_crypto_service_key = symmetric_crypto_service_key.as_str(),
postgres_cert = postgres_cert.as_str(),
"detected development run",
);
config.crypto.encryption_key_file = veritech_encryption_key_path.parse().ok();
config.symmetric_crypto_service = SymmetricCryptoServiceConfigFile {
active_key: Some(symmetric_crypto_service_key),
active_key_base64: None,
extra_keys: vec![],
};
config.pg.certificate = Some(CertificateSource::Path(postgres_cert.clone().try_into()?));
config.layer_db_config.pg_pool_config.certificate =
Some(CertificateSource::Path(postgres_cert.clone().try_into()?));
config.layer_db_config.pg_pool_config.dbname = "si_layer_db".to_string();
Ok(())
}
fn cargo_development(dir: String, config: &mut ConfigFile) -> Result<()> {
let veritech_encryption_key_path = Path::new(&dir)
.join("../../lib/veritech-server/src/dev.encryption.key")
.to_string_lossy()
.to_string();
let symmetric_crypto_service_key = Path::new(&dir)
.join("../../lib/dal/dev.donkey.key")
.to_string_lossy()
.to_string();
let postgres_cert = Path::new(&dir)
.join("../../config/keys/dev.postgres.root.crt")
.to_string_lossy()
.to_string();
warn!(
veritech_encryption_key_path = veritech_encryption_key_path.as_str(),
symmetric_crypto_service_key = symmetric_crypto_service_key.as_str(),
postgres_cert = postgres_cert.as_str(),
"detected development run",
);
config.crypto.encryption_key_file = veritech_encryption_key_path.parse().ok();
config.symmetric_crypto_service = SymmetricCryptoServiceConfigFile {
active_key: Some(symmetric_crypto_service_key),
active_key_base64: None,
extra_keys: vec![],
};
config.pg.certificate = Some(CertificateSource::Path(postgres_cert.clone().try_into()?));
config.layer_db_config.pg_pool_config.certificate =
Some(CertificateSource::Path(postgres_cert.clone().try_into()?));
config.layer_db_config.pg_pool_config.dbname = "si_layer_db".to_string();
Ok(())
}
