mirror of
https://github.com/openobserve/openobserve.git
synced 2026-11-05 16:10:43 +00:00
85d261b5d0
| # | Severity | Category | File | | :--- | :--- | :--- | :--- | | 1 | **High** | SSRF bypass (IPv6) | `ssrf_guard.rs` | | 2 | **High** | SSRF bypass (retry) | `enrichment_table/mod.rs` | | 3 | **High** | SSRF defense-in-depth | `alert.rs` | | 4 | **Critical** | CORS + credentials | `router/mod.rs` | | 5 | **Critical** | SQL Injection | `postgres.rs`, `sqlite.rs` | | 6 | **High** | Privilege escalation (token) | `users.rs` | | 7 | **Medium** | Panic/DoS | `users.rs` | | 8 | **Low** | Integer overflow | `alerts/mod.rs` | | 9 | **Medium** | XSS | `alertSummaryGenerator.ts`, `colorizeQuery.ts` | | 10 | **Medium** | Credential exposure in logs | `validator.rs`, `organization.rs`, `users.rs` | --------- Co-authored-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: Shrinath Rao <shnath@openobserve.ai> Co-authored-by: Prabhat Sharma <hi.prabhat@gmail.com>
1349 lines
44 KiB
Rust
1349 lines
44 KiB
Rust
// Copyright 2026 OpenObserve Inc.
|
|
//
|
|
// This program is free software: you can redistribute it and/or modify
|
|
// it under the terms of the GNU Affero General Public License as published by
|
|
// the Free Software Foundation, either version 3 of the License, or
|
|
// (at your option) any later version.
|
|
//
|
|
// This program is distributed in the hope that it will be useful
|
|
// but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
// GNU Affero General Public License for more details.
|
|
//
|
|
// You should have received a copy of the GNU Affero General Public License
|
|
// along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
use std::{
|
|
collections::HashSet,
|
|
str::FromStr,
|
|
sync::{Arc, LazyLock as Lazy},
|
|
time::Duration,
|
|
};
|
|
|
|
use async_trait::async_trait;
|
|
use bytes::Bytes;
|
|
use config::{
|
|
metrics::{DB_QUERY_NUMS, DB_QUERY_TIME},
|
|
utils::{hash::Sum64, util::zero_or},
|
|
};
|
|
use hashbrown::HashMap;
|
|
use sqlx::{
|
|
Pool, Postgres,
|
|
postgres::{PgConnectOptions, PgPoolOptions},
|
|
};
|
|
use tokio::sync::{OnceCell, mpsc};
|
|
|
|
use super::{DBIndex, IndexStatement};
|
|
use crate::errors::*;
|
|
|
|
pub static CLIENT: Lazy<Pool<Postgres>> = Lazy::new(|| connect(false, false));
|
|
pub static CLIENT_RO: Lazy<Pool<Postgres>> = Lazy::new(|| connect(true, false));
|
|
pub static CLIENT_DDL: Lazy<Pool<Postgres>> = Lazy::new(|| connect(false, true));
|
|
static INDICES: OnceCell<HashSet<DBIndex>> = OnceCell::const_new();
|
|
|
|
fn connect(readonly: bool, ddl: bool) -> Pool<Postgres> {
|
|
let cfg = config::get_config();
|
|
|
|
let mut dsn = match (readonly, ddl) {
|
|
(true, false) => cfg.common.meta_postgres_ro_dsn.clone(),
|
|
(false, false) => cfg.common.meta_postgres_dsn.clone(),
|
|
(_, true) => cfg.common.meta_ddl_dsn.clone(),
|
|
};
|
|
if dsn.is_empty() {
|
|
// default fallback for any case is the original dsn, which is checked
|
|
// in config.rs to be non-empty
|
|
dsn = cfg.common.meta_postgres_dsn.clone();
|
|
}
|
|
let db_opts = PgConnectOptions::from_str(&dsn).expect("postgres connect options create failed");
|
|
|
|
let acquire_timeout = zero_or(cfg.limit.sql_db_connections_acquire_timeout, 30);
|
|
let idle_timeout = zero_or(cfg.limit.sql_db_connections_idle_timeout, 600);
|
|
let max_lifetime = zero_or(cfg.limit.sql_db_connections_max_lifetime, 1800);
|
|
|
|
PgPoolOptions::new()
|
|
.min_connections(cfg.limit.sql_db_connections_min)
|
|
.max_connections(cfg.limit.sql_db_connections_max)
|
|
.acquire_timeout(Duration::from_secs(acquire_timeout))
|
|
.idle_timeout(Some(Duration::from_secs(idle_timeout)))
|
|
.max_lifetime(Some(Duration::from_secs(max_lifetime)))
|
|
.connect_lazy_with(db_opts)
|
|
}
|
|
|
|
async fn cache_indices() -> HashSet<DBIndex> {
|
|
let client = CLIENT.clone();
|
|
DB_QUERY_NUMS
|
|
.with_label_values(&["select", "pg_indexes"])
|
|
.inc();
|
|
let sql = r#"SELECT indexname, tablename FROM pg_indexes;"#;
|
|
let res = sqlx::query_as::<_, (String, String)>(sql)
|
|
.fetch_all(&client)
|
|
.await;
|
|
match res {
|
|
Ok(r) => r
|
|
.into_iter()
|
|
.map(|(name, table)| DBIndex { name, table })
|
|
.collect(),
|
|
Err(_) => HashSet::new(),
|
|
}
|
|
}
|
|
|
|
pub async fn shutdown_ddl_pool() -> Result<()> {
|
|
CLIENT_DDL.close().await;
|
|
Ok(())
|
|
}
|
|
pub struct PostgresDb {}
|
|
|
|
impl PostgresDb {
|
|
pub fn new() -> Self {
|
|
Self {}
|
|
}
|
|
}
|
|
|
|
impl Default for PostgresDb {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl super::Db for PostgresDb {
|
|
async fn create_table(&self) -> Result<()> {
|
|
create_table().await
|
|
}
|
|
|
|
async fn stats(&self) -> Result<super::Stats> {
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let keys_count: i64 = sqlx::query_scalar(r#"SELECT COUNT(*)::BIGINT AS num FROM meta;"#)
|
|
.fetch_one(&pool)
|
|
.await
|
|
.unwrap_or_default();
|
|
Ok(super::Stats {
|
|
bytes_len: 0,
|
|
keys_count,
|
|
})
|
|
}
|
|
|
|
async fn get(&self, key: &str) -> Result<Bytes> {
|
|
let (module, key1, key2) = super::parse_key(key);
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let query = "SELECT value FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 ORDER BY start_dt DESC;";
|
|
let value: String = match sqlx::query_scalar(query)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.fetch_one(&pool)
|
|
.await
|
|
{
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
if let sqlx::Error::RowNotFound = e {
|
|
return Err(Error::from(DbError::KeyNotExists(key.to_string())));
|
|
} else {
|
|
return Err(Error::from(DbError::DBOperError(
|
|
e.to_string(),
|
|
key.to_string(),
|
|
)));
|
|
}
|
|
}
|
|
};
|
|
Ok(Bytes::from(value))
|
|
}
|
|
|
|
async fn put(
|
|
&self,
|
|
key: &str,
|
|
value: Bytes,
|
|
need_watch: bool,
|
|
start_dt: Option<i64>,
|
|
) -> Result<()> {
|
|
let (module, key1, key2) = super::parse_key(key);
|
|
let pool = CLIENT.clone();
|
|
let local_start_dt = start_dt.unwrap_or_default();
|
|
let mut tx = pool.begin().await?;
|
|
DB_QUERY_NUMS.with_label_values(&["insert", "meta"]).inc();
|
|
let start = std::time::Instant::now();
|
|
if let Err(e) = sqlx::query(
|
|
r#"INSERT INTO meta (module, key1, key2, start_dt, value) VALUES ($1, $2, $3, $4, '') ON CONFLICT DO NOTHING;"#
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(local_start_dt)
|
|
.execute(&mut *tx)
|
|
.await
|
|
{
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback put meta error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
// need commit it first to avoid the deadlock of insert and update
|
|
if let Err(e) = tx.commit().await {
|
|
log::error!("[POSTGRES] commit put meta error: {e}");
|
|
return Err(e.into());
|
|
}
|
|
|
|
DB_QUERY_NUMS.with_label_values(&["update", "meta"]).inc();
|
|
let mut tx = pool.begin().await?;
|
|
if let Err(e) = sqlx::query(
|
|
r#"UPDATE meta SET value = $1 WHERE module = $2 AND key1 = $3 AND key2 = $4 AND start_dt = $5;"#
|
|
)
|
|
.bind(String::from_utf8(value.to_vec()).unwrap_or_default())
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(local_start_dt)
|
|
.execute(&mut *tx)
|
|
.await
|
|
{
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback put meta error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
if let Err(e) = tx.commit().await {
|
|
log::error!("[POSTGRES] commit put meta error: {e}");
|
|
return Err(e.into());
|
|
}
|
|
let time = start.elapsed().as_secs_f64();
|
|
|
|
// event watch
|
|
if need_watch {
|
|
let cluster_coordinator = super::get_coordinator().await;
|
|
cluster_coordinator
|
|
.put(key, Bytes::from(""), true, start_dt)
|
|
.await?;
|
|
}
|
|
DB_QUERY_TIME
|
|
.with_label_values(&["put_item", "meta"])
|
|
.observe(time);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_for_update(
|
|
&self,
|
|
key: &str,
|
|
need_watch: bool,
|
|
start_dt: Option<i64>,
|
|
update_fn: Box<super::UpdateFn>,
|
|
) -> Result<()> {
|
|
let (module, key1, key2) = super::parse_key(key);
|
|
let pool = CLIENT.clone();
|
|
let mut tx = pool.begin().await?;
|
|
let lock_key = format!("get_for_update_{key}");
|
|
let lock_id = config::utils::hash::gxhash::new().sum64(&lock_key);
|
|
let lock_id = if lock_id > i64::MAX as u64 {
|
|
(lock_id >> 1) as i64
|
|
} else {
|
|
lock_id as i64
|
|
};
|
|
DB_QUERY_NUMS.with_label_values(&["get_lock", ""]).inc();
|
|
if let Err(e) = sqlx::query("SELECT pg_advisory_xact_lock($1)")
|
|
.bind(lock_id)
|
|
.execute(&mut *tx)
|
|
.await
|
|
{
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
};
|
|
let mut need_watch_dt = 0;
|
|
let row = if let Some(start_dt) = start_dt {
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
match sqlx::query_as::<_,super::MetaRecord>(
|
|
r#"SELECT id, module, key1, key2, start_dt, value FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 AND start_dt = $4;"#
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(start_dt)
|
|
.fetch_one(&mut *tx)
|
|
.await
|
|
{
|
|
Ok(v) => Some(v),
|
|
Err(e) => {
|
|
if e.to_string().contains("no rows returned") {
|
|
None
|
|
} else {
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
match sqlx::query_as::<_,super::MetaRecord>(
|
|
r#"SELECT id, module, key1, key2, start_dt, value FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 ORDER BY start_dt DESC, id DESC;"#
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.fetch_one(&mut *tx)
|
|
.await
|
|
{
|
|
Ok(v) => Some(v),
|
|
Err(e) => {
|
|
if e.to_string().contains("no rows returned") {
|
|
None
|
|
} else {
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
}
|
|
}
|
|
};
|
|
let exist = row.is_some();
|
|
let row_id = row.as_ref().map(|r| r.id);
|
|
let value = row.map(|r| Bytes::from(r.value));
|
|
let (value, new_value) = match update_fn(value) {
|
|
Err(e) => {
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e);
|
|
}
|
|
Ok(None) => {
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Ok(());
|
|
}
|
|
Ok(Some(v)) => v,
|
|
};
|
|
|
|
// update value
|
|
if let Some(value) = value {
|
|
let ret = if exist {
|
|
DB_QUERY_NUMS.with_label_values(&["update", "meta"]).inc();
|
|
sqlx::query(r#"UPDATE meta SET value = $1 WHERE id = $2;"#)
|
|
.bind(String::from_utf8(value.to_vec()).unwrap_or_default())
|
|
.bind(row_id.unwrap())
|
|
.execute(&mut *tx)
|
|
.await
|
|
} else {
|
|
DB_QUERY_NUMS.with_label_values(&["insert", "meta"]).inc();
|
|
sqlx::query(
|
|
r#"INSERT INTO meta (module, key1, key2, start_dt, value) VALUES ($1, $2, $3, $4, $5);"#
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(start_dt.unwrap_or_default())
|
|
.bind(String::from_utf8(value.to_vec()).unwrap_or_default())
|
|
.execute(&mut *tx)
|
|
.await
|
|
};
|
|
if let Err(e) = ret {
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
}
|
|
|
|
// new value
|
|
if let Some((new_key, new_value, new_start_dt)) = new_value {
|
|
need_watch_dt = new_start_dt.unwrap_or_default();
|
|
let (module, key1, key2) = super::parse_key(&new_key);
|
|
DB_QUERY_NUMS.with_label_values(&["insert", "meta"]).inc();
|
|
if let Err(e) = sqlx::query(
|
|
r#"INSERT INTO meta (module, key1, key2, start_dt, value) VALUES ($1, $2, $3, $4, $5);"#
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(new_start_dt.unwrap_or_default())
|
|
.bind(String::from_utf8(new_value.to_vec()).unwrap_or_default())
|
|
.execute(&mut *tx)
|
|
.await
|
|
{
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback get_for_update error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
}
|
|
|
|
if let Err(e) = tx.commit().await {
|
|
log::error!("[POSTGRES] commit get_for_update error: {e}");
|
|
return Err(e.into());
|
|
}
|
|
|
|
// event watch
|
|
if need_watch {
|
|
let start_dt = if need_watch_dt > 0 {
|
|
Some(need_watch_dt)
|
|
} else {
|
|
start_dt
|
|
};
|
|
let cluster_coordinator = super::get_coordinator().await;
|
|
cluster_coordinator
|
|
.put(key, Bytes::from(""), true, start_dt)
|
|
.await?;
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn delete(
|
|
&self,
|
|
key: &str,
|
|
with_prefix: bool,
|
|
need_watch: bool,
|
|
start_dt: Option<i64>,
|
|
) -> Result<()> {
|
|
// event watch
|
|
if need_watch {
|
|
// find all keys then send event
|
|
let items = if with_prefix {
|
|
self.list_keys(key).await?
|
|
} else {
|
|
vec![key.to_string()]
|
|
};
|
|
let cluster_coordinator = super::get_coordinator().await;
|
|
tokio::task::spawn(async move {
|
|
for key in items {
|
|
if let Err(e) = cluster_coordinator
|
|
.delete(&key, false, true, start_dt)
|
|
.await
|
|
{
|
|
log::error!("[POSTGRES] send event error: {e}");
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
let (module, key1, key2) = super::parse_key(key);
|
|
let pool = CLIENT.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["delete", "meta"]).inc();
|
|
|
|
if with_prefix {
|
|
if key1.is_empty() {
|
|
// module = $1 [, start_dt = $2]
|
|
if let Some(dt) = start_dt {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND start_dt = $2")
|
|
.bind(&module)
|
|
.bind(dt)
|
|
.execute(&pool)
|
|
.await?;
|
|
} else {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1")
|
|
.bind(&module)
|
|
.execute(&pool)
|
|
.await?;
|
|
}
|
|
} else if key2.is_empty() {
|
|
// module = $1, key1 = $2 [, start_dt = $3]
|
|
if let Some(dt) = start_dt {
|
|
sqlx::query(
|
|
"DELETE FROM meta WHERE module = $1 AND key1 = $2 AND start_dt = $3",
|
|
)
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(dt)
|
|
.execute(&pool)
|
|
.await?;
|
|
} else {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2")
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.execute(&pool)
|
|
.await?;
|
|
}
|
|
} else {
|
|
// module = $1, key1 = $2, key2 = $3, key2/% = $4 [, start_dt = $5]
|
|
if let Some(dt) = start_dt {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4) AND start_dt = $5")
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(format!("{}/%", key2))
|
|
.bind(dt)
|
|
.execute(&pool)
|
|
.await?;
|
|
} else {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND (key2 = $3 OR key2 LIKE $4)")
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(format!("{}/%", key2))
|
|
.execute(&pool)
|
|
.await?;
|
|
}
|
|
}
|
|
} else {
|
|
// module = $1, key1 = $2, key2 = $3 [, start_dt = $4]
|
|
if let Some(dt) = start_dt {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3 AND start_dt = $4")
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.bind(dt)
|
|
.execute(&pool)
|
|
.await?;
|
|
} else {
|
|
sqlx::query("DELETE FROM meta WHERE module = $1 AND key1 = $2 AND key2 = $3")
|
|
.bind(&module)
|
|
.bind(&key1)
|
|
.bind(&key2)
|
|
.execute(&pool)
|
|
.await?;
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn list(&self, prefix: &str) -> Result<HashMap<String, Bytes>> {
|
|
let (module, key1, key2) = super::parse_key(prefix);
|
|
let mut sql =
|
|
"SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string();
|
|
let mut params = Vec::new();
|
|
|
|
if !module.is_empty() {
|
|
sql.push_str(" AND module = $1");
|
|
params.push(module);
|
|
}
|
|
if !key1.is_empty() {
|
|
sql.push_str(&format!(" AND key1 = ${}", params.len() + 1));
|
|
params.push(key1);
|
|
}
|
|
if !key2.is_empty() {
|
|
sql.push_str(&format!(
|
|
" AND (key2 = ${} OR key2 LIKE ${})",
|
|
params.len() + 1,
|
|
params.len() + 2
|
|
));
|
|
let key2_prefix = format!("{}/%", key2);
|
|
params.push(key2);
|
|
params.push(key2_prefix);
|
|
}
|
|
sql.push_str(" ORDER BY start_dt ASC");
|
|
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql);
|
|
for p in params {
|
|
query = query.bind(p);
|
|
}
|
|
let ret = query.fetch_all(&pool).await?;
|
|
Ok(ret
|
|
.into_iter()
|
|
.map(|r| {
|
|
(
|
|
super::build_key(&r.module, &r.key1, &r.key2, r.start_dt),
|
|
Bytes::from(r.value),
|
|
)
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
async fn list_keys(&self, prefix: &str) -> Result<Vec<String>> {
|
|
let (module, key1, key2) = super::parse_key(prefix);
|
|
let mut sql =
|
|
"SELECT id, module, key1, key2, start_dt, '' AS value FROM meta WHERE 1=1 ".to_string();
|
|
let mut params = Vec::new();
|
|
|
|
if !module.is_empty() {
|
|
sql.push_str(" AND module = $1");
|
|
params.push(module);
|
|
}
|
|
if !key1.is_empty() {
|
|
sql.push_str(&format!(" AND key1 = ${}", params.len() + 1));
|
|
params.push(key1);
|
|
}
|
|
if !key2.is_empty() {
|
|
sql.push_str(&format!(
|
|
" AND (key2 = ${} OR key2 LIKE ${})",
|
|
params.len() + 1,
|
|
params.len() + 2
|
|
));
|
|
let key2_prefix = format!("{}/%", key2);
|
|
params.push(key2);
|
|
params.push(key2_prefix);
|
|
}
|
|
sql.push_str(" ORDER BY start_dt ASC");
|
|
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql);
|
|
for p in params {
|
|
query = query.bind(p);
|
|
}
|
|
let ret = query.fetch_all(&pool).await?;
|
|
Ok(ret
|
|
.into_iter()
|
|
.map(|r| format!("/{}/{}/{}", r.module, r.key1, r.key2))
|
|
.collect())
|
|
}
|
|
|
|
async fn list_values(&self, prefix: &str) -> Result<Vec<Bytes>> {
|
|
let mut items = self.list(prefix).await?;
|
|
let mut keys = items.keys().map(|k| k.to_string()).collect::<Vec<_>>();
|
|
keys.sort();
|
|
Ok(keys
|
|
.into_iter()
|
|
.map(|k| items.remove(&k).unwrap())
|
|
.collect())
|
|
}
|
|
|
|
async fn list_values_by_start_dt(
|
|
&self,
|
|
prefix: &str,
|
|
start_dt: Option<(i64, i64)>,
|
|
) -> Result<Vec<(i64, Bytes)>> {
|
|
if start_dt.is_none() || start_dt == Some((0, 0)) {
|
|
let vals = self.list_values(prefix).await?;
|
|
return Ok(vals.into_iter().map(|v| (0, v)).collect());
|
|
}
|
|
|
|
let (min_dt, max_dt) = start_dt.unwrap();
|
|
let (sql, str_params) = build_list_by_start_dt_sql(prefix, min_dt, max_dt);
|
|
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let mut query = sqlx::query_as::<_, super::MetaRecord>(&sql);
|
|
for p in str_params {
|
|
query = query.bind(p);
|
|
}
|
|
// min_dt and max_dt are i64 — bind them directly so PostgreSQL receives
|
|
// them as BIGINT rather than TEXT, avoiding "operator does not exist: bigint >= text".
|
|
query = query.bind(min_dt);
|
|
query = query.bind(max_dt);
|
|
let ret = query.fetch_all(&pool).await?;
|
|
Ok(ret
|
|
.into_iter()
|
|
.map(|r| (r.start_dt, Bytes::from(r.value)))
|
|
.collect())
|
|
}
|
|
|
|
async fn count(&self, prefix: &str) -> Result<i64> {
|
|
let (module, key1, key2) = super::parse_key(prefix);
|
|
let mut sql = "SELECT COUNT(*) AS num FROM meta WHERE 1=1".to_string();
|
|
let mut params = Vec::new();
|
|
|
|
if !module.is_empty() {
|
|
sql.push_str(" AND module = $1");
|
|
params.push(module);
|
|
}
|
|
if !key1.is_empty() {
|
|
sql.push_str(&format!(" AND key1 = ${}", params.len() + 1));
|
|
params.push(key1);
|
|
}
|
|
if !key2.is_empty() {
|
|
sql.push_str(&format!(
|
|
" AND (key2 = ${} OR key2 LIKE ${})",
|
|
params.len() + 1,
|
|
params.len() + 2
|
|
));
|
|
let key2_prefix = format!("{}/%", key2);
|
|
params.push(key2);
|
|
params.push(key2_prefix);
|
|
}
|
|
let pool = CLIENT_RO.clone();
|
|
DB_QUERY_NUMS.with_label_values(&["select", "meta"]).inc();
|
|
let mut query = sqlx::query_scalar::<_, i64>(&sql);
|
|
for p in params {
|
|
query = query.bind(p);
|
|
}
|
|
let count: i64 = query.fetch_one(&pool).await?;
|
|
Ok(count)
|
|
}
|
|
|
|
async fn watch(&self, _prefix: &str) -> Result<Arc<mpsc::Receiver<super::Event>>> {
|
|
Err(Error::NotImplemented)
|
|
}
|
|
|
|
async fn close(&self) -> Result<()> {
|
|
Ok(())
|
|
}
|
|
|
|
async fn add_start_dt_column(&self) -> Result<()> {
|
|
create_meta_backup().await?;
|
|
add_start_dt_column().await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
pub async fn create_table() -> Result<()> {
|
|
let pool = CLIENT_DDL.clone();
|
|
|
|
DB_QUERY_NUMS.with_label_values(&["create", "meta"]).inc();
|
|
// create table
|
|
_ = sqlx::query(
|
|
r#"
|
|
CREATE TABLE IF NOT EXISTS meta
|
|
(
|
|
id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
|
module VARCHAR(100) not null,
|
|
key1 VARCHAR(256) not null,
|
|
key2 VARCHAR(256) not null,
|
|
start_dt BIGINT not null,
|
|
value TEXT not null
|
|
);
|
|
"#,
|
|
)
|
|
.execute(&pool)
|
|
.await?;
|
|
|
|
// create start_dt column for old version <= 0.9.2
|
|
DB_QUERY_NUMS
|
|
.with_label_values(&["select", "information_schema.columns"])
|
|
.inc();
|
|
let has_start_dt = sqlx::query_scalar::<_,i64>("SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='meta' AND column_name='start_dt';")
|
|
.fetch_one(&pool)
|
|
.await?;
|
|
if has_start_dt == 0 {
|
|
add_start_dt_column().await?;
|
|
}
|
|
|
|
// create table index
|
|
create_index(IndexStatement::new(
|
|
"meta_module_idx",
|
|
"meta",
|
|
false,
|
|
&["module"],
|
|
))
|
|
.await?;
|
|
create_index(IndexStatement::new(
|
|
"meta_module_key1_idx",
|
|
"meta",
|
|
false,
|
|
&["module", "key1"],
|
|
))
|
|
.await?;
|
|
create_index(IndexStatement::new(
|
|
"meta_module_start_dt_idx",
|
|
"meta",
|
|
true,
|
|
&["module", "key1", "key2", "start_dt"],
|
|
))
|
|
.await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn add_start_dt_column() -> Result<()> {
|
|
DB_QUERY_NUMS.with_label_values(&["alter", "meta"]).inc();
|
|
add_column("meta", "start_dt", "BIGINT NOT NULL DEFAULT 0").await?;
|
|
|
|
// Proceed to drop the index if it exists and create a new one if it does not exist
|
|
create_index(IndexStatement::new(
|
|
"meta_module_start_dt_idx",
|
|
"meta",
|
|
true,
|
|
&["module", "key1", "key2", "start_dt"],
|
|
))
|
|
.await?;
|
|
delete_index("meta_module_key2_idx", "meta").await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn create_meta_backup() -> Result<()> {
|
|
let pool = CLIENT_DDL.clone();
|
|
let mut tx = pool.begin().await?;
|
|
DB_QUERY_NUMS
|
|
.with_label_values(&["create", "meta_backup_20240330"])
|
|
.inc();
|
|
if let Err(e) =
|
|
sqlx::query(r#"CREATE TABLE IF NOT EXISTS meta_backup_20240330 AS SELECT * FROM meta;"#)
|
|
.execute(&mut *tx)
|
|
.await
|
|
{
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] rollback create table meta_backup_20240330 error: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
if let Err(e) = tx.commit().await {
|
|
log::error!("[POSTGRES] commit create table meta_backup_20240330 error: {e}");
|
|
return Err(e.into());
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn create_index(index: IndexStatement<'_>) -> Result<()> {
|
|
let client = CLIENT_DDL.clone();
|
|
let indices = INDICES.get_or_init(cache_indices).await;
|
|
if indices.contains(&DBIndex {
|
|
name: index.idx_name.into(),
|
|
table: index.table.into(),
|
|
}) {
|
|
return Ok(());
|
|
}
|
|
let unique_str = if index.unique { "UNIQUE" } else { "" };
|
|
log::info!(
|
|
"[POSTGRES] creating index {} on table {}",
|
|
index.idx_name,
|
|
index.table
|
|
);
|
|
DB_QUERY_NUMS
|
|
.with_label_values(&["create", index.table])
|
|
.inc();
|
|
let sql = format!(
|
|
"CREATE {} INDEX IF NOT EXISTS {} ON {} ({});",
|
|
unique_str,
|
|
index.idx_name,
|
|
index.table,
|
|
index.fields.join(",")
|
|
);
|
|
|
|
let start = std::time::Instant::now();
|
|
sqlx::query(&sql).execute(&client).await?;
|
|
let time = start.elapsed().as_secs_f64();
|
|
DB_QUERY_TIME
|
|
.with_label_values(&["create_index", index.table])
|
|
.observe(time);
|
|
log::info!("[POSTGRES] index {} created successfully", index.idx_name);
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn delete_index(idx_name: &str, table: &str) -> Result<()> {
|
|
let client = CLIENT_DDL.clone();
|
|
let indices = INDICES.get_or_init(cache_indices).await;
|
|
if !indices.contains(&DBIndex {
|
|
name: idx_name.into(),
|
|
table: table.into(),
|
|
}) {
|
|
return Ok(());
|
|
}
|
|
log::info!("[POSTGRES] deleting index {idx_name} on table {table}");
|
|
let sql = format!("DROP INDEX IF EXISTS {idx_name};");
|
|
DB_QUERY_NUMS.with_label_values(&["drop", table]).inc();
|
|
let start = std::time::Instant::now();
|
|
sqlx::query(&sql).execute(&client).await?;
|
|
let time = start.elapsed().as_secs_f64();
|
|
DB_QUERY_TIME
|
|
.with_label_values(&["drop_index", table])
|
|
.observe(time);
|
|
log::info!("[POSTGRES] index {idx_name} deleted successfully");
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn add_column(table: &str, column: &str, data_type: &str) -> Result<()> {
|
|
let pool = CLIENT_DDL.clone();
|
|
let check_sql = format!(
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';"
|
|
);
|
|
let has_column = sqlx::query_scalar::<_, i64>(&check_sql)
|
|
.fetch_one(&pool)
|
|
.await?;
|
|
if has_column > 0 {
|
|
return Ok(());
|
|
}
|
|
|
|
let alert_sql = format!("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {column} {data_type};");
|
|
let mut tx = pool.begin().await?;
|
|
if let Err(e) = sqlx::query(&alert_sql).execute(&mut *tx).await {
|
|
log::error!("[POSTGRES] Error in adding column {column}: {e}");
|
|
if let Err(e) = tx.rollback().await {
|
|
log::error!("[POSTGRES] Error in rolling back transaction: {e}");
|
|
}
|
|
return Err(e.into());
|
|
}
|
|
if let Err(e) = tx.commit().await {
|
|
log::info!("[POSTGRES] Error in committing transaction: {e}");
|
|
return Err(e.into());
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn drop_column(table: &str, column: &str) -> Result<()> {
|
|
let pool = CLIENT_DDL.clone();
|
|
let check_sql = format!(
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';"
|
|
);
|
|
let has_column = sqlx::query_scalar::<_, i64>(&check_sql)
|
|
.fetch_one(&pool)
|
|
.await?;
|
|
if has_column > 0 {
|
|
let alert_sql = format!("ALTER TABLE {table} DROP COLUMN {column};");
|
|
sqlx::query(&alert_sql).execute(&pool).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Builds the SQL string and string-typed bind parameters for `list_values_by_start_dt`.
|
|
///
|
|
/// Returns `(sql, str_params)` where `str_params` contains the string-typed positional
|
|
/// parameters (module / key1 / key2). The caller must then bind `min_dt` and `max_dt`
|
|
/// as `i64` **after** the string params so that PostgreSQL sees them as BIGINT, not TEXT.
|
|
///
|
|
/// Keeping this logic in a standalone function makes the SQL generation unit-testable
|
|
/// without requiring a live database connection.
|
|
pub(crate) fn build_list_by_start_dt_sql(
|
|
prefix: &str,
|
|
min_dt: i64,
|
|
max_dt: i64,
|
|
) -> (String, Vec<String>) {
|
|
let (module, key1, key2) = super::parse_key(prefix);
|
|
let mut sql = "SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1".to_string();
|
|
let mut params: Vec<String> = Vec::new();
|
|
|
|
if !module.is_empty() {
|
|
sql.push_str(" AND module = $1");
|
|
params.push(module);
|
|
}
|
|
if !key1.is_empty() {
|
|
sql.push_str(&format!(" AND key1 = ${}", params.len() + 1));
|
|
params.push(key1);
|
|
}
|
|
if !key2.is_empty() {
|
|
sql.push_str(&format!(
|
|
" AND (key2 = ${} OR key2 LIKE ${})",
|
|
params.len() + 1,
|
|
params.len() + 2
|
|
));
|
|
let key2_prefix = format!("{key2}/%");
|
|
params.push(key2);
|
|
params.push(key2_prefix);
|
|
}
|
|
// min_dt / max_dt are intentionally NOT pushed into params here.
|
|
// The caller binds them as i64 after the string-params loop so that the
|
|
// PostgreSQL driver receives them as BIGINT, not TEXT.
|
|
let n = params.len();
|
|
sql.push_str(&format!(
|
|
" AND start_dt >= ${} AND start_dt <= ${}",
|
|
n + 1,
|
|
n + 2
|
|
));
|
|
// Suppress unused-variable warning: min_dt / max_dt are consumed by the caller.
|
|
let _ = (min_dt, max_dt);
|
|
sql.push_str(" ORDER BY start_dt ASC");
|
|
|
|
(sql, params)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_postgres_db_new() {
|
|
let db = PostgresDb::new();
|
|
assert_eq!(std::mem::size_of_val(&db), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_postgres_db_default() {
|
|
let db = PostgresDb::default();
|
|
assert_eq!(std::mem::size_of_val(&db), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_full() {
|
|
let key = "/module/key1/key2";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "key1");
|
|
assert_eq!(k2, "key2");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_no_leading_slash() {
|
|
let key = "module/key1/key2";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "key1");
|
|
assert_eq!(k2, "key2");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_module_only() {
|
|
let key = "/module";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "");
|
|
assert_eq!(k2, "");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_module_and_key1() {
|
|
let key = "/module/key1";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "key1");
|
|
assert_eq!(k2, "");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_with_slashes_in_key2() {
|
|
let key = "/module/key1/key2/subkey/subsubkey";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "key1");
|
|
assert_eq!(k2, "key2/subkey/subsubkey");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_empty() {
|
|
let key = "";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "");
|
|
assert_eq!(k1, "");
|
|
assert_eq!(k2, "");
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_key_without_start_dt() {
|
|
let key = super::super::build_key("module", "key1", "key2", 0);
|
|
assert_eq!(key, "/module/key1/key2");
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_key_with_start_dt() {
|
|
let key = super::super::build_key("module", "key1", "key2", 1234567890);
|
|
assert_eq!(key, "/module/key1/key2/1234567890");
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_key_empty_parts() {
|
|
let key = super::super::build_key("", "", "", 0);
|
|
assert_eq!(key, "//");
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_key_with_special_chars() {
|
|
let key = super::super::build_key("mod-ule", "key_1", "key.2", 0);
|
|
assert_eq!(key, "/mod-ule/key_1/key.2");
|
|
}
|
|
|
|
#[test]
|
|
fn test_index_statement_new() {
|
|
let stmt = IndexStatement::new("test_idx", "test_table", false, &["col1", "col2"]);
|
|
assert_eq!(stmt.idx_name, "test_idx");
|
|
assert_eq!(stmt.table, "test_table");
|
|
assert!(!stmt.unique);
|
|
assert_eq!(stmt.fields, vec!["col1", "col2"]);
|
|
}
|
|
|
|
#[test]
|
|
fn test_index_statement_unique() {
|
|
let stmt = IndexStatement::new("unique_idx", "table", true, &["id"]);
|
|
assert_eq!(stmt.idx_name, "unique_idx");
|
|
assert_eq!(stmt.table, "table");
|
|
assert!(stmt.unique);
|
|
assert_eq!(stmt.fields, vec!["id"]);
|
|
}
|
|
|
|
#[test]
|
|
fn test_db_index_hash_same() {
|
|
use std::collections::HashSet;
|
|
|
|
let idx1 = DBIndex {
|
|
name: "test_idx".to_string(),
|
|
table: "test_table".to_string(),
|
|
};
|
|
let idx2 = DBIndex {
|
|
name: "test_idx".to_string(),
|
|
table: "test_table".to_string(),
|
|
};
|
|
|
|
let mut set = HashSet::new();
|
|
set.insert(idx1);
|
|
assert!(set.contains(&idx2));
|
|
}
|
|
|
|
#[test]
|
|
fn test_db_index_hash_different_name() {
|
|
use std::collections::HashSet;
|
|
|
|
let idx1 = DBIndex {
|
|
name: "idx1".to_string(),
|
|
table: "table".to_string(),
|
|
};
|
|
let idx2 = DBIndex {
|
|
name: "idx2".to_string(),
|
|
table: "table".to_string(),
|
|
};
|
|
|
|
let mut set = HashSet::new();
|
|
set.insert(idx1);
|
|
assert!(!set.contains(&idx2));
|
|
}
|
|
|
|
#[test]
|
|
fn test_add_column_check_sql_format() {
|
|
let table = "test_table";
|
|
let column = "new_column";
|
|
|
|
let check_sql = format!(
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';"
|
|
);
|
|
assert_eq!(
|
|
check_sql,
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='test_table' AND column_name='new_column';"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_add_column_alter_sql_format() {
|
|
let table = "test_table";
|
|
let column = "new_column";
|
|
let data_type = "VARCHAR(255) NOT NULL DEFAULT ''";
|
|
|
|
// PostgreSQL supports IF NOT EXISTS
|
|
let alert_sql =
|
|
format!("ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {column} {data_type};");
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE test_table ADD COLUMN IF NOT EXISTS new_column VARCHAR(255) NOT NULL DEFAULT '';"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_drop_column_check_sql_format() {
|
|
let table = "test_table";
|
|
let column = "old_column";
|
|
|
|
let check_sql = format!(
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='{table}' AND column_name='{column}';"
|
|
);
|
|
assert_eq!(
|
|
check_sql,
|
|
"SELECT count(*) FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name='test_table' AND column_name='old_column';"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_drop_column_alter_sql_format() {
|
|
let table = "test_table";
|
|
let column = "old_column";
|
|
|
|
let alert_sql = format!("ALTER TABLE {table} DROP COLUMN {column};");
|
|
assert_eq!(alert_sql, "ALTER TABLE test_table DROP COLUMN old_column;");
|
|
}
|
|
|
|
#[test]
|
|
fn test_add_column_sql_with_various_postgres_types() {
|
|
let table = "users";
|
|
|
|
// BIGINT type
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "big_id", "BIGINT NOT NULL DEFAULT 0"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS big_id BIGINT NOT NULL DEFAULT 0;"
|
|
);
|
|
|
|
// VARCHAR type
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "email", "VARCHAR(255)"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS email VARCHAR(255);"
|
|
);
|
|
|
|
// TEXT type
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "description", "TEXT"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS description TEXT;"
|
|
);
|
|
|
|
// TIMESTAMP type
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "created_at", "TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP;"
|
|
);
|
|
|
|
// BOOLEAN type
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "is_active", "BOOLEAN NOT NULL DEFAULT TRUE"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS is_active BOOLEAN NOT NULL DEFAULT TRUE;"
|
|
);
|
|
|
|
// JSONB type (PostgreSQL specific)
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "metadata", "JSONB"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS metadata JSONB;"
|
|
);
|
|
|
|
// UUID type (PostgreSQL specific)
|
|
let alert_sql = format!(
|
|
"ALTER TABLE {} ADD COLUMN IF NOT EXISTS {} {};",
|
|
table, "uuid", "UUID"
|
|
);
|
|
assert_eq!(
|
|
alert_sql,
|
|
"ALTER TABLE users ADD COLUMN IF NOT EXISTS uuid UUID;"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_sql_quote_escaping() {
|
|
let key = "key'with'quotes";
|
|
let escaped = key.replace('\'', "''");
|
|
assert_eq!(escaped, "key''with''quotes");
|
|
}
|
|
|
|
#[test]
|
|
fn test_sql_like_pattern_construction() {
|
|
let key2 = "prefix";
|
|
let pattern = format!("{key2}/%");
|
|
assert_eq!(pattern, "prefix/%");
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_key_roundtrip() {
|
|
let original_parts = ("module", "key1", "key2");
|
|
let key = super::super::build_key(original_parts.0, original_parts.1, original_parts.2, 0);
|
|
let (module, k1, k2) = super::super::parse_key(&key);
|
|
assert_eq!(module, original_parts.0);
|
|
assert_eq!(k1, original_parts.1);
|
|
assert_eq!(k2, original_parts.2);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_with_numbers() {
|
|
let key = "/123/456/789";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "123");
|
|
assert_eq!(k1, "456");
|
|
assert_eq!(k2, "789");
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_with_hyphens_underscores() {
|
|
let key = "/test-module/test_key1/test-key-2";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "test-module");
|
|
assert_eq!(k1, "test_key1");
|
|
assert_eq!(k2, "test-key-2");
|
|
}
|
|
|
|
#[test]
|
|
fn test_index_statement_empty_fields() {
|
|
let stmt = IndexStatement::new("empty_idx", "table", false, &[]);
|
|
assert_eq!(stmt.fields.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_key_trailing_slash() {
|
|
let key = "/module/key1/key2/";
|
|
let (module, k1, k2) = super::super::parse_key(key);
|
|
assert_eq!(module, "module");
|
|
assert_eq!(k1, "key1");
|
|
assert!(k2.starts_with("key2"));
|
|
}
|
|
|
|
// ── build_list_by_start_dt_sql unit tests ─────────────────────────────────
|
|
// These tests verify that the SQL and parameter vector produced by the helper
|
|
// are correct for every combination of module / key1 / key2 presence, and that
|
|
// min_dt / max_dt placeholder indices are always one past the last string param.
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_full_prefix() {
|
|
let (sql, params) = build_list_by_start_dt_sql("/module/key1/key2", 100, 200);
|
|
// 3 string params: module=$1, key1=$2, key2=$3 / key2-prefix=$4
|
|
assert_eq!(params.len(), 4);
|
|
assert_eq!(params[0], "module");
|
|
assert_eq!(params[1], "key1");
|
|
assert_eq!(params[2], "key2");
|
|
assert_eq!(params[3], "key2/%");
|
|
assert!(
|
|
sql.contains("AND module = $1"),
|
|
"module param missing: {sql}"
|
|
);
|
|
assert!(sql.contains("AND key1 = $2"), "key1 param missing: {sql}");
|
|
assert!(
|
|
sql.contains("AND (key2 = $3 OR key2 LIKE $4)"),
|
|
"key2 param missing: {sql}"
|
|
);
|
|
// min_dt=$5, max_dt=$6 (one past the 4 string params)
|
|
assert!(
|
|
sql.contains("AND start_dt >= $5 AND start_dt <= $6"),
|
|
"start_dt bounds wrong: {sql}"
|
|
);
|
|
assert!(
|
|
sql.ends_with("ORDER BY start_dt ASC"),
|
|
"missing ORDER BY: {sql}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_module_only() {
|
|
let (sql, params) = build_list_by_start_dt_sql("/module", 1, 999);
|
|
assert_eq!(params.len(), 1);
|
|
assert_eq!(params[0], "module");
|
|
assert!(sql.contains("AND module = $1"));
|
|
// min_dt=$2, max_dt=$3
|
|
assert!(
|
|
sql.contains("AND start_dt >= $2 AND start_dt <= $3"),
|
|
"start_dt bounds wrong: {sql}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_module_and_key1() {
|
|
let (sql, params) = build_list_by_start_dt_sql("/module/key1", 10, 20);
|
|
assert_eq!(params.len(), 2);
|
|
assert_eq!(params[0], "module");
|
|
assert_eq!(params[1], "key1");
|
|
assert!(sql.contains("AND module = $1"));
|
|
assert!(sql.contains("AND key1 = $2"));
|
|
// min_dt=$3, max_dt=$4
|
|
assert!(
|
|
sql.contains("AND start_dt >= $3 AND start_dt <= $4"),
|
|
"start_dt bounds wrong: {sql}"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_empty_prefix() {
|
|
let (sql, params) = build_list_by_start_dt_sql("", 0, 9999);
|
|
// No string params — min_dt=$1, max_dt=$2
|
|
assert!(params.is_empty());
|
|
assert!(
|
|
sql.contains("AND start_dt >= $1 AND start_dt <= $2"),
|
|
"start_dt bounds wrong for empty prefix: {sql}"
|
|
);
|
|
assert!(sql.contains("WHERE 1=1"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_key2_prefix_pattern() {
|
|
let (sql, params) = build_list_by_start_dt_sql("/m/k1/k2", 0, 0);
|
|
// key2 LIKE pattern must end with /%
|
|
assert_eq!(params[3], "k2/%");
|
|
assert!(sql.contains("OR key2 LIKE $4"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_build_list_by_start_dt_sql_starts_with_base_select() {
|
|
let (sql, _) = build_list_by_start_dt_sql("/m/k1/k2", 1, 2);
|
|
assert!(
|
|
sql.starts_with("SELECT id, module, key1, key2, start_dt, value FROM meta WHERE 1=1"),
|
|
"unexpected SQL start: {sql}"
|
|
);
|
|
}
|
|
}
|