use super::system_store::{
ProjectionTaskSummary, SagaStatus, SagaSummary, SystemStoreError, SystemStoreResult,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Placeholder {
Numbered,
Positional,
}
impl Placeholder {
pub fn render(self, n: usize) -> String {
match self {
Self::Numbered => format!("${n}"),
Self::Positional => "?".to_string(),
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SqlDialect {
pub placeholder: Placeholder,
}
#[allow(dead_code)]
impl SqlDialect {
pub const POSTGRES: Self = Self {
placeholder: Placeholder::Numbered,
};
pub const MYSQL: Self = Self {
placeholder: Placeholder::Positional,
};
pub const SQLITE: Self = Self {
placeholder: Placeholder::Positional,
};
pub fn placeholder(self, n: usize) -> String {
self.placeholder.render(n)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WhereClause {
pub where_sql: String,
pub limit_placeholder: String,
pub offset_placeholder: String,
}
pub fn build_eq_where(dialect: SqlDialect, filters: &[(&str, bool)]) -> WhereClause {
let mut clauses: Vec<String> = Vec::new();
let mut bind_index: usize = 0;
for (column, present) in filters {
if *present {
bind_index += 1;
clauses.push(format!("{column} = {}", dialect.placeholder(bind_index)));
}
}
let where_sql = if clauses.is_empty() {
String::new()
} else {
format!("WHERE {}", clauses.join(" AND "))
};
WhereClause {
where_sql,
limit_placeholder: dialect.placeholder(bind_index + 1),
offset_placeholder: dialect.placeholder(bind_index + 2),
}
}
pub fn normalize_limit_offset(limit: i64, offset: i64) -> (i64, i64) {
let limit = if limit <= 0 { 100 } else { limit };
(limit, offset.max(0))
}
pub fn apply_saga_summary_bucket(
summary: &mut SagaSummary,
backend: &'static str,
status: &str,
n: i64,
) -> SystemStoreResult<()> {
match SagaStatus::parse(status) {
Some(SagaStatus::Indeterminate) => summary.indeterminate = n,
Some(SagaStatus::InProgress) => summary.in_progress = n,
Some(SagaStatus::Pending) => summary.pending = n,
Some(SagaStatus::Committed) => summary.committed = n,
Some(SagaStatus::Compensated) => summary.compensated = n,
Some(SagaStatus::Failed) => summary.failed = n,
Some(SagaStatus::InDoubt) => summary.in_doubt = n,
Some(SagaStatus::FailedCompensation) => summary.failed_compensation = n,
Some(SagaStatus::ManualReview) => summary.manual_review = n,
None => {
return Err(SystemStoreError::SchemaMismatch {
backend,
detail: format!(
"unexpected saga status '{status}' (CHECK constraint should prevent this)"
),
});
}
}
Ok(())
}
pub fn apply_projection_summary_bucket(
summary: &mut ProjectionTaskSummary,
backend: &'static str,
status: &str,
n: i64,
) -> SystemStoreResult<()> {
match status {
"PENDING" => summary.pending = n,
"IN_PROGRESS" => summary.in_progress = n,
"COMPLETED" => summary.completed = n,
"FAILED" => summary.failed = n,
"DEAD_LETTER" => summary.dead_letter = n,
_ => {
return Err(SystemStoreError::SchemaMismatch {
backend,
detail: format!(
"unexpected status '{status}' (CHECK constraint should prevent this)"
),
});
}
}
Ok(())
}
pub(super) fn admin_audit_verify_page_size() -> i64 {
std::env::var("UDB_ADMIN_AUDIT_VERIFY_PAGE_SIZE")
.ok()
.and_then(|value| value.parse::<i64>().ok())
.filter(|&n| n > 0)
.unwrap_or(1000)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn numbered_placeholders_match_bind_order() {
let d = SqlDialect::POSTGRES;
let w = build_eq_where(d, &[("tenant_id", true), ("status", true)]);
assert_eq!(w.where_sql, "WHERE tenant_id = $1 AND status = $2");
assert_eq!(w.limit_placeholder, "$3");
assert_eq!(w.offset_placeholder, "$4");
}
#[test]
fn positional_placeholders_use_question_marks() {
let w = build_eq_where(SqlDialect::MYSQL, &[("tenant_id", true), ("status", true)]);
assert_eq!(w.where_sql, "WHERE tenant_id = ? AND status = ?");
assert_eq!(w.limit_placeholder, "?");
assert_eq!(w.offset_placeholder, "?");
}
#[test]
fn absent_filters_are_skipped_and_renumbered() {
let d = SqlDialect::POSTGRES;
let w = build_eq_where(
d,
&[
("tenant_id", false),
("status", true),
("tx_id", false),
("correlation_id", true),
],
);
assert_eq!(w.where_sql, "WHERE status = $1 AND correlation_id = $2");
assert_eq!(w.limit_placeholder, "$3");
assert_eq!(w.offset_placeholder, "$4");
}
#[test]
fn no_filters_yields_empty_where_and_first_placeholders() {
let d = SqlDialect::POSTGRES;
let w = build_eq_where(d, &[("a", false), ("b", false)]);
assert_eq!(w.where_sql, "");
assert_eq!(w.limit_placeholder, "$1");
assert_eq!(w.offset_placeholder, "$2");
let w2 = build_eq_where(SqlDialect::SQLITE, &[("a", false)]);
assert_eq!(w2.where_sql, "");
assert_eq!(w2.limit_placeholder, "?");
assert_eq!(w2.offset_placeholder, "?");
}
#[test]
fn normalize_limit_offset_applies_defaults() {
assert_eq!(normalize_limit_offset(0, -5), (100, 0));
assert_eq!(normalize_limit_offset(-1, 3), (100, 3));
assert_eq!(normalize_limit_offset(50, 10), (50, 10));
}
#[test]
fn saga_summary_bucket_maps_every_status() {
let mut s = SagaSummary::default();
for st in SagaStatus::all() {
apply_saga_summary_bucket(&mut s, "postgres", st.as_str(), 1).unwrap();
}
assert_eq!(s.total() as usize, SagaStatus::all().len());
}
#[test]
fn saga_summary_bucket_rejects_unknown_status() {
let mut s = SagaSummary::default();
let err = apply_saga_summary_bucket(&mut s, "mysql", "bogus", 1).unwrap_err();
assert!(matches!(err, SystemStoreError::SchemaMismatch { .. }));
}
#[test]
fn projection_summary_bucket_maps_every_status() {
let mut s = ProjectionTaskSummary::default();
for st in [
"PENDING",
"IN_PROGRESS",
"COMPLETED",
"FAILED",
"DEAD_LETTER",
] {
apply_projection_summary_bucket(&mut s, "sqlite", st, 1).unwrap();
}
assert_eq!(s.total(), 5);
let err = apply_projection_summary_bucket(&mut s, "sqlite", "WAT", 1).unwrap_err();
assert!(matches!(err, SystemStoreError::SchemaMismatch { .. }));
}
}