use std::{fmt, time::Duration};
use serde::{Deserialize, Deserializer, Serialize, Serializer, de, ser::SerializeStruct};
use crate::{Password, StateStoreError};
const DEFAULT_POSTGRES_PORT: u16 = 5432;
const DEFAULT_OPERATION_TIMEOUT_MS: u64 = 5_000;
pub enum PostgresStateStoreTlsMode {
Disable,
}
impl Clone for PostgresStateStoreTlsMode {
fn clone(&self) -> Self {
*self
}
}
impl Copy for PostgresStateStoreTlsMode {}
impl fmt::Debug for PostgresStateStoreTlsMode {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Disable => f.write_str("Disable"),
}
}
}
impl PartialEq for PostgresStateStoreTlsMode {
fn eq(&self, other: &Self) -> bool {
matches!((self, other), (Self::Disable, Self::Disable))
}
}
impl Eq for PostgresStateStoreTlsMode {}
impl Serialize for PostgresStateStoreTlsMode {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
match self {
Self::Disable => serializer.serialize_str("Disable"),
}
}
}
impl<'de> Deserialize<'de> for PostgresStateStoreTlsMode {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
match String::deserialize(deserializer)?.as_str() {
"Disable" => Ok(Self::Disable),
other => Err(de::Error::unknown_variant(other, &["Disable"])),
}
}
}
pub struct PostgresStateStoreConfig {
host: String,
port: u16,
database: String,
username: String,
password: Option<Password>,
tls_mode: PostgresStateStoreTlsMode,
operation_timeout_ms: u64,
}
impl PostgresStateStoreConfig {
#[must_use]
pub fn new(
host: impl Into<String>,
database: impl Into<String>,
username: impl Into<String>,
) -> Self {
Self {
host: host.into(),
port: DEFAULT_POSTGRES_PORT,
database: database.into(),
username: username.into(),
password: None,
tls_mode: PostgresStateStoreTlsMode::Disable,
operation_timeout_ms: DEFAULT_OPERATION_TIMEOUT_MS,
}
}
#[must_use]
pub fn localhost(database: impl Into<String>, username: impl Into<String>) -> Self {
Self::new("127.0.0.1", database, username)
}
#[must_use]
pub fn with_port(mut self, port: u16) -> Self {
self.port = port;
self
}
#[must_use]
pub fn with_password(mut self, password: Password) -> Self {
self.password = Some(password);
self
}
#[must_use]
pub fn without_password(mut self) -> Self {
self.password = None;
self
}
pub fn with_operation_timeout(mut self, timeout: Duration) -> Result<Self, StateStoreError> {
self.operation_timeout_ms = u64::try_from(timeout.as_millis()).map_err(|err| {
StateStoreError::invalid_config({
let mut message =
String::from("PostgreSQL state store operation_timeout overflow: ");
message.push_str(&err.to_string());
message
})
})?;
Ok(self)
}
#[must_use]
pub fn host(&self) -> &str {
&self.host
}
#[must_use]
pub fn port(&self) -> u16 {
self.port
}
#[must_use]
pub fn database(&self) -> &str {
&self.database
}
#[must_use]
pub fn username(&self) -> &str {
&self.username
}
#[must_use]
pub fn password(&self) -> Option<&Password> {
self.password.as_ref()
}
#[must_use]
pub fn tls_mode(&self) -> PostgresStateStoreTlsMode {
self.tls_mode
}
#[must_use]
pub fn operation_timeout(&self) -> Duration {
Duration::from_millis(self.operation_timeout_ms)
}
#[cfg(feature = "postgres")]
pub(crate) fn operation_timeout_ms(&self) -> u64 {
self.operation_timeout_ms
}
pub fn validate(&self) -> Result<(), StateStoreError> {
PostgresConfigText::new("PostgreSQL state store host", &self.host).validate()?;
PostgresConfigText::new("PostgreSQL state store database", &self.database).validate()?;
PostgresConfigText::new("PostgreSQL state store username", &self.username).validate()?;
if self.port == 0 {
return Err(StateStoreError::invalid_config(
"PostgreSQL state store port must be greater than zero",
));
}
if self.operation_timeout_ms == 0 {
return Err(StateStoreError::invalid_config(
"PostgreSQL state store operation_timeout must be greater than zero",
));
}
Ok(())
}
pub(crate) fn from_json_value(value: &serde_json::Value) -> Result<Self, String> {
let object = value
.as_object()
.ok_or_else(|| "PostgresStateStoreConfig must be an object".to_owned())?;
let host = required_string(object, "host")?;
let port = required_u64(object, "port")
.and_then(|value| u16::try_from(value).map_err(|err| err.to_string()))?;
let database = required_string(object, "database")?;
let username = required_string(object, "username")?;
let password = match object.get("password") {
Some(value) if !value.is_null() => {
Some(Password::deserialize(value).map_err(|err| err.to_string())?)
}
_ => None,
};
let tls_mode = match object.get("tls_mode") {
Some(value) => {
PostgresStateStoreTlsMode::deserialize(value).map_err(|err| err.to_string())?
}
None => PostgresStateStoreTlsMode::Disable,
};
let operation_timeout_ms = required_u64(object, "operation_timeout_ms")?;
Ok(Self {
host,
port,
database,
username,
password,
tls_mode,
operation_timeout_ms,
})
}
}
impl Clone for PostgresStateStoreConfig {
fn clone(&self) -> Self {
Self {
host: self.host.clone(),
port: self.port,
database: self.database.clone(),
username: self.username.clone(),
password: self.password.clone(),
tls_mode: self.tls_mode,
operation_timeout_ms: self.operation_timeout_ms,
}
}
}
impl fmt::Debug for PostgresStateStoreConfig {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("PostgresStateStoreConfig")
.field("host", &self.host)
.field("port", &self.port)
.field("database", &self.database)
.field("username", &self.username)
.field("password", &self.password.as_ref().map(|_| "[redacted]"))
.field("tls_mode", &self.tls_mode)
.field("operation_timeout_ms", &self.operation_timeout_ms)
.finish()
}
}
impl Serialize for PostgresStateStoreConfig {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
let mut state = serializer.serialize_struct("PostgresStateStoreConfig", 7)?;
state.serialize_field("host", &self.host)?;
state.serialize_field("port", &self.port)?;
state.serialize_field("database", &self.database)?;
state.serialize_field("username", &self.username)?;
state.serialize_field("password", &self.password)?;
state.serialize_field("tls_mode", &self.tls_mode)?;
state.serialize_field("operation_timeout_ms", &self.operation_timeout_ms)?;
state.end()
}
}
impl<'de> Deserialize<'de> for PostgresStateStoreConfig {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = serde_json::Value::deserialize(deserializer)?;
Self::from_json_value(&value).map_err(de::Error::custom)
}
}
struct PostgresConfigText<'a> {
label: &'static str,
value: &'a str,
}
impl<'a> PostgresConfigText<'a> {
fn new(label: &'static str, value: &'a str) -> Self {
Self { label, value }
}
fn validate(&self) -> Result<(), StateStoreError> {
if self.value.is_empty() {
return Err(StateStoreError::invalid_config(format!(
"{} must not be empty",
self.label
)));
}
if self.value.trim() != self.value {
return Err(StateStoreError::invalid_config(format!(
"{} must not contain leading or trailing whitespace",
self.label
)));
}
if self
.value
.chars()
.any(|character| character == '\0' || character.is_control())
{
return Err(StateStoreError::invalid_config(format!(
"{} must not contain NUL or control characters",
self.label
)));
}
Ok(())
}
}
fn required_string(
object: &serde_json::Map<String, serde_json::Value>,
field: &'static str,
) -> Result<String, String> {
object
.get(field)
.ok_or_else(|| required_message(field))
.and_then(|value| {
value
.as_str()
.map(ToOwned::to_owned)
.ok_or_else(|| type_message(field, "a string"))
})
}
fn required_u64(
object: &serde_json::Map<String, serde_json::Value>,
field: &'static str,
) -> Result<u64, String> {
object
.get(field)
.ok_or_else(|| required_message(field))
.and_then(|value| value.as_u64().ok_or_else(|| type_message(field, "a u64")))
}
fn required_message(field: &str) -> String {
let mut message = String::from("PostgresStateStoreConfig.");
message.push_str(field);
message.push_str(" is required");
message
}
fn type_message(field: &str, expected: &str) -> String {
let mut message = String::from("PostgresStateStoreConfig.");
message.push_str(field);
message.push_str(" must be ");
message.push_str(expected);
message
}