use super::client::Client;
use crate::client::*;
use crate::server_models::*;
use http::Uri;
use log::{error, info, warn};
use std::env;
use std::fmt::Display;
use std::future::Future;
use std::hash::Hash;
use std::{
collections::HashMap,
error::Error,
time::Duration,
};
const BASE_URL_ENV: &str = "INTEGRESQL_BASE_URL";
const DEFAULT_TIMEOUT_ENV: &str = "INTEGRESQL_IMEOUT_SECONDS";
pub(crate) const DEFAULT_BASE_URL: &str = "http://integresql:5000/api";
pub trait TemplateInitializer {
fn setup(self, config: ConnectionSettings) -> Result<(), Box<dyn Error>>;
}
impl<F> TemplateInitializer for F
where
F: FnOnce(ConnectionSettings) -> Result<(), Box<dyn Error>>,
{
fn setup(self, config: ConnectionSettings) -> Result<(), Box<dyn Error>> {
self(config)
}
}
pub trait AsyncTemplateInitializer {
fn setup(
self,
config: ConnectionSettings,
) -> impl Future<Output = Result<(), Box<dyn Error + 'static>>> + Send;
}
impl<F, Fut> AsyncTemplateInitializer for F
where
F: FnOnce(ConnectionSettings) -> Fut + Send,
Fut: Future<Output = Result<(), Box<dyn Error + 'static>>> + Send,
{
fn setup(
self,
config: ConnectionSettings,
) -> impl Future<Output = Result<(), Box<dyn Error + 'static>>> + Send {
self(config)
}
}
pub struct TemplateDb {
client: Client,
template_hash: TemplateHash,
}
impl Display for TemplateDb {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "TemplateDb({})", self.template_hash)
}
}
impl TemplateDb {
pub fn get_writable_test_db(&self) -> Result<ConnectionSettings, IntegresqlError> {
let response = self.client.get_test_db(self.template_hash)?;
Ok(self.make_test_db(response, false))
}
pub fn get_readonly_test_db(&self) -> Result<ConnectionSettings, IntegresqlError> {
let response = self.client.get_test_db(self.template_hash)?;
Ok(self.make_test_db(response, true))
}
pub fn get_template_id(&self) -> String {
self.template_hash.to_string()
}
fn make_test_db(&self, response: GetTestDbResponse, reuse: bool) -> ConnectionSettings {
let drop_action = if reuse {
DropAction::Unlock(self.client.clone(), response.id)
} else {
DropAction::Recreate(self.client.clone(), response.id)
};
ConnectionSettings {
host: response.database.config.host,
port: response.database.config.port,
username: response.database.config.username,
password: response.database.config.password,
database: response.database.config.database,
additional_params: response.database.config.additional_params,
template_hash: response.database.template_hash,
id: Some(response.id),
drop_action: Some(drop_action),
}
}
}
#[derive(Debug, Clone)]
pub struct DbManager {
client: Client,
}
impl Default for DbManager {
fn default() -> Self {
DbManager::from_env()
}
}
impl DbManager {
pub fn from_env() -> Self {
let base_url = env::var(BASE_URL_ENV)
.unwrap_or_else(|_| DEFAULT_BASE_URL.to_string())
.parse::<Uri>()
.expect("Invalid URI in environment variable INTEGRESQL_CLIENT_BASE_URL");
let timeout = match env::var(DEFAULT_TIMEOUT_ENV) {
Ok(val) => val
.parse::<u64>()
.map(Duration::from_secs)
.unwrap_or(DEFAULT_TIMEOUT),
Err(_) => DEFAULT_TIMEOUT,
};
info!(
"Configured Integresql client from environment: base_url={} timeout_seconds={}",
base_url,
timeout.as_secs()
);
let client = Client::new(base_url, timeout);
DbManager { client }
}
pub fn new(base_uri: Uri, timeout: Duration) -> Self {
let client = Client::new(base_uri, timeout);
DbManager { client }
}
pub fn clear_db_tracking(&self) -> Result<(), IntegresqlError> {
self.client.clear_db_tracking()
}
pub fn discard_template(&self, template_key: impl Hash) -> Result<(), IntegresqlError> {
let template_hash = TemplateHash::from_hash(template_key);
self.client.discard_template(template_hash)
}
pub async fn get_template_db_async(
&self,
template_key: impl Hash,
initializer: impl AsyncTemplateInitializer,
) -> Result<TemplateDb, IntegresqlError> {
let template_hash = TemplateHash::from_hash(template_key);
let supplier = TemplateDb {
client: self.client.clone(),
template_hash,
};
let template_settings = match supplier.client.initialize_template(supplier.template_hash) {
InitializeTemplateResult::Success(db) => db,
InitializeTemplateResult::TemplateAlreadyInitialized => return Ok(supplier),
InitializeTemplateResult::Err(e) => return Err(e),
};
let connection_settings = ConnectionSettings {
host: template_settings.config.host,
port: template_settings.config.port,
username: template_settings.config.username,
password: template_settings.config.password,
database: template_settings.config.database,
additional_params: template_settings.config.additional_params,
template_hash: supplier.template_hash,
id: None, drop_action: None, };
let result = initializer.setup(connection_settings).await;
Self::handle_template_setup_result(supplier, result)
}
pub fn get_template_db_sync(
&self,
template_key: impl Hash,
initializer: impl TemplateInitializer,
) -> Result<TemplateDb, IntegresqlError> {
let template_hash = TemplateHash::from_hash(template_key);
let supplier = TemplateDb {
client: self.client.clone(),
template_hash,
};
let template_settings = match supplier.client.initialize_template(supplier.template_hash) {
InitializeTemplateResult::Success(db) => db,
InitializeTemplateResult::TemplateAlreadyInitialized => return Ok(supplier),
InitializeTemplateResult::Err(e) => return Err(e),
};
let connection_settings = ConnectionSettings {
host: template_settings.config.host,
port: template_settings.config.port,
username: template_settings.config.username,
password: template_settings.config.password,
database: template_settings.config.database,
additional_params: template_settings.config.additional_params,
template_hash: supplier.template_hash,
id: None, drop_action: None, };
let result = initializer.setup(connection_settings);
Self::handle_template_setup_result(supplier, result)
}
fn handle_template_setup_result(
supplier: TemplateDb,
result: Result<(), Box<dyn Error>>,
) -> Result<TemplateDb, IntegresqlError> {
match result {
Ok(_) => {
info!(
"Template setup completed successfully template_hash={}",
supplier.template_hash
);
supplier.client.finalize_template(supplier.template_hash)?;
Ok(supplier)
}
Err(e) => {
warn!(
"Template setup failed for hash template_hash={} error={}",
supplier.template_hash, e
);
supplier.client.discard_template(supplier.template_hash)?;
Err(IntegresqlError::SetupError(e.to_string()))
}
}
}
}
#[derive(Debug, Clone)]
enum DropAction {
Recreate(Client, i32),
Unlock(Client, i32),
}
#[derive(Debug, Clone)]
pub struct ConnectionSettings {
pub host: String,
pub port: u16,
pub username: String,
pub password: String,
pub database: String,
pub additional_params: Option<HashMap<String, String>>,
pub id: Option<i32>,
template_hash: TemplateHash,
drop_action: Option<DropAction>,
}
impl Drop for ConnectionSettings {
fn drop(&mut self) {
if let Some(drop_action) = self.drop_action.take() {
match drop_action {
DropAction::Recreate(client, id) => {
client.recreate_test_db(self.template_hash, id)
.unwrap_or_else(|e| {
error!("Failed to recreate test database template_hash={} test_db_id={} error={}", self.template_hash, id, e);
});
}
DropAction::Unlock(client, id) => {
client.unlock_test_db(self.template_hash, id)
.unwrap_or_else(|e| {
error!("Failed to unlock test database template_hash={} test_db_id={} error={}", self.template_hash, id, e);
});
}
}
}
}
}
impl ConnectionSettings {
pub fn to_libpq_url(&self) -> String {
{
let host: &str = &self.host;
let port = self.port;
let username: &str = &self.username;
let password: &str = &self.password;
let database: &str = &self.database;
let additional_params = &self.additional_params;
let additional_params = additional_params
.as_ref()
.map(|params| {
params
.iter()
.map(|(k, v)| format!("{}={}", k, v))
.collect::<Vec<_>>()
.join("&")
})
.unwrap_or_default();
format!(
"postgres://{}:{}@{}:{}/{}?{}",
username, password, host, port, database, additional_params
)
}
}
pub fn template_hash(&self) -> String {
self.template_hash.to_string()
}
}