v0.5.3-pre.002
This commit is contained in:
@@ -1,42 +1,131 @@
|
||||
// file: ks-store/src/postgres/store.rs
|
||||
// version: 4
|
||||
// version: 9
|
||||
|
||||
//! Store implementation scaffold for the `ks-store` crate.
|
||||
//! PostgreSQL store implementation kept behind the backend-agnostic `Store` facade.
|
||||
|
||||
/// 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,
|
||||
/// PostgreSQL store connection options interpreted only inside `ks-store`.
|
||||
#[derive(Clone, Eq, PartialEq)]
|
||||
pub(crate) struct PostgresStoreOptions {
|
||||
database_url: std::string::String,
|
||||
max_connections: u32,
|
||||
connect_timeout_ms: u64,
|
||||
auto_initialize_schema: bool,
|
||||
option_count: u32,
|
||||
}
|
||||
|
||||
impl PostgresStoreOptions {
|
||||
/// Creates validated PostgreSQL store options.
|
||||
pub fn new(
|
||||
impl crate::PostgresStoreOptions {
|
||||
/// Creates validated PostgreSQL store options for crate-internal tests and adapters.
|
||||
#[cfg(test)]
|
||||
pub(crate) 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> {
|
||||
return Self::new_with_option_count(
|
||||
database_url,
|
||||
max_connections,
|
||||
connect_timeout_ms,
|
||||
auto_initialize_schema,
|
||||
4,
|
||||
);
|
||||
}
|
||||
|
||||
/// Interprets the selected opaque backend options supplied to `Store::open`.
|
||||
pub(crate) fn from_backend_options(options: &serde_json::Value) -> ks_core::Result<Self> {
|
||||
let object = match options.as_object() {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_invalid",
|
||||
"postgres backend options must be a JSON object",
|
||||
));
|
||||
},
|
||||
};
|
||||
let database_url = match object.get("url").and_then(serde_json::Value::as_str) {
|
||||
std::option::Option::Some(value) => value.to_string(),
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_incomplete",
|
||||
"postgres backend requires a non-empty connection URL",
|
||||
));
|
||||
},
|
||||
};
|
||||
let max_connections_u64 =
|
||||
match object.get("max_connections").and_then(serde_json::Value::as_u64) {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_incomplete",
|
||||
"postgres backend requires max_connections greater than zero",
|
||||
));
|
||||
},
|
||||
};
|
||||
let max_connections = match u32::try_from(max_connections_u64) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_invalid",
|
||||
"postgres max_connections does not fit into u32",
|
||||
));
|
||||
},
|
||||
};
|
||||
let connect_timeout_ms =
|
||||
match object.get("connect_timeout_ms").and_then(serde_json::Value::as_u64) {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_incomplete",
|
||||
"postgres backend requires connect_timeout_ms greater than zero",
|
||||
));
|
||||
},
|
||||
};
|
||||
let auto_initialize_schema =
|
||||
match object.get("auto_initialize_schema").and_then(serde_json::Value::as_bool) {
|
||||
std::option::Option::Some(value) => value,
|
||||
std::option::Option::None => {
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_incomplete",
|
||||
"postgres backend requires auto_initialize_schema",
|
||||
));
|
||||
},
|
||||
};
|
||||
let option_count = match u32::try_from(object.len()) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_) => u32::MAX,
|
||||
};
|
||||
return Self::new_with_option_count(
|
||||
database_url,
|
||||
max_connections,
|
||||
connect_timeout_ms,
|
||||
auto_initialize_schema,
|
||||
option_count,
|
||||
);
|
||||
}
|
||||
|
||||
fn new_with_option_count(
|
||||
database_url: impl std::convert::Into<std::string::String>,
|
||||
max_connections: u32,
|
||||
connect_timeout_ms: u64,
|
||||
auto_initialize_schema: bool,
|
||||
option_count: u32,
|
||||
) -> 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",
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_incomplete",
|
||||
"postgres backend requires a non-empty connection URL",
|
||||
));
|
||||
}
|
||||
if max_connections == 0 {
|
||||
return std::result::Result::Err(ks_core::Error::db(
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_invalid",
|
||||
"postgres max_connections must be greater than zero",
|
||||
));
|
||||
}
|
||||
if connect_timeout_ms == 0 {
|
||||
return std::result::Result::Err(ks_core::Error::db(
|
||||
return std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_options_invalid",
|
||||
"postgres connect_timeout_ms must be greater than zero",
|
||||
));
|
||||
}
|
||||
@@ -45,173 +134,142 @@ impl PostgresStoreOptions {
|
||||
max_connections,
|
||||
connect_timeout_ms,
|
||||
auto_initialize_schema,
|
||||
option_count,
|
||||
});
|
||||
}
|
||||
|
||||
/// Returns whether automatic schema initialization is enabled.
|
||||
pub(crate) fn auto_initialize_schema(&self) -> bool {
|
||||
return self.auto_initialize_schema;
|
||||
}
|
||||
|
||||
/// Returns a backend-neutral safe configuration summary.
|
||||
pub(crate) fn configuration_summary(&self) -> crate::StoreConfigurationSummary {
|
||||
return crate::StoreConfigurationSummary {
|
||||
enabled: true,
|
||||
backend_code: "postgres".to_string(),
|
||||
connection_configured: !self.database_url.trim().is_empty(),
|
||||
auto_initialize_schema: self.auto_initialize_schema,
|
||||
backend_option_count: self.option_count,
|
||||
};
|
||||
}
|
||||
|
||||
/// Returns a DSN masked for diagnostics.
|
||||
pub fn masked_dsn(&self) -> std::string::String {
|
||||
return crate::mask_postgres_dsn(self.database_url.as_str());
|
||||
pub(crate) fn masked_connection_descriptor(&self) -> std::string::String {
|
||||
return 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 {
|
||||
/// PostgreSQL store handle kept private to `ks-store`.
|
||||
#[derive(Clone)]
|
||||
pub(crate) 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> {
|
||||
impl crate::PostgresStore {
|
||||
/// Connects to PostgreSQL from validated backend options.
|
||||
pub(crate) async fn connect(options: crate::PostgresStoreOptions) -> ks_core::Result<Self> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "connection_open", "open PostgreSQL store connection");
|
||||
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)
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "connection_open", connected = true, "PostgreSQL store connection opened");
|
||||
std::result::Result::Ok(Self { options, pool })
|
||||
},
|
||||
std::result::Result::Err(_error) => {
|
||||
tracing::error!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "connection_open", connected = false, "PostgreSQL store connection failed");
|
||||
std::result::Result::Err(crate::storage_contract_error(
|
||||
"store_backend_connection_failed",
|
||||
"postgres connection failed",
|
||||
))
|
||||
},
|
||||
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 {
|
||||
/// Returns the underlying PostgreSQL pool to backend-private repositories.
|
||||
pub(crate) 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;
|
||||
pub(crate) async fn initialize_store_schema(&self) -> ks_core::Result<()> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize", "initialize PostgreSQL store schema");
|
||||
let raw_result = crate::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;
|
||||
let core_result = crate::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;
|
||||
let result = crate::apply_decode_store_schema(&self.pool).await;
|
||||
if result.is_ok() {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "schema_initialize", initialized = true, "PostgreSQL store schema initialized");
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
/// 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;
|
||||
#[cfg(test)]
|
||||
pub(crate) async fn initialize_raw_store_schema(&self) -> ks_core::Result<()> {
|
||||
return crate::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<()> {
|
||||
/// Applies the idempotent minimal Core Solana store schema.
|
||||
#[cfg(test)]
|
||||
pub(crate) 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;
|
||||
return crate::apply_core_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;
|
||||
pub(crate) async fn backend_descriptor(
|
||||
&self,
|
||||
) -> ks_core::Result<crate::StoreBackendDescriptor> {
|
||||
let schema_result = crate::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()),
|
||||
"PostgreSQL",
|
||||
std::option::Option::Some(self.options.masked_connection_descriptor()),
|
||||
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;
|
||||
/// Reads a PostgreSQL health snapshot without exposing backend error details.
|
||||
pub(crate) async fn health_snapshot(&self) -> ks_core::Result<crate::StoreHealthSnapshot> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "health_check", "run PostgreSQL store health check");
|
||||
let health_result = crate::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::option::Option::Some(std::string::String::from(
|
||||
"backend health check succeeded",
|
||||
)),
|
||||
),
|
||||
std::result::Result::Err(error) => crate::StoreHealthSnapshot::new(
|
||||
std::result::Result::Err(_error) => crate::StoreHealthSnapshot::new(
|
||||
"postgres",
|
||||
crate::StoreHealthStatus::Unhealthy,
|
||||
std::option::Option::Some(error.to_string()),
|
||||
std::option::Option::Some(std::string::String::from("backend health check failed")),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
/// 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;
|
||||
pub(crate) async fn migration_snapshot(
|
||||
&self,
|
||||
) -> ks_core::Result<crate::StoreMigrationSnapshot> {
|
||||
let migration_table_result = crate::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(
|
||||
@@ -219,7 +277,7 @@ impl PostgresStore {
|
||||
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",
|
||||
"no migration history table detected; crate-managed schema initialization is active",
|
||||
)),
|
||||
))
|
||||
},
|
||||
@@ -230,132 +288,121 @@ impl PostgresStore {
|
||||
};
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
/// Reads a complete backend-neutral diagnostic snapshot.
|
||||
pub(crate) async fn backend_diagnostics(
|
||||
&self,
|
||||
) -> ks_core::Result<crate::StoreBackendDiagnostics> {
|
||||
let descriptor = match self.backend_descriptor().await {
|
||||
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 {
|
||||
let health = match self.health_snapshot().await {
|
||||
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 {
|
||||
let migrations = match self.migration_snapshot().await {
|
||||
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 {
|
||||
let backend_version = match crate::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 {
|
||||
return std::result::Result::Ok(crate::StoreBackendDiagnostics {
|
||||
descriptor,
|
||||
health,
|
||||
migrations,
|
||||
server_version,
|
||||
backend_version,
|
||||
});
|
||||
}
|
||||
|
||||
/// Lists bounded raw transaction candidates enriched with core and ledger diagnostics.
|
||||
pub async fn replay_transaction_candidates(
|
||||
/// Lists bounded raw transaction candidates enriched with Core and ledger diagnostics.
|
||||
pub(crate) 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;
|
||||
filter: &crate::ReplayTransactionFilter,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::ReplayTransactionCandidate>> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_transaction_candidates", "query PostgreSQL replay transaction candidates");
|
||||
return crate::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(
|
||||
/// Lists bounded program summaries across top-level, inner and reliably linked logs.
|
||||
pub(crate) 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;
|
||||
filter: &crate::ReplayProgramFilter,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::ReplayProgramSummary>> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_program_summaries", "query PostgreSQL replay program summaries");
|
||||
return crate::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(
|
||||
/// Lists bounded mint, owner or account-key summaries from Core facts.
|
||||
pub(crate) 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;
|
||||
filter: &crate::ReplayEntityFilter,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::ReplayEntitySummary>> {
|
||||
tracing::debug!(target: crate::TRACING_TARGET, backend = "postgres", domain = "ks-store.pg", action = "replay_entity_summaries", "query PostgreSQL replay entity summaries");
|
||||
return crate::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(
|
||||
/// Reads diagnostics for raw store resources without changing the schema.
|
||||
pub(crate) async fn raw_resource_diagnostics(
|
||||
&self,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
||||
) -> ks_core::Result<std::vec::Vec<crate::StoreResourceDiagnostics>> {
|
||||
let specs = crate::raw_store_table_diagnostic_specs();
|
||||
return self.table_diagnostics(&specs).await;
|
||||
return self.resource_diagnostics(&specs).await;
|
||||
}
|
||||
|
||||
/// Reads diagnostics for core Solana store tables without changing the schema.
|
||||
pub async fn core_table_diagnostics(
|
||||
/// Reads diagnostics for Core and processing store resources without changing the schema.
|
||||
pub(crate) async fn core_resource_diagnostics(
|
||||
&self,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
||||
) -> ks_core::Result<std::vec::Vec<crate::StoreResourceDiagnostics>> {
|
||||
let specs = crate::core_store_table_diagnostic_specs();
|
||||
return self.table_diagnostics(&specs).await;
|
||||
return self.resource_diagnostics(&specs).await;
|
||||
}
|
||||
|
||||
/// Reads diagnostics for decode and materialization store tables without changing the schema.
|
||||
pub async fn decode_table_diagnostics(
|
||||
/// Reads diagnostics for decode and materialization resources without changing the schema.
|
||||
pub(crate) async fn decode_resource_diagnostics(
|
||||
&self,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
||||
) -> ks_core::Result<std::vec::Vec<crate::StoreResourceDiagnostics>> {
|
||||
let specs = crate::decode_store_table_diagnostic_specs();
|
||||
return self.table_diagnostics(&specs).await;
|
||||
return self.resource_diagnostics(&specs).await;
|
||||
}
|
||||
|
||||
/// Reads diagnostics for every known raw/core/decode Solana store table.
|
||||
pub async fn known_table_diagnostics(
|
||||
/// Reads diagnostics for every known logical store resource.
|
||||
pub(crate) async fn known_resource_diagnostics(
|
||||
&self,
|
||||
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
||||
) -> ks_core::Result<std::vec::Vec<crate::StoreResourceDiagnostics>> {
|
||||
let mut diagnostics = std::vec::Vec::new();
|
||||
let raw_result = self.raw_table_diagnostics().await;
|
||||
let raw_tables = match raw_result {
|
||||
let raw_resources = match self.raw_resource_diagnostics().await {
|
||||
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 {
|
||||
diagnostics.extend(raw_resources);
|
||||
let core_resources = match self.core_resource_diagnostics().await {
|
||||
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 {
|
||||
diagnostics.extend(core_resources);
|
||||
let decode_resources = match self.decode_resource_diagnostics().await {
|
||||
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);
|
||||
}
|
||||
diagnostics.extend(decode_resources);
|
||||
return std::result::Result::Ok(diagnostics);
|
||||
}
|
||||
|
||||
async fn table_diagnostics(
|
||||
async fn resource_diagnostics(
|
||||
&self,
|
||||
specs: &[crate::PostgresTableDiagnosticSpec],
|
||||
) -> ks_core::Result<std::vec::Vec<crate::PostgresTableDiagnostics>> {
|
||||
) -> ks_core::Result<std::vec::Vec<crate::StoreResourceDiagnostics>> {
|
||||
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 {
|
||||
let available = match crate::table_exists(&self.pool, spec.table_name).await {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
||||
};
|
||||
let statistics = if exists {
|
||||
let statistics = if available {
|
||||
let statistics_result =
|
||||
crate::postgres::query::load_table_statistics(&self.pool, spec.table_name)
|
||||
.await;
|
||||
crate::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),
|
||||
@@ -363,11 +410,11 @@ impl PostgresStore {
|
||||
} else {
|
||||
std::option::Option::None
|
||||
};
|
||||
diagnostics.push(crate::PostgresTableDiagnostics {
|
||||
table_name: spec.table_name.to_string(),
|
||||
domain: spec.domain.to_string(),
|
||||
diagnostics.push(crate::StoreResourceDiagnostics {
|
||||
resource_code: spec.resource_code.to_string(),
|
||||
model_code: spec.model_code.to_string(),
|
||||
role: spec.role.to_string(),
|
||||
exists,
|
||||
available,
|
||||
statistics,
|
||||
});
|
||||
}
|
||||
@@ -377,8 +424,7 @@ impl PostgresStore {
|
||||
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;
|
||||
let version_result = crate::load_latest_migration_version(&self.pool).await;
|
||||
return match version_result {
|
||||
std::result::Result::Ok(current_version) => {
|
||||
std::result::Result::Ok(crate::StoreMigrationSnapshot::new(
|
||||
@@ -386,7 +432,7 @@ impl PostgresStore {
|
||||
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",
|
||||
"migration history table detected",
|
||||
)),
|
||||
))
|
||||
},
|
||||
@@ -395,22 +441,18 @@ impl PostgresStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a DSN masked for logs and UI diagnostics.
|
||||
pub fn mask_postgres_dsn(dsn: &str) -> std::string::String {
|
||||
/// Returns a PostgreSQL connection descriptor masked for logs and diagnostics.
|
||||
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);
|
||||
let queryless = 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('?'),
|
||||
)
|
||||
mask_scheme_remainder(scheme, remainder, trimmed_dsn.contains('?'))
|
||||
},
|
||||
std::option::Option::None => crate::postgres::store::mask_plain_dsn(queryless.as_str()),
|
||||
std::option::Option::None => mask_plain_dsn(queryless.as_str()),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -422,7 +464,7 @@ fn strip_query(dsn: &str) -> std::string::String {
|
||||
}
|
||||
|
||||
fn mask_scheme_remainder(scheme: &str, remainder: &str, had_query: bool) -> std::string::String {
|
||||
let suffix = crate::postgres::store::query_suffix(had_query);
|
||||
let suffix = query_suffix(had_query);
|
||||
return match remainder.rsplit_once('@') {
|
||||
std::option::Option::Some((_userinfo, host_path)) => {
|
||||
format!("{scheme}://***:***@{host_path}{suffix}")
|
||||
@@ -448,21 +490,51 @@ fn query_suffix(had_query: bool) -> std::string::String {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
#[test]
|
||||
fn options_reject_empty_database_url() {
|
||||
fn backend_options_reject_empty_database_url() {
|
||||
let result = crate::PostgresStoreOptions::new(" ", 1, 1000, false);
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn options_reject_zero_max_connections() {
|
||||
fn backend_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());
|
||||
fn incomplete_backend_options_do_not_echo_secret_values() {
|
||||
let options = serde_json::json!({
|
||||
"url": "postgres://operator:STORE-SECRET-CANARY@localhost/db",
|
||||
"connect_timeout_ms": 5000,
|
||||
"auto_initialize_schema": true
|
||||
});
|
||||
let error = match crate::PostgresStoreOptions::from_backend_options(&options) {
|
||||
std::result::Result::Ok(_) => panic!("incomplete backend options must be rejected"),
|
||||
std::result::Result::Err(error) => error,
|
||||
};
|
||||
assert!(!error.to_string().contains("STORE-SECRET-CANARY"));
|
||||
assert!(!error.to_string().contains("postgres://"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn opaque_backend_options_are_sanitized_in_summary() {
|
||||
let options = serde_json::json!({
|
||||
"url": "postgres://operator:STORE-SECRET-CANARY@localhost/db",
|
||||
"max_connections": 4,
|
||||
"connect_timeout_ms": 5000,
|
||||
"auto_initialize_schema": true
|
||||
});
|
||||
let parsed = crate::PostgresStoreOptions::from_backend_options(&options);
|
||||
let value = match parsed {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => panic!("backend options must parse: {error}"),
|
||||
};
|
||||
let serialized = match serde_json::to_string(&value.configuration_summary()) {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(error) => panic!("summary must serialize: {error}"),
|
||||
};
|
||||
assert!(!serialized.contains("STORE-SECRET-CANARY"));
|
||||
assert!(!serialized.contains("postgres://"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -471,7 +543,7 @@ mod tests {
|
||||
std::result::Result::Ok(value) => value,
|
||||
std::result::Result::Err(_error) => return,
|
||||
};
|
||||
let _postgres_guard = crate::postgres::test_serial::postgres_test_guard().await;
|
||||
let _postgres_guard = crate::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,
|
||||
@@ -488,25 +560,24 @@ mod tests {
|
||||
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");
|
||||
let masked = super::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");
|
||||
super::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");
|
||||
let masked = super::mask_postgres_dsn("host=localhost password=secret dbname=khadhroony");
|
||||
assert_eq!(masked, "<postgres-dsn-redacted>");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user