513 lines
21 KiB
Rust
513 lines
21 KiB
Rust
// file: ks-store/src/postgres/store.rs
|
|
// version: 3
|
|
|
|
//! Store implementation scaffold for the `ks-store` crate.
|
|
|
|
/// PostgreSQL store connection options.
|
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
|
pub struct PostgresStoreOptions {
|
|
/// Database URL or DSN.
|
|
pub database_url: std::string::String,
|
|
/// Maximum connection count.
|
|
pub max_connections: u32,
|
|
/// Connection timeout in milliseconds.
|
|
pub connect_timeout_ms: u64,
|
|
/// Enables idempotent raw schema initialization at startup.
|
|
pub auto_initialize_schema: bool,
|
|
}
|
|
|
|
impl PostgresStoreOptions {
|
|
/// Creates validated PostgreSQL store options.
|
|
pub fn new(
|
|
database_url: impl std::convert::Into<std::string::String>,
|
|
max_connections: u32,
|
|
connect_timeout_ms: u64,
|
|
auto_initialize_schema: bool,
|
|
) -> ks_core::Result<Self> {
|
|
let database_url_value = database_url.into();
|
|
if database_url_value.trim().is_empty() {
|
|
return std::result::Result::Err(ks_core::Error::db(
|
|
"postgres database url must not be empty",
|
|
));
|
|
}
|
|
if max_connections == 0 {
|
|
return std::result::Result::Err(ks_core::Error::db(
|
|
"postgres max_connections must be greater than zero",
|
|
));
|
|
}
|
|
if connect_timeout_ms == 0 {
|
|
return std::result::Result::Err(ks_core::Error::db(
|
|
"postgres connect_timeout_ms must be greater than zero",
|
|
));
|
|
}
|
|
return std::result::Result::Ok(Self {
|
|
database_url: database_url_value,
|
|
max_connections,
|
|
connect_timeout_ms,
|
|
auto_initialize_schema,
|
|
});
|
|
}
|
|
|
|
/// Returns a DSN masked for diagnostics.
|
|
pub fn masked_dsn(&self) -> std::string::String {
|
|
return crate::mask_postgres_dsn(self.database_url.as_str());
|
|
}
|
|
}
|
|
|
|
/// PostgreSQL diagnostic snapshot.
|
|
#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
|
|
pub struct PostgresBackendDiagnostics {
|
|
/// Backend descriptor safe for UI display.
|
|
pub descriptor: crate::StoreBackendDescriptor,
|
|
/// Backend health snapshot.
|
|
pub health: crate::StoreHealthSnapshot,
|
|
/// Migration status snapshot.
|
|
pub migrations: crate::StoreMigrationSnapshot,
|
|
/// Full PostgreSQL server version string when available.
|
|
pub server_version: std::option::Option<std::string::String>,
|
|
}
|
|
|
|
/// Read-only statistics for one PostgreSQL table.
|
|
#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
|
|
pub struct PostgresTableStatistics {
|
|
/// Number of rows currently stored in the table.
|
|
pub row_count: i64,
|
|
/// Lowest observed Solana slot when the table contains a slot column and rows.
|
|
pub min_slot: std::option::Option<i64>,
|
|
/// Highest observed Solana slot when the table contains a slot column and rows.
|
|
pub max_slot: std::option::Option<i64>,
|
|
/// Latest insertion timestamp rendered by PostgreSQL for UI diagnostics.
|
|
pub latest_created_at: std::option::Option<std::string::String>,
|
|
}
|
|
|
|
/// Read-only diagnostics for one expected PostgreSQL table.
|
|
#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
|
|
pub struct PostgresTableDiagnostics {
|
|
/// Expected table name.
|
|
pub table_name: std::string::String,
|
|
/// Logical Solana domain encoded in the table name.
|
|
pub domain: std::string::String,
|
|
/// Human-readable role of the table.
|
|
pub role: std::string::String,
|
|
/// Whether the table exists in the current PostgreSQL search path.
|
|
pub exists: bool,
|
|
/// Table statistics when the table exists.
|
|
pub statistics: std::option::Option<crate::PostgresTableStatistics>,
|
|
}
|
|
|
|
/// PostgreSQL store handle.
|
|
#[derive(Clone, Debug)]
|
|
pub struct PostgresStore {
|
|
options: crate::PostgresStoreOptions,
|
|
pool: sqlx::PgPool,
|
|
}
|
|
|
|
impl PostgresStore {
|
|
/// Connects to PostgreSQL from typed store options.
|
|
pub async fn connect(options: crate::PostgresStoreOptions) -> ks_core::Result<Self> {
|
|
let pool_options = sqlx::postgres::PgPoolOptions::new()
|
|
.max_connections(options.max_connections)
|
|
.acquire_timeout(std::time::Duration::from_millis(options.connect_timeout_ms));
|
|
let connect_result = pool_options.connect(options.database_url.as_str()).await;
|
|
return match connect_result {
|
|
std::result::Result::Ok(pool) => {
|
|
let store = Self { options, pool };
|
|
if store.options.auto_initialize_schema {
|
|
let schema_result = store.initialize_store_schema().await;
|
|
if let std::result::Result::Err(error) = schema_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
}
|
|
std::result::Result::Ok(store)
|
|
},
|
|
std::result::Result::Err(error) => std::result::Result::Err(ks_core::Error::db(
|
|
format!("postgres connection failed: {error}"),
|
|
)),
|
|
};
|
|
}
|
|
|
|
/// Creates a store handle from an existing PostgreSQL pool.
|
|
pub fn from_pool(options: crate::PostgresStoreOptions, pool: sqlx::PgPool) -> Self {
|
|
return Self { options, pool };
|
|
}
|
|
|
|
/// Returns the underlying PostgreSQL pool.
|
|
pub fn pool(&self) -> &sqlx::PgPool {
|
|
return &self.pool;
|
|
}
|
|
|
|
/// Returns the connection options used to create this store.
|
|
pub fn options(&self) -> &crate::PostgresStoreOptions {
|
|
return &self.options;
|
|
}
|
|
|
|
/// Applies each idempotent store schema once per invocation in dependency order.
|
|
pub async fn initialize_store_schema(&self) -> ks_core::Result<()> {
|
|
let raw_result = crate::postgres::query::apply_raw_store_schema(&self.pool).await;
|
|
if let std::result::Result::Err(error) = raw_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let core_result = crate::postgres::query::apply_core_store_schema(&self.pool).await;
|
|
if let std::result::Result::Err(error) = core_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return crate::postgres::query::apply_decode_store_schema(&self.pool).await;
|
|
}
|
|
|
|
/// Applies the idempotent minimal raw Solana store schema.
|
|
pub async fn initialize_raw_store_schema(&self) -> ks_core::Result<()> {
|
|
return crate::postgres::query::apply_raw_store_schema(&self.pool).await;
|
|
}
|
|
|
|
/// Applies the idempotent minimal core Solana store schema.
|
|
pub async fn initialize_core_store_schema(&self) -> ks_core::Result<()> {
|
|
let raw_result = self.initialize_raw_store_schema().await;
|
|
if let std::result::Result::Err(error) = raw_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return crate::postgres::query::apply_core_store_schema(&self.pool).await;
|
|
}
|
|
|
|
/// Applies the idempotent common decode and materialization store schema.
|
|
pub async fn initialize_decode_store_schema(&self) -> ks_core::Result<()> {
|
|
let core_result = self.initialize_core_store_schema().await;
|
|
if let std::result::Result::Err(error) = core_result {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return crate::postgres::query::apply_decode_store_schema(&self.pool).await;
|
|
}
|
|
|
|
/// Reads a UI-safe backend descriptor.
|
|
pub async fn backend_descriptor(&self) -> ks_core::Result<crate::StoreBackendDescriptor> {
|
|
let schema_result = crate::postgres::query::load_current_schema(&self.pool).await;
|
|
return match schema_result {
|
|
std::result::Result::Ok(schema) => crate::StoreBackendDescriptor::new(
|
|
crate::StoreBackendKind::Postgres,
|
|
"postgres",
|
|
std::option::Option::Some(self.options.masked_dsn()),
|
|
std::option::Option::Some(schema),
|
|
),
|
|
std::result::Result::Err(error) => std::result::Result::Err(error),
|
|
};
|
|
}
|
|
|
|
/// Reads a PostgreSQL health snapshot.
|
|
pub async fn health_snapshot(&self) -> ks_core::Result<crate::StoreHealthSnapshot> {
|
|
let health_result = crate::postgres::query::run_health_check(&self.pool).await;
|
|
return match health_result {
|
|
std::result::Result::Ok(()) => crate::StoreHealthSnapshot::new(
|
|
"postgres",
|
|
crate::StoreHealthStatus::Healthy,
|
|
std::option::Option::Some(std::string::String::from("SELECT 1 succeeded")),
|
|
),
|
|
std::result::Result::Err(error) => crate::StoreHealthSnapshot::new(
|
|
"postgres",
|
|
crate::StoreHealthStatus::Unhealthy,
|
|
std::option::Option::Some(error.to_string()),
|
|
),
|
|
};
|
|
}
|
|
|
|
/// Reads a non-destructive migration snapshot.
|
|
pub async fn migration_snapshot(&self) -> ks_core::Result<crate::StoreMigrationSnapshot> {
|
|
let migration_table_result =
|
|
crate::postgres::query::load_migration_table_name(&self.pool).await;
|
|
return match migration_table_result {
|
|
std::result::Result::Ok(std::option::Option::None) => {
|
|
std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
|
|
crate::StoreMigrationStatus::NotInitialized,
|
|
std::option::Option::None,
|
|
std::vec::Vec::new(),
|
|
std::option::Option::Some(std::string::String::from(
|
|
"no sqlx migration table detected; 0.3.1 canonical acquisition/core schemas use idempotent crate-managed DDL",
|
|
)),
|
|
))
|
|
},
|
|
std::result::Result::Ok(std::option::Option::Some(_table_name)) => {
|
|
self.migration_snapshot_from_existing_table().await
|
|
},
|
|
std::result::Result::Err(error) => std::result::Result::Err(error),
|
|
};
|
|
}
|
|
|
|
/// Reads a complete PostgreSQL diagnostic snapshot.
|
|
pub async fn backend_diagnostics(&self) -> ks_core::Result<crate::PostgresBackendDiagnostics> {
|
|
let descriptor_result = self.backend_descriptor().await;
|
|
let descriptor = match descriptor_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let health_result = self.health_snapshot().await;
|
|
let health = match health_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let migrations_result = self.migration_snapshot().await;
|
|
let migrations = match migrations_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let server_version = match crate::postgres::query::load_server_version(&self.pool).await {
|
|
std::result::Result::Ok(value) => std::option::Option::Some(value),
|
|
std::result::Result::Err(_error) => std::option::Option::None,
|
|
};
|
|
return std::result::Result::Ok(crate::PostgresBackendDiagnostics {
|
|
descriptor,
|
|
health,
|
|
migrations,
|
|
server_version,
|
|
});
|
|
}
|
|
|
|
/// Lists bounded raw transaction candidates enriched with core and ledger diagnostics.
|
|
pub async fn replay_transaction_candidates(
|
|
&self,
|
|
filter: &crate::PostgresReplayTransactionFilter,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresReplayTransactionCandidate>> {
|
|
return crate::postgres::query::list_replay_transaction_candidates(&self.pool, filter)
|
|
.await;
|
|
}
|
|
|
|
/// Lists bounded program summaries across outer, inner and reliably linked logs.
|
|
pub async fn replay_program_summaries(
|
|
&self,
|
|
filter: &crate::PostgresReplayProgramFilter,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresReplayProgramSummary>> {
|
|
return crate::postgres::query::list_replay_program_summaries(&self.pool, filter).await;
|
|
}
|
|
|
|
/// Lists bounded mint, owner or account-key summaries from core tables.
|
|
pub async fn replay_entity_summaries(
|
|
&self,
|
|
filter: &crate::PostgresReplayEntityFilter,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresReplayEntitySummary>> {
|
|
return crate::postgres::query::list_replay_entity_summaries(&self.pool, filter).await;
|
|
}
|
|
|
|
/// Reads diagnostics for raw Solana store tables without changing the schema.
|
|
pub async fn raw_table_diagnostics(
|
|
&self,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
|
let specs = crate::raw_store_table_diagnostic_specs();
|
|
return self.table_diagnostics(&specs).await;
|
|
}
|
|
|
|
/// Reads diagnostics for core Solana store tables without changing the schema.
|
|
pub async fn core_table_diagnostics(
|
|
&self,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
|
let specs = crate::core_store_table_diagnostic_specs();
|
|
return self.table_diagnostics(&specs).await;
|
|
}
|
|
|
|
/// Reads diagnostics for decode and materialization store tables without changing the schema.
|
|
pub async fn decode_table_diagnostics(
|
|
&self,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
|
let specs = crate::decode_store_table_diagnostic_specs();
|
|
return self.table_diagnostics(&specs).await;
|
|
}
|
|
|
|
/// Reads diagnostics for every known raw/core/decode Solana store table.
|
|
pub async fn known_table_diagnostics(
|
|
&self,
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
|
let mut diagnostics = std::vec::Vec::new();
|
|
let raw_result = self.raw_table_diagnostics().await;
|
|
let raw_tables = match raw_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
for table in raw_tables {
|
|
diagnostics.push(table);
|
|
}
|
|
let core_result = self.core_table_diagnostics().await;
|
|
let core_tables = match core_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
for table in core_tables {
|
|
diagnostics.push(table);
|
|
}
|
|
let decode_result = self.decode_table_diagnostics().await;
|
|
let decode_tables = match decode_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
for table in decode_tables {
|
|
diagnostics.push(table);
|
|
}
|
|
return std::result::Result::Ok(diagnostics);
|
|
}
|
|
|
|
async fn table_diagnostics(
|
|
&self,
|
|
specs: &[crate::PostgresTableDiagnosticSpec],
|
|
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
|
let mut diagnostics = std::vec::Vec::new();
|
|
for spec in specs {
|
|
let exists_result =
|
|
crate::postgres::query::table_exists(&self.pool, spec.table_name).await;
|
|
let exists = match exists_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let statistics = if exists {
|
|
let statistics_result =
|
|
crate::postgres::query::load_table_statistics(&self.pool, spec.table_name)
|
|
.await;
|
|
match statistics_result {
|
|
std::result::Result::Ok(value) => std::option::Option::Some(value),
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
}
|
|
} else {
|
|
std::option::Option::None
|
|
};
|
|
diagnostics.push(crate::PostgresTableDiagnostics {
|
|
table_name: spec.table_name.to_string(),
|
|
domain: spec.domain.to_string(),
|
|
role: spec.role.to_string(),
|
|
exists,
|
|
statistics,
|
|
});
|
|
}
|
|
return std::result::Result::Ok(diagnostics);
|
|
}
|
|
|
|
async fn migration_snapshot_from_existing_table(
|
|
&self,
|
|
) -> ks_core::Result<crate::StoreMigrationSnapshot> {
|
|
let version_result =
|
|
crate::postgres::query::load_latest_migration_version(&self.pool).await;
|
|
return match version_result {
|
|
std::result::Result::Ok(current_version) => {
|
|
std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
|
|
crate::StoreMigrationStatus::Current,
|
|
current_version,
|
|
std::vec::Vec::new(),
|
|
std::option::Option::Some(std::string::String::from(
|
|
"sqlx migration table detected; canonical acquisition/core schema remains idempotent and crate-managed in 0.3.1",
|
|
)),
|
|
))
|
|
},
|
|
std::result::Result::Err(error) => std::result::Result::Err(error),
|
|
};
|
|
}
|
|
}
|
|
|
|
/// Returns a DSN masked for logs and UI diagnostics.
|
|
pub fn mask_postgres_dsn(dsn: &str) -> std::string::String {
|
|
let trimmed_dsn = dsn.trim();
|
|
if trimmed_dsn.is_empty() {
|
|
return std::string::String::from("");
|
|
}
|
|
let queryless = crate::postgres::store::strip_query(trimmed_dsn);
|
|
return match queryless.split_once("://") {
|
|
std::option::Option::Some((scheme, remainder)) => {
|
|
crate::postgres::store::mask_scheme_remainder(
|
|
scheme,
|
|
remainder,
|
|
trimmed_dsn.contains('?'),
|
|
)
|
|
},
|
|
std::option::Option::None => crate::postgres::store::mask_plain_dsn(queryless.as_str()),
|
|
};
|
|
}
|
|
|
|
fn strip_query(dsn: &str) -> std::string::String {
|
|
return match dsn.split_once('?') {
|
|
std::option::Option::Some((prefix, _query)) => std::string::String::from(prefix),
|
|
std::option::Option::None => std::string::String::from(dsn),
|
|
};
|
|
}
|
|
|
|
fn mask_scheme_remainder(scheme: &str, remainder: &str, had_query: bool) -> std::string::String {
|
|
let suffix = crate::postgres::store::query_suffix(had_query);
|
|
return match remainder.rsplit_once('@') {
|
|
std::option::Option::Some((_userinfo, host_path)) => {
|
|
format!("{scheme}://***:***@{host_path}{suffix}")
|
|
},
|
|
std::option::Option::None => format!("{scheme}://{remainder}{suffix}"),
|
|
};
|
|
}
|
|
|
|
fn mask_plain_dsn(dsn: &str) -> std::string::String {
|
|
if dsn.contains("password=") {
|
|
return std::string::String::from("<postgres-dsn-redacted>");
|
|
}
|
|
return std::string::String::from(dsn);
|
|
}
|
|
|
|
fn query_suffix(had_query: bool) -> std::string::String {
|
|
if had_query {
|
|
return std::string::String::from("?<redacted>");
|
|
}
|
|
return std::string::String::from("");
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
#[test]
|
|
fn options_reject_empty_database_url() {
|
|
let result = crate::PostgresStoreOptions::new(" ", 1, 1000, false);
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn options_reject_zero_max_connections() {
|
|
let result = crate::PostgresStoreOptions::new("postgres://localhost/db", 0, 1000, false);
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn options_reject_zero_connect_timeout() {
|
|
let result = crate::PostgresStoreOptions::new("postgres://localhost/db", 1, 0, false);
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn optional_postgres_healthcheck_from_env() {
|
|
let database_url = match std::env::var("KB_POSTGRES_TEST_URL") {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_error) => return,
|
|
};
|
|
let _postgres_guard = crate::postgres::test_serial::postgres_test_guard().await;
|
|
let options_result = crate::PostgresStoreOptions::new(database_url, 1, 5000, false);
|
|
let options = match options_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => panic!("unexpected options error: {error}"),
|
|
};
|
|
let store_result = crate::PostgresStore::connect(options).await;
|
|
let store = match store_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => panic!("unexpected connect error: {error}"),
|
|
};
|
|
let health_result = store.health_snapshot().await;
|
|
let health = match health_result {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => panic!("unexpected health error: {error}"),
|
|
};
|
|
assert_eq!(health.status, crate::StoreHealthStatus::Healthy);
|
|
return;
|
|
}
|
|
|
|
#[test]
|
|
fn mask_postgres_dsn_masks_userinfo() {
|
|
let masked = crate::mask_postgres_dsn("postgres://user:secret@localhost:5432/db");
|
|
assert_eq!(masked, "postgres://***:***@localhost:5432/db");
|
|
}
|
|
|
|
#[test]
|
|
fn mask_postgres_dsn_masks_query_string() {
|
|
let masked =
|
|
crate::mask_postgres_dsn("postgres://localhost/db?sslmode=require&password=secret");
|
|
assert_eq!(masked, "postgres://localhost/db?<redacted>");
|
|
}
|
|
|
|
#[test]
|
|
fn mask_postgres_dsn_masks_plain_password_dsn() {
|
|
let masked = crate::mask_postgres_dsn("host=localhost password=secret dbname=khadhroony");
|
|
assert_eq!(masked, "<postgres-dsn-redacted>");
|
|
}
|
|
}
|