dd4650a
use sea_orm::ConnectOptions;
use serde::{Deserialize, Serialize};
use std::time::Duration;
#[derive(Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct DatabaseEndpoint {
pub url: Option<String>,
pub host: Option<String>,
pub port: Option<u16>,
pub username: Option<String>,
pub password: Option<String>,
pub database: Option<String>,
pub schema_search_path: Option<String>,
pub application_name: Option<String>,
pub ssl_mode: Option<SslMode>,
pub max_connections: Option<u32>,
pub min_connections: Option<u32>,
pub connect_timeout_secs: Option<u64>,
pub idle_timeout_secs: Option<u64>,
pub acquire_timeout_secs: Option<u64>,
pub max_lifetime_secs: Option<u64>,
pub sqlx_logging: Option<bool>,
}
impl std::fmt::Debug for DatabaseEndpoint {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// `url` 常内嵌 `user:password@host`，两者都只能以掩码形式出现。
formatter
.debug_struct("DatabaseEndpoint")
.field("url", &self.url.as_ref().map(|_| "<redacted>"))
.field("host", &self.host)
.field("port", &self.port)
.field("username", &self.username)
.field("password", &self.password.as_ref().map(|_| "<redacted>"))
.field("database", &self.database)
.field("schema_search_path", &self.schema_search_path)
.field("application_name", &self.application_name)
.field("ssl_mode", &self.ssl_mode)
.field("max_connections", &self.max_connections)
.field("min_connections", &self.min_connections)
.field("connect_timeout_secs", &self.connect_timeout_secs)
.field("idle_timeout_secs", &self.idle_timeout_secs)
.field("acquire_timeout_secs", &self.acquire_timeout_secs)
.field("max_lifetime_secs", &self.max_lifetime_secs)
.field("sqlx_logging", &self.sqlx_logging)
.finish()
}
}
impl DatabaseEndpoint {
pub fn url(&self) -> String {
if let Some(url) = self.url.as_deref().filter(|s| !s.is_empty()) {
return url.to_string();
}
let host = self.host.as_deref().unwrap_or("localhost");
let port = self.port.unwrap_or(5432);
let user = url_encode(self.username.as_deref().unwrap_or("postgres"));
let db = url_encode(self.database.as_deref().unwrap_or("postgres"));
let password = match &self.password {
Some(p) if !p.is_empty() => format!(":{}", url_encode(p)),
_ => String::new(),
};
let mut url = format!("postgres://{user}{password}@{host}:{port}/{db}");
let mut params: Vec<String> = Vec::new();
if let Some(app) = self.application_name.as_deref().filter(|s| !s.is_empty()) {
params.push(format!("application_name={}", url_encode(app)));
}
if let Some(mode) = self.ssl_mode {
params.push(format!("sslmode={}", mode.as_query_value()));
}
if !params.is_empty() {
url.push('?');
url.push_str(¶ms.join("&"));
}
url
}
pub fn validate(&self) -> Result<(), String> {
if self.url.as_deref().is_some_and(|s| !s.is_empty()) {
return Ok(());
}
if self.host.as_deref().filter(|s| !s.is_empty()).is_none() {
return Err("endpoint.host must be set when endpoint.url is not provided".to_string());
}
if matches!(self.port, Some(0)) {
return Err("endpoint.port must be greater than 0".to_string());
}
Ok(())
}
pub fn inherit_from(&mut self, fallback: &Self) {
let describes_target = self.host.is_some()
|| self.port.is_some()
|| self.database.is_some()
|| self.username.is_some()
|| self.password.is_some();
if !describes_target && self.url.is_none() {
self.url = fallback.url.clone();
}
macro_rules! fill {
($($field:ident),* $(,)?) => {
$(
if self.$field.is_none() {
self.$field = fallback.$field.clone();
}
)*
};
}
fill!(
host,
port,
username,
password,
database,
schema_search_path,
application_name,
ssl_mode,
max_connections,
min_connections,
connect_timeout_secs,
idle_timeout_secs,
acquire_timeout_secs,
max_lifetime_secs,
sqlx_logging,
);
}
pub fn apply_to(&self, opts: &mut ConnectOptions) -> Result<(), String> {
self.validate()?;
if let Some(v) = self.schema_search_path.as_deref().filter(|s| !s.is_empty()) {
opts.set_schema_search_path(v);
}
if let Some(v) = self.sqlx_logging {
opts.sqlx_logging(v);
}
if let Some(v) = self.max_connections {
opts.max_connections(v);
}
if let Some(v) = self.min_connections {
opts.min_connections(v);
}
if let Some(v) = self.connect_timeout_secs {
opts.connect_timeout(Duration::from_secs(v));
}
if let Some(v) = self.idle_timeout_secs {
opts.idle_timeout(Duration::from_secs(v));
}
if let Some(v) = self.acquire_timeout_secs {
opts.acquire_timeout(Duration::from_secs(v));
}
if let Some(v) = self.max_lifetime_secs {
opts.max_lifetime(Duration::from_secs(v));
}
Ok(())
}
pub fn connect_options(&self) -> Result<ConnectOptions, String> {
let mut opts = ConnectOptions::new(self.url());
self.apply_to(&mut opts)?;
Ok(opts)
}
}
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default)]
pub struct DatabaseGroup {
#[serde(flatten)]
pub primary: DatabaseEndpoint,
pub replicas: Vec<DatabaseEndpoint>,
}
impl DatabaseGroup {
pub fn primary_url(&self) -> String {
self.primary.url()
}
pub fn replica_urls(&self) -> Vec<String> {
self.replicas.iter().map(DatabaseEndpoint::url).collect()
}
pub fn replica_count(&self) -> usize {
self.replicas.len()
}
pub fn has_replicas(&self) -> bool {
!self.replicas.is_empty()
}
pub fn validate(&self) -> Result<(), String> {
self.primary.validate()?;
for (i, replica) in self.replicas.iter().enumerate() {
replica
.validate()
.map_err(|e| format!("replicas[{i}]: {e}"))?;
}
Ok(())
}
pub fn primary_connect_options(&self) -> Result<ConnectOptions, String> {
self.validate()?;
self.primary.connect_options()
}
pub fn replica_connect_options(&self, index: usize) -> Result<ConnectOptions, String> {
let replica = self
.replicas
.get(index)
.ok_or_else(|| format!("replica index {index} out of range"))?;
replica.connect_options()
}
pub fn connect_options_for(&self, role: DatabaseRole) -> Result<ConnectOptions, String> {
match role {
DatabaseRole::Primary => self.primary_connect_options(),
DatabaseRole::Replica => {
if self.replicas.is_empty() {
return self.primary_connect_options();
}
self.replica_connect_options(0)
}
}
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SslMode {
#[default]
Disable,
Prefer,
Require,
VerifyCa,
VerifyFull,
}
impl SslMode {
pub fn as_query_value(&self) -> &'static str {
match self {
SslMode::Disable => "disable",
SslMode::Prefer => "prefer",
SslMode::Require => "require",
SslMode::VerifyCa => "verify-ca",
SslMode::VerifyFull => "verify-full",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DatabaseRole {
Primary,
Replica,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct DatabaseConfig {
#[serde(flatten)]
pub default: DatabaseGroup,
}
impl DatabaseConfig {
pub fn default_group(&self) -> &DatabaseGroup {
&self.default
}
pub fn validate(&self) -> Result<(), String> {
self.default
.validate()
.map_err(|reason| format!("database: {reason}"))
}
pub fn primary_url(&self) -> String {
self.default.primary_url()
}
pub fn replica_url(&self, index: usize) -> Option<String> {
self.default.replicas.get(index).map(DatabaseEndpoint::url)
}
pub fn replica_urls(&self) -> Vec<String> {
self.default.replica_urls()
}
pub fn replica_count(&self) -> usize {
self.default.replica_count()
}
pub fn has_replicas(&self) -> bool {
self.default.has_replicas()
}
pub fn primary_connect_options(&self) -> Result<ConnectOptions, String> {
self.default.primary_connect_options()
}
pub fn replica_connect_options(&self, index: usize) -> Result<ConnectOptions, String> {
self.default.replica_connect_options(index)
}
pub fn connect_options_for(&self, role: DatabaseRole) -> Result<ConnectOptions, String> {
self.default.connect_options_for(role)
}
}
pub(crate) fn url_encode(value: &str) -> String {
let mut out = String::with_capacity(value.len());
for b in value.as_bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(*b as char);
}
_ => out.push_str(&format!("%{:02X}", b)),
}
}
out
}
impl crate::config::loader::Section for DatabaseConfig {
const KEY: &'static str = "database";
}