use super::{
BackoffSpec, CheckConstraint, ColumnType, ConsumerSpec, IndexDefinition, IndexMethod,
JobConfig, MigrationHistory, ProvidedExtension, ScheduleDefinition, ScheduleSpec, SearchSpec,
SqlMigration, StreamDefinition, SubscriptionDefinition,
};
use crate::ext::{callers, name_violation};
use crate::nats_names::{is_valid_consumer_name, is_valid_stream_name, pattern_violation};
use std::collections::BTreeSet;
use std::fmt::Display;
use std::ops::RangeInclusive;
pub const IDENTIFIER_MAX: usize = 63;
const RESERVED_SCHEMAS: [&str; 2] = ["public", "information_schema"];
const RESERVED_SCHEMA_PREFIXES: [&str; 2] = ["pg_", "_cufflink"];
const DESCRIPTION_MAX: usize = 4096;
fn within<T: PartialOrd + Display>(
path: &str,
field: &str,
value: Option<T>,
range: &RangeInclusive<T>,
) -> Result<(), String> {
match value {
Some(value) if !range.contains(&value) => Err(format!(
"{path}.{field}: {value} is outside {}..={}",
range.start(),
range.end()
)),
_ => Ok(()),
}
}
fn positive(path: &str, field: &str, value: Option<i64>) -> Result<(), String> {
match value {
Some(value) if value < 1 => Err(format!("{path}.{field}: {value} is not positive")),
_ => Ok(()),
}
}
fn distinct<'a>(
path: &str,
what: &str,
values: impl IntoIterator<Item = &'a str>,
) -> Result<(), String> {
let mut seen = BTreeSet::new();
match values.into_iter().find(|value| !seen.insert(*value)) {
Some(repeated) => Err(listed_twice(path, what, repeated)),
None => Ok(()),
}
}
fn listed_twice(path: &str, what: &str, repeated: &str) -> String {
format!("{path}: {what} '{repeated}' is listed twice")
}
pub fn is_lowercase_identifier(name: &str) -> bool {
let mut bytes = name.bytes();
name.len() <= IDENTIFIER_MAX
&& bytes
.next()
.is_some_and(|b| b.is_ascii_lowercase() || b == b'_')
&& bytes.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
}
pub fn is_identifier(name: &str) -> bool {
let mut bytes = name.bytes();
name.len() <= IDENTIFIER_MAX
&& bytes
.next()
.is_some_and(|b| b.is_ascii_alphabetic() || b == b'_')
&& bytes.all(|b| b.is_ascii_alphanumeric() || b == b'_')
}
fn identifier(path: &str, name: &str) -> Result<(), String> {
if is_identifier(name) {
Ok(())
} else {
Err(format!(
"{path}: '{name}' is not an identifier (ASCII letters, digits and '_', \
not starting with a digit, at most {IDENTIFIER_MAX} characters)"
))
}
}
pub fn folded(name: &str) -> String {
name.to_ascii_lowercase()
}
pub fn folded_ident(name: &str) -> String {
format!("\"{}\"", folded(name).replace('"', "\"\""))
}
pub fn is_reserved_schema(name: &str) -> bool {
let name = folded(name);
RESERVED_SCHEMAS.contains(&name.as_str())
|| RESERVED_SCHEMA_PREFIXES
.iter()
.any(|prefix| name.starts_with(prefix))
}
pub fn validate_schema_name(path: &str, name: &str) -> Result<(), String> {
if !is_lowercase_identifier(name) {
return Err(format!(
"{path}: '{name}' is not a schema name (lowercase letters, digits and '_', \
not starting with a digit, at most {IDENTIFIER_MAX} characters)"
));
}
if is_reserved_schema(name) {
return Err(format!("{path}: schema '{name}' is reserved"));
}
Ok(())
}
pub fn validate_sql_fragment(path: &str, sql: &str) -> Result<(), String> {
if sql.trim().is_empty() {
return Err(format!("{path}: is empty"));
}
if let Some(found) = [";", "--", "/*", "*/"]
.into_iter()
.find(|token| sql.contains(token))
{
return Err(format!("{path}: contains '{found}'"));
}
match depth(sql) {
Ok(0) => Ok(()),
Ok(_) => Err(format!("{path}: has unbalanced parentheses")),
Err(reason) => Err(format!("{path}: {reason}")),
}
}
fn depth(sql: &str) -> Result<u32, &'static str> {
let mut chars = sql.chars();
let mut depth = 0u32;
while let Some(c) = chars.next() {
depth = match c {
'(' => depth.checked_add(1).ok_or("has unbalanced parentheses")?,
')' => depth.checked_sub(1).ok_or("has unbalanced parentheses")?,
'$' => return Err("contains '$' outside a quoted string"),
'\'' | '"' => {
quoted(&mut chars, c)?;
depth
}
_ => depth,
};
}
Ok(depth)
}
fn quoted(chars: &mut std::str::Chars<'_>, quote: char) -> Result<(), &'static str> {
let mut previous = None;
loop {
match chars.next() {
None => return Err("has a quote that is not closed"),
Some(c) if c == quote => {
if quote == '\'' && previous == Some('\\') {
return Err("has a backslash before a quote in a string literal");
}
if !chars.as_str().starts_with(quote) {
return Ok(());
}
chars.next();
previous = None;
}
c => previous = c,
}
}
}
pub fn validate_consumer_spec(path: &str, spec: &ConsumerSpec) -> Result<(), String> {
if let Some(name) = spec.durable_name.as_deref() {
if !is_valid_consumer_name(name) {
return Err(format!(
"{path}.durable_name: '{name}' is not a consumer name \
(ASCII letters, digits, '-' and '_', at most 200 characters)"
));
}
}
within(
path,
"ack_wait_secs",
spec.ack_wait_secs,
&ConsumerSpec::ACK_WAIT_SECS,
)?;
match spec.max_deliver {
Some(n) if n != ConsumerSpec::UNLIMITED_DELIVERIES => {
within(path, "max_deliver", Some(n), &ConsumerSpec::MAX_DELIVER)
.map_err(|e| format!("{e} (or -1 for no limit)"))?
}
_ => {}
}
if spec.backoff_secs.len() > ConsumerSpec::MAX_BACKOFF_STEPS {
return Err(format!(
"{path}.backoff_secs: {} steps, more than the {} allowed",
spec.backoff_secs.len(),
ConsumerSpec::MAX_BACKOFF_STEPS
));
}
spec.backoff_secs
.iter()
.enumerate()
.try_for_each(|(i, s)| {
within(
path,
&format!("backoff_secs[{i}]"),
Some(*s),
&ConsumerSpec::BACKOFF_SECS,
)
})?;
within(
path,
"max_ack_pending",
spec.max_ack_pending,
&ConsumerSpec::MAX_ACK_PENDING,
)?;
within(path, "batch", spec.batch, &ConsumerSpec::BATCH)?;
within(
path,
"concurrency",
spec.concurrency,
&ConsumerSpec::CONCURRENCY,
)?;
patterns(&format!("{path}.filter_subjects"), &spec.filter_subjects)?;
distinct(
&format!("{path}.filter_subjects"),
"subject",
spec.filter_subjects.iter().map(String::as_str),
)
}
fn patterns(field: &str, declared: &[String]) -> Result<(), String> {
declared
.iter()
.enumerate()
.find_map(|(i, pattern)| {
pattern_violation(pattern).map(|reason| format!("{field}[{i}]: {reason}"))
})
.map_or(Ok(()), Err)
}
pub fn validate_subscription_delivery(
path: &str,
subscription: &SubscriptionDefinition,
) -> Result<(), String> {
pattern_violation(&subscription.subject)
.map_or(Ok(()), |reason| Err(format!("{path}.subject: {reason}")))?;
within(
path,
"timeout_secs",
subscription.timeout_secs,
&SubscriptionDefinition::TIMEOUT_SECS,
)?;
subscription.consumer.as_ref().map_or(Ok(()), |spec| {
validate_consumer_spec(&format!("{path}.consumer"), spec)
})
}
pub fn validate_stream_definition(path: &str, stream: &StreamDefinition) -> Result<(), String> {
if !is_valid_stream_name(&stream.name) {
return Err(format!(
"{path}.name: '{}' is not a stream name (A-Z, 0-9 and '_', at most 64 characters)",
stream.name
));
}
if stream.subjects.is_empty() {
return Err(format!("{path}.subjects: is empty"));
}
patterns(&format!("{path}.subjects"), &stream.subjects)?;
distinct(
&format!("{path}.subjects"),
"subject",
stream.subjects.iter().map(String::as_str),
)?;
within(path, "max_age_secs", stream.max_age_secs, &(1..=u64::MAX))?;
positive(path, "max_msgs", stream.max_msgs)?;
positive(path, "max_bytes", stream.max_bytes)?;
within(
path,
"duplicate_window_secs",
stream.duplicate_window_secs,
&(1..=u64::MAX),
)?;
match (stream.duplicate_window_secs, stream.max_age_secs) {
(Some(window), Some(age)) if window > age => Err(format!(
"{path}.duplicate_window_secs: {window} is longer than max_age_secs {age}"
)),
_ => Ok(()),
}?;
match stream.description.as_deref() {
Some(text) if text.len() > DESCRIPTION_MAX => Err(format!(
"{path}.description: is longer than {DESCRIPTION_MAX} bytes"
)),
_ => Ok(()),
}
}
pub fn validate_job_config(path: &str, job: &JobConfig) -> Result<(), String> {
distinct(
&format!("{path}.retry_on"),
"failure",
job.retry_on.iter().map(|retry| retry.as_str()),
)?;
let Some(backoff) = job.backoff else {
return Ok(());
};
let path = format!("{path}.backoff");
within(
&path,
"base_secs",
Some(backoff.base_secs),
&BackoffSpec::SECS,
)?;
within(
&path,
"max_secs",
Some(backoff.max_secs),
&BackoffSpec::SECS,
)?;
if backoff.max_secs < backoff.base_secs {
return Err(format!(
"{path}.max_secs: {} is below base_secs {}",
backoff.max_secs, backoff.base_secs
));
}
Ok(())
}
pub fn validate_schedule(path: &str, schedule: &ScheduleDefinition) -> Result<(), String> {
if !ScheduleDefinition::is_valid_name(&schedule.name) {
return Err(format!(
"{path}.name: '{}' is not a schedule name (a-z, 0-9 and '_', at most {} characters)",
schedule.name,
ScheduleDefinition::NAME_MAX
));
}
identifier(&format!("{path}.handler"), &schedule.handler)?;
match &schedule.spec {
ScheduleSpec::Every { secs } if *secs < ScheduleDefinition::MIN_INTERVAL_SECS => {
return Err(format!(
"{path}.spec.secs: {secs} is below the {} second minimum",
ScheduleDefinition::MIN_INTERVAL_SECS
));
}
ScheduleSpec::Every { .. } if schedule.timezone.is_some() => {
return Err(format!(
"{path}.timezone: applies to a cron schedule, not to every(..)"
));
}
ScheduleSpec::Cron { expr } if !matches!(expr.split_whitespace().count(), 5 | 6) => {
return Err(format!(
"{path}.spec.expr: '{expr}' does not have five fields, or six with seconds"
));
}
_ => {}
}
match schedule.timezone.as_deref() {
Some(zone) if !is_timezone_name(zone) => {
return Err(format!(
"{path}.timezone: '{zone}' is not an IANA time zone name"
));
}
_ => {}
}
within(
path,
"timeout_secs",
schedule.timeout_secs,
&ScheduleDefinition::TIMEOUT_SECS,
)?;
match &schedule.job {
Some(_) if schedule.timeout_secs.is_some() => Err(format!(
"{path}.timeout_secs: applies to an inline handler; a job's is job.timeout_secs"
)),
Some(job) => validate_job_config(&format!("{path}.job"), job),
None => Ok(()),
}
}
fn is_timezone_name(zone: &str) -> bool {
(1..=64).contains(&zone.len())
&& zone.split('/').all(|part| {
!part.is_empty()
&& part
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-' | b'+'))
})
}
pub fn validate_schedules(path: &str, schedules: &[ScheduleDefinition]) -> Result<(), String> {
schedules
.iter()
.enumerate()
.try_for_each(|(i, schedule)| validate_schedule(&format!("{path}[{i}]"), schedule))?;
distinct(path, "schedule", schedules.iter().map(|s| s.name.as_str()))
}
pub fn validate_streams(path: &str, streams: &[StreamDefinition]) -> Result<(), String> {
streams
.iter()
.enumerate()
.try_for_each(|(i, stream)| validate_stream_definition(&format!("{path}[{i}]"), stream))?;
distinct(path, "stream", streams.iter().map(|s| s.name.as_str()))
}
pub fn is_migration_name(name: &str) -> bool {
!name.is_empty()
&& name.len() <= 255
&& name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')
}
pub const MAX_MIGRATION_VERSION: u32 = i32::MAX as u32;
pub fn validate_sql_migrations(
path: &str,
migrations: &[SqlMigration],
history: Option<&MigrationHistory>,
) -> Result<(), String> {
migrations.iter().enumerate().try_for_each(|(i, m)| {
let path = format!("{path}[{i}]");
if m.version > MAX_MIGRATION_VERSION {
return Err(format!(
"{path}.version: {} is above {MAX_MIGRATION_VERSION}",
m.version
));
}
if !is_migration_name(&m.name) {
return Err(format!(
"{path}.name: '{}' is not a migration name (ASCII letters, digits and '_')",
m.name
));
}
if m.sql.trim().is_empty() {
return Err(format!("{path}.sql: is empty"));
}
if m.no_transaction != SqlMigration::declares_no_transaction(&m.sql) {
return Err(format!(
"{path}.no_transaction: disagrees with the first line of the SQL \
('{}' opts out of the transaction)",
SqlMigration::NO_TRANSACTION_MARKER
));
}
Ok(())
})?;
let ascending = migrations
.windows(2)
.find(|pair| pair[0].version >= pair[1].version);
if let Some(pair) = ascending {
return Err(format!(
"{path}: version {} follows version {}; versions are distinct and ascending",
pair[1].version, pair[0].version
));
}
history.map_or(Ok(()), |history| {
validate_migration_history("migration_history", history)
})
}
pub fn validate_migration_history(path: &str, history: &MigrationHistory) -> Result<(), String> {
let parts: Vec<&str> = history.table.split('.').collect();
if parts.len() > 2 || !parts.iter().all(|part| is_lowercase_identifier(part)) {
return Err(format!(
"{path}.table: '{}' is not 'table' or 'schema.table' in lowercase",
history.table
));
}
within(
path,
"baseline",
history.baseline,
&(0..=MAX_MIGRATION_VERSION),
)
}
pub fn validate_column_type(path: &str, column_type: &ColumnType) -> Result<(), String> {
match column_type {
ColumnType::Numeric {
precision: None,
scale: Some(_),
} => Err(format!("{path}: a numeric scale needs a precision")),
ColumnType::Numeric {
precision: Some(precision),
scale,
} => {
within(
path,
"precision",
Some(*precision),
&(1..=ColumnType::MAX_NUMERIC_PRECISION),
)?;
within(path, "scale", *scale, &(0..=*precision))
}
ColumnType::Varchar { length } => within(
path,
"length",
Some(*length),
&(1..=ColumnType::MAX_VARCHAR_LENGTH),
),
_ => Ok(()),
}
}
pub fn validate_check_constraint(path: &str, check: &CheckConstraint) -> Result<(), String> {
identifier(&format!("{path}.name"), &check.name)?;
validate_sql_fragment(&format!("{path}.expr"), &check.expr)
}
pub fn validate_check_constraints(path: &str, checks: &[CheckConstraint]) -> Result<(), String> {
checks
.iter()
.enumerate()
.try_for_each(|(i, check)| validate_check_constraint(&format!("{path}[{i}]"), check))?;
distinct(path, "check", checks.iter().map(|c| c.name.as_str()))
}
pub fn validate_index(path: &str, index: &IndexDefinition) -> Result<(), String> {
identifier(&format!("{path}.name"), &index.name)?;
if index.columns.is_empty() {
return Err(format!("{path}.columns: is empty"));
}
index
.columns
.iter()
.enumerate()
.try_for_each(|(i, column)| identifier(&format!("{path}.columns[{i}]"), column))?;
distinct(
&format!("{path}.columns"),
"column",
index.columns.iter().map(String::as_str),
)?;
if index.unique && index.method != IndexMethod::Btree {
return Err(format!(
"{path}.method: a unique index is a btree index, not {}",
index.method.as_sql()
));
}
if index.columns.len() > 1 && matches!(index.method, IndexMethod::Hash) {
return Err(format!("{path}.method: a hash index has one column"));
}
index.predicate.as_deref().map_or(Ok(()), |predicate| {
validate_sql_fragment(&format!("{path}.predicate"), predicate)
})
}
pub fn validate_search_spec(path: &str, search: &SearchSpec) -> Result<(), String> {
if search.columns.is_empty() && search.vector_column.is_none() {
return Err(format!("{path}: names neither columns nor a vector_column"));
}
search
.columns
.iter()
.enumerate()
.try_for_each(|(i, column)| identifier(&format!("{path}.columns[{i}]"), column))?;
distinct(
&format!("{path}.columns"),
"column",
search.columns.iter().map(String::as_str),
)?;
if !is_lowercase_identifier(&search.config) {
return Err(format!(
"{path}.config: '{}' is not a text search configuration name",
search.config
));
}
search.vector_column.as_deref().map_or(Ok(()), |column| {
identifier(&format!("{path}.vector_column"), column)
})
}
pub fn validate_audit_redact(path: &str, rules: &[String]) -> Result<(), String> {
rules.iter().enumerate().try_for_each(|(i, rule)| {
rule.split('.')
.try_for_each(|key| {
let name = key.strip_suffix("[]").unwrap_or(key);
identifier(
&format!("{path}[{i}]"),
name.strip_suffix('*').unwrap_or(name),
)
})
.map_err(|e| {
format!(
"{e}; a rule is a column name or 'prefix*', and the keys inside it joined by \
'.', a key whose value is an array followed by '[]'"
)
})
})?;
distinct(path, "rule", rules.iter().map(String::as_str))
}
pub fn validate_extensions(path: &str, names: &[String]) -> Result<(), String> {
refused_extension(path, names).map_or(Ok(()), |(_, reason)| Err(reason))
}
pub fn refused_extension(path: &str, names: &[String]) -> Option<(usize, String)> {
let mut seen = BTreeSet::new();
names.iter().enumerate().find_map(|(i, name)| {
name_violation(name)
.map(|violation| format!("{path}[{i}]: {violation}"))
.or_else(|| {
(!seen.insert(name.as_str())).then(|| listed_twice(path, "extension", name))
})
.map(|reason| (i, reason))
})
}
pub fn validate_provides(path: &str, provides: &[ProvidedExtension]) -> Result<(), String> {
if provides.is_empty() {
return Err(format!(
"{path}: a native service provides at least one extension"
));
}
let names: Vec<String> = provides.iter().map(|p| p.name.clone()).collect();
validate_extensions(path, &names)?;
provides.iter().enumerate().try_for_each(|(i, provided)| {
callers(provided.callers.clone())
.map(drop)
.map_err(|e| format!("{path}[{i}].{e}"))
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::nats_names::SUBJECT_MAX;
use crate::{
CatchUp, DiscardSpec, HandlerType, Overlap, RetryOn, StreamRetention, StreamStorage,
};
fn refused(result: Result<(), String>, path: &str, reason: &str) {
let error = result.expect_err(reason);
assert!(error.starts_with(path), "'{error}' does not name {path}");
assert!(error.contains(reason), "'{error}' lacks '{reason}'");
}
#[test]
fn schema_names() {
for valid in ["note_service", "_private", "a", "leads2", &"s".repeat(63)] {
assert_eq!(
validate_schema_name("schema_name", valid),
Ok(()),
"{valid}"
);
}
for (invalid, reason) in [
("", "is not a schema name"),
("Note_Service", "is not a schema name"),
("note-service", "is not a schema name"),
("1notes", "is not a schema name"),
("notes\"; DROP", "is not a schema name"),
(&*"s".repeat(64), "is not a schema name"),
("public", "is reserved"),
("information_schema", "is reserved"),
("_cufflink", "is reserved"),
("_cufflink_outbox", "is reserved"),
("pg_catalog", "is reserved"),
("pg_anything", "is reserved"),
] {
refused(
validate_schema_name("schema_name", invalid),
"schema_name",
reason,
);
}
}
#[test]
fn extensions() {
let names = |names: &[&str]| {
names
.iter()
.map(|name| name.to_string())
.collect::<Vec<_>>()
};
assert_eq!(validate_extensions("extensions", &[]), Ok(()));
assert_eq!(
validate_extensions("extensions", &names(&["google.token", "pdf.render-v2"])),
Ok(())
);
assert_eq!(
validate_extensions("extensions", &names(&["google.token", "Pdf"])),
Err(
"extensions[1]: 'Pdf' is not an extension name (segments of a-z, 0-9 and '-' \
joined by '.', at most 128 bytes)"
.into()
)
);
assert_eq!(
validate_extensions("extensions", &names(&["a.b", "c", "a.b"])),
Err("extensions: extension 'a.b' is listed twice".into())
);
assert_eq!(
refused_extension("extensions", &names(&["a.b", "a.b", "Bad"])),
Some((1, "extensions: extension 'a.b' is listed twice".into())),
"the first refused is named, where it is"
);
assert_eq!(
refused_extension("extensions", &names(&["a.b", "Bad", "a.b"])).map(|(i, _)| i),
Some(1)
);
}
#[test]
fn provided_extensions() {
let provided = |name: &str, callers: &[&str]| ProvidedExtension {
name: name.into(),
callers: callers.iter().map(|c| c.to_string()).collect(),
};
assert_eq!(
validate_provides(
"provides",
&[
provided("google.token", &["reports-service", "reports-service"]),
provided("pdf.render", &["*"]),
]
),
Ok(())
);
for (provides, reason) in [
(
vec![],
"provides: a native service provides at least one extension",
),
(
vec![provided("google.token", &["*"]), provided("Pdf", &["*"])],
"provides[1]: 'Pdf' is not an extension name (segments of a-z, 0-9 and '-' \
joined by '.', at most 128 bytes)",
),
(
vec![provided("a.b", &["*"]), provided("a.b", &["x"])],
"provides: extension 'a.b' is listed twice",
),
(
vec![provided("a.b", &["*"]), provided("c", &[])],
"provides[1].callers: at least one caller is required",
),
(
vec![provided("a.b", &["reports", "a b"])],
"provides[0].callers: 'a b' is neither a service name nor '*'",
),
] {
assert_eq!(validate_provides("provides", &provides), Err(reason.into()));
}
}
#[test]
fn a_schema_is_reserved_however_it_is_capitalised() {
for reserved in [
"public",
"Public",
"PG_CATALOG",
"pg_toast",
"Information_Schema",
"_cufflink",
"_Cufflink_audit",
] {
assert!(is_reserved_schema(reserved), "{reserved}");
}
for free in ["order", "user", "publics", "cufflink", "pg", "note_service"] {
assert!(!is_reserved_schema(free), "{free}");
}
}
#[test]
fn a_folded_identifier_names_what_the_unquoted_name_does() {
for (name, quoted) in [
("note_service", "\"note_service\""),
("Order_Desk_2", "\"order_desk_2\""),
("order", "\"order\""),
("USER", "\"user\""),
("2fa", "\"2fa\""),
("a\"b", "\"a\"\"b\""),
] {
assert_eq!(folded_ident(name), quoted);
}
assert_eq!(folded("Order_Desk_2"), "order_desk_2");
}
#[test]
fn sql_fragments() {
for valid in [
"price_cents >= 0",
"deleted_at IS NULL",
"(a > 0) AND (b IN ('x', 'y'))",
"lower(email) = email",
"note <> ')'",
"label IN ('(', 'it''s (')",
"\"odd)name\" > 0 AND \"say \"\"(\"\"\" IS NULL",
"email ~ '^\\S+@\\S+$'",
"code = E'\\\\n'",
] {
assert_eq!(validate_sql_fragment("expr", valid), Ok(()), "{valid}");
}
for (invalid, reason) in [
("", "is empty"),
(" ", "is empty"),
("a > 0; DROP TABLE t", "contains ';'"),
("a > 0 -- always", "contains '--'"),
("a > 0 /* always */", "contains '/*'"),
("a > 0 */", "contains '*/'"),
("(a > 0", "unbalanced"),
("a > 0)", "unbalanced"),
(")a > 0(", "unbalanced"),
(
"'(' = '(' ) , DROP COLUMN b CASCADE , ADD CONSTRAINT z CHECK ( ')' = ')'",
"unbalanced",
),
(
"\"(\" = 1 ) , DROP COLUMN b , ADD CHECK ( \")\" = 1",
"unbalanced",
),
(
"a = E'\\' ) , DROP COLUMN b , ADD CHECK ( '' = ''",
"a backslash before a quote",
),
("a = E'\\'')'", "a backslash before a quote"),
("a = $$x$$", "contains '$'"),
("a = $1", "contains '$'"),
("a = 'open", "not closed"),
("\"open = 1", "not closed"),
] {
refused(validate_sql_fragment("expr", invalid), "expr", reason);
}
}
#[test]
fn a_valid_consumer_spec_passes() {
let spec = ConsumerSpec {
durable_name: Some("notes-service-lead-consumer".into()),
ack_wait_secs: Some(30),
max_deliver: Some(3),
backoff_secs: vec![10, 60, 300],
max_ack_pending: Some(1000),
batch: Some(32),
deliver_policy: None,
filter_subjects: vec!["leads.deleted.notes".into(), "leads.*.set".into()],
concurrency: Some(4),
dead_letter: Some(true),
};
assert_eq!(validate_consumer_spec("c", &spec), Ok(()));
assert_eq!(
validate_consumer_spec("c", &ConsumerSpec::default()),
Ok(())
);
let unlimited = ConsumerSpec {
max_deliver: Some(-1),
..Default::default()
};
assert_eq!(validate_consumer_spec("c", &unlimited), Ok(()));
}
#[test]
fn each_consumer_field_is_bounded() {
let cases: Vec<(ConsumerSpec, &str)> = vec![
(
ConsumerSpec {
durable_name: Some("a.b".into()),
..Default::default()
},
"c.durable_name",
),
(
ConsumerSpec {
durable_name: Some(String::new()),
..Default::default()
},
"c.durable_name",
),
(
ConsumerSpec {
ack_wait_secs: Some(4),
..Default::default()
},
"c.ack_wait_secs",
),
(
ConsumerSpec {
ack_wait_secs: Some(3601),
..Default::default()
},
"c.ack_wait_secs",
),
(
ConsumerSpec {
max_deliver: Some(0),
..Default::default()
},
"c.max_deliver",
),
(
ConsumerSpec {
max_deliver: Some(-2),
..Default::default()
},
"c.max_deliver",
),
(
ConsumerSpec {
max_deliver: Some(101),
..Default::default()
},
"c.max_deliver",
),
(
ConsumerSpec {
backoff_secs: vec![1; 11],
..Default::default()
},
"c.backoff_secs",
),
(
ConsumerSpec {
backoff_secs: vec![5, 0],
..Default::default()
},
"c.backoff_secs[1]",
),
(
ConsumerSpec {
backoff_secs: vec![86_401],
..Default::default()
},
"c.backoff_secs[0]",
),
(
ConsumerSpec {
max_ack_pending: Some(0),
..Default::default()
},
"c.max_ack_pending",
),
(
ConsumerSpec {
max_ack_pending: Some(100_001),
..Default::default()
},
"c.max_ack_pending",
),
(
ConsumerSpec {
batch: Some(0),
..Default::default()
},
"c.batch",
),
(
ConsumerSpec {
batch: Some(257),
..Default::default()
},
"c.batch",
),
(
ConsumerSpec {
concurrency: Some(0),
..Default::default()
},
"c.concurrency",
),
(
ConsumerSpec {
concurrency: Some(65),
..Default::default()
},
"c.concurrency",
),
(
ConsumerSpec {
filter_subjects: vec!["leads.ok".into(), "leads..bad".into()],
..Default::default()
},
"c.filter_subjects[1]",
),
(
ConsumerSpec {
filter_subjects: vec![format!("leads.{}", "x".repeat(SUBJECT_MAX))],
..Default::default()
},
"c.filter_subjects[0]",
),
(
ConsumerSpec {
filter_subjects: vec!["leads.a".into(), "leads.a".into()],
..Default::default()
},
"c.filter_subjects",
),
];
for (spec, path) in cases {
refused(validate_consumer_spec("c", &spec), path, "");
}
}
#[test]
fn subscription_delivery_checks_the_subject_the_timeout_and_the_consumer() {
let plain = || SubscriptionDefinition::new("x", "h", HandlerType::Wasm);
assert_eq!(validate_subscription_delivery("s", &plain()), Ok(()));
for subject in ["dw.{tenant}.leads.*", "cufflink.tick.*", "dw.{tenant}.>"] {
let sub = SubscriptionDefinition::new(subject, "h", HandlerType::Wasm);
assert_eq!(
validate_subscription_delivery("s", &sub),
Ok(()),
"{subject}"
);
}
for (subject, reason) in [
(String::new(), "'' is not a subject pattern".to_string()),
(
"leads..created".into(),
"'leads..created' is not a subject pattern".into(),
),
(
"leads.>.created".into(),
"'leads.>.created' is not a subject pattern".into(),
),
(
format!("dw.{{tenant}}.{}", "x".repeat(SUBJECT_MAX)),
"a subject pattern of 1036 bytes is longer".into(),
),
] {
refused(
validate_subscription_delivery(
"s",
&SubscriptionDefinition::new(subject, "h", HandlerType::Wasm),
),
"s.subject",
&reason,
);
}
for timeout in [1, 60, 300] {
let sub = SubscriptionDefinition {
timeout_secs: Some(timeout),
..plain()
};
assert_eq!(validate_subscription_delivery("s", &sub), Ok(()));
}
for timeout in [0, 301] {
let sub = SubscriptionDefinition {
timeout_secs: Some(timeout),
..plain()
};
refused(
validate_subscription_delivery("s", &sub),
"s.timeout_secs",
"outside 1..=300",
);
}
let sub = SubscriptionDefinition {
consumer: Some(ConsumerSpec {
batch: Some(0),
..Default::default()
}),
..plain()
};
refused(
validate_subscription_delivery("s", &sub),
"s.consumer.batch",
"outside",
);
}
fn stream() -> StreamDefinition {
StreamDefinition {
name: "NOTES".into(),
subjects: vec!["notes.created".into(), "notes.callback.>".into()],
retention: StreamRetention::Limits,
max_age_secs: Some(2_592_000),
max_msgs: Some(100_000),
max_bytes: Some(268_435_456),
storage: StreamStorage::File,
duplicate_window_secs: Some(120),
discard: DiscardSpec::Old,
description: Some("notes".into()),
}
}
#[test]
fn streams() {
assert_eq!(validate_stream_definition("st", &stream()), Ok(()));
let cases: Vec<(StreamDefinition, &str, &str)> = vec![
(
StreamDefinition {
name: "notes".into(),
..stream()
},
"st.name",
"is not a stream name",
),
(
StreamDefinition {
subjects: vec![],
..stream()
},
"st.subjects",
"is empty",
),
(
StreamDefinition {
subjects: vec!["notes.>.x".into()],
..stream()
},
"st.subjects[0]",
"is not a subject pattern",
),
(
StreamDefinition {
subjects: vec![
"notes.created".into(),
format!("notes.{}", "x".repeat(SUBJECT_MAX)),
],
..stream()
},
"st.subjects[1]",
"a subject pattern of 1030 bytes is longer than the 1024 bytes",
),
(
StreamDefinition {
subjects: vec!["a.b".into(), "a.b".into()],
..stream()
},
"st.subjects",
"listed twice",
),
(
StreamDefinition {
max_age_secs: Some(0),
..stream()
},
"st.max_age_secs",
"outside",
),
(
StreamDefinition {
max_msgs: Some(0),
..stream()
},
"st.max_msgs",
"not positive",
),
(
StreamDefinition {
max_bytes: Some(-1),
..stream()
},
"st.max_bytes",
"not positive",
),
(
StreamDefinition {
duplicate_window_secs: Some(0),
..stream()
},
"st.duplicate_window_secs",
"outside",
),
(
StreamDefinition {
max_age_secs: Some(60),
duplicate_window_secs: Some(120),
..stream()
},
"st.duplicate_window_secs",
"longer than max_age_secs",
),
(
StreamDefinition {
description: Some("d".repeat(4097)),
..stream()
},
"st.description",
"longer than",
),
];
for (invalid, path, reason) in cases {
refused(validate_stream_definition("st", &invalid), path, reason);
}
refused(
validate_streams("streams", &[stream(), stream()]),
"streams",
"stream 'NOTES' is listed twice",
);
refused(
validate_streams(
"streams",
&[
stream(),
StreamDefinition {
name: "bad".into(),
..stream()
},
],
),
"streams[1].name",
"is not a stream name",
);
}
fn schedule(spec: ScheduleSpec) -> ScheduleDefinition {
ScheduleDefinition {
name: "order_retry".into(),
handler: "retry_orders".into(),
spec,
timezone: None,
overlap: Overlap::Skip,
catch_up: CatchUp::Latest,
job: None,
timeout_secs: None,
}
}
fn cron(expr: &str) -> ScheduleSpec {
ScheduleSpec::Cron { expr: expr.into() }
}
#[test]
fn valid_schedules_pass() {
for valid in [
schedule(ScheduleSpec::Every { secs: 10 }),
schedule(cron("0 * * * *")),
schedule(cron("*/30 0 * * * *")),
ScheduleDefinition {
timezone: Some("Europe/London".into()),
timeout_secs: Some(300),
..schedule(cron("0 9 * * 1-5"))
},
ScheduleDefinition {
timezone: Some("Etc/GMT+5".into()),
..schedule(cron("0 9 * * *"))
},
ScheduleDefinition {
job: Some(JobConfig {
retry_on: vec![RetryOn::Error],
backoff: Some(BackoffSpec::default()),
..Default::default()
}),
..schedule(ScheduleSpec::Every { secs: 300 })
},
] {
assert_eq!(validate_schedule("sc", &valid), Ok(()), "{valid:?}");
}
}
#[test]
fn invalid_schedules_name_their_field() {
let cases: Vec<(ScheduleDefinition, &str, &str)> = vec![
(
ScheduleDefinition {
name: "Order-Retry".into(),
..schedule(ScheduleSpec::Every { secs: 30 })
},
"sc.name",
"is not a schedule name",
),
(
ScheduleDefinition {
handler: "retry orders".into(),
..schedule(ScheduleSpec::Every { secs: 30 })
},
"sc.handler",
"is not an identifier",
),
(
schedule(ScheduleSpec::Every { secs: 9 }),
"sc.spec.secs",
"below the 10 second minimum",
),
(
ScheduleDefinition {
timezone: Some("Europe/London".into()),
..schedule(ScheduleSpec::Every { secs: 30 })
},
"sc.timezone",
"applies to a cron schedule",
),
(schedule(cron("")), "sc.spec.expr", "five fields"),
(schedule(cron("* * * *")), "sc.spec.expr", "five fields"),
(
schedule(cron("* * * * * * *")),
"sc.spec.expr",
"five fields",
),
(
ScheduleDefinition {
timezone: Some("Europe//London".into()),
..schedule(cron("0 * * * *"))
},
"sc.timezone",
"is not an IANA time zone name",
),
(
ScheduleDefinition {
timezone: Some("Europe/Lon don".into()),
..schedule(cron("0 * * * *"))
},
"sc.timezone",
"is not an IANA time zone name",
),
(
ScheduleDefinition {
timeout_secs: Some(0),
..schedule(cron("0 * * * *"))
},
"sc.timeout_secs",
"outside 1..=300",
),
(
ScheduleDefinition {
timeout_secs: Some(301),
..schedule(cron("0 * * * *"))
},
"sc.timeout_secs",
"outside 1..=300",
),
(
ScheduleDefinition {
timeout_secs: Some(30),
job: Some(JobConfig::default()),
..schedule(cron("0 * * * *"))
},
"sc.timeout_secs",
"applies to an inline handler",
),
(
ScheduleDefinition {
job: Some(JobConfig {
retry_on: vec![RetryOn::Error, RetryOn::Error],
..Default::default()
}),
..schedule(cron("0 * * * *"))
},
"sc.job.retry_on",
"listed twice",
),
];
for (invalid, path, reason) in cases {
refused(validate_schedule("sc", &invalid), path, reason);
}
let twice = [
schedule(ScheduleSpec::Every { secs: 30 }),
schedule(cron("0 * * * *")),
];
refused(
validate_schedules("schedules", &twice),
"schedules",
"schedule 'order_retry' is listed twice",
);
}
#[test]
fn job_backoff_is_bounded_and_ordered() {
let job = |base_secs, max_secs| JobConfig {
backoff: Some(BackoffSpec {
base_secs,
max_secs,
}),
..Default::default()
};
assert_eq!(validate_job_config("job", &JobConfig::default()), Ok(()));
assert_eq!(validate_job_config("job", &job(30, 3600)), Ok(()));
assert_eq!(validate_job_config("job", &job(60, 60)), Ok(()));
refused(
validate_job_config("job", &job(0, 60)),
"job.backoff.base_secs",
"outside",
);
refused(
validate_job_config("job", &job(1, 86_401)),
"job.backoff.max_secs",
"outside",
);
refused(
validate_job_config("job", &job(120, 60)),
"job.backoff.max_secs",
"below base_secs",
);
}
fn migration(version: u32, name: &str, sql: &str) -> SqlMigration {
SqlMigration {
version,
name: name.into(),
sql: sql.into(),
no_transaction: SqlMigration::declares_no_transaction(sql),
}
}
#[test]
fn sql_migrations() {
let valid = [
migration(1, "initial_tables", "CREATE TABLE notes ();"),
migration(6, "add_note_kind", "ALTER TABLE notes ADD kind TEXT;"),
migration(
7,
"callback_index",
"-- cufflink:no-transaction\nCREATE INDEX CONCURRENTLY i ON notes (kind);",
),
];
assert_eq!(validate_sql_migrations("migrations", &valid, None), Ok(()));
assert_eq!(validate_sql_migrations("migrations", &[], None), Ok(()));
assert_eq!(
validate_sql_migrations("migrations", &[migration(0, "zero", "SELECT 1;")], None),
Ok(())
);
let cases: Vec<(Vec<SqlMigration>, &str, &str)> = vec![
(
vec![
migration(1, "a", "SELECT 1;"),
migration(1, "b", "SELECT 1;"),
],
"migrations",
"version 1 follows version 1",
),
(
vec![
migration(6, "a", "SELECT 1;"),
migration(2, "b", "SELECT 1;"),
],
"migrations",
"version 2 follows version 6",
),
(
vec![migration(1, "add-kind", "SELECT 1;")],
"migrations[0].name",
"is not a migration name",
),
(
vec![migration(1, "", "SELECT 1;")],
"migrations[0].name",
"is not a migration name",
),
(
vec![migration(1, "a", " \n")],
"migrations[0].sql",
"is empty",
),
(
vec![migration(2_147_483_648, "a", "SELECT 1;")],
"migrations[0].version",
"is above",
),
(
vec![SqlMigration {
no_transaction: true,
..migration(1, "a", "SELECT 1;")
}],
"migrations[0].no_transaction",
"disagrees",
),
(
vec![SqlMigration {
no_transaction: false,
..migration(1, "a", "-- cufflink:no-transaction\nSELECT 1;")
}],
"migrations[0].no_transaction",
"disagrees",
),
];
for (invalid, path, reason) in cases {
refused(
validate_sql_migrations("migrations", &invalid, None),
path,
reason,
);
}
}
#[test]
fn migration_history() {
let history = |table: &str, baseline| MigrationHistory {
table: table.into(),
baseline,
..Default::default()
};
for valid in [
history("schema_history", None),
history("public.schema_history_deferred", Some(15)),
history("refinery_schema_history", Some(0)),
] {
assert_eq!(validate_migration_history("h", &valid), Ok(()));
assert_eq!(
validate_sql_migrations("migrations", &[], Some(&valid)),
Ok(())
);
}
for invalid in [
"",
"a.b.c",
"schema.",
".table",
"sche ma.t",
"t\"; DROP",
"Schema_History",
"Public.schema_history",
] {
refused(
validate_migration_history("h", &history(invalid, None)),
"h.table",
"is not 'table' or 'schema.table'",
);
}
refused(
validate_migration_history("h", &history("t", Some(2_147_483_648))),
"h.baseline",
"outside",
);
refused(
validate_sql_migrations("migrations", &[], Some(&history("a.b.c", None))),
"migration_history.table",
"is not",
);
}
#[test]
fn column_types() {
let numeric = |precision, scale| ColumnType::Numeric { precision, scale };
for valid in [
ColumnType::Text,
ColumnType::Bytea,
ColumnType::JsonbArray,
numeric(None, None),
numeric(Some(1), None),
numeric(Some(12), Some(2)),
numeric(Some(12), Some(12)),
numeric(Some(1000), Some(0)),
ColumnType::Varchar { length: 1 },
ColumnType::Varchar { length: 10_485_760 },
] {
assert_eq!(validate_column_type("c", &valid), Ok(()), "{valid:?}");
}
for (invalid, path, reason) in [
(numeric(None, Some(2)), "c", "needs a precision"),
(numeric(Some(0), None), "c.precision", "outside"),
(numeric(Some(1001), None), "c.precision", "outside"),
(numeric(Some(4), Some(5)), "c.scale", "outside 0..=4"),
(ColumnType::Varchar { length: 0 }, "c.length", "outside"),
(
ColumnType::Varchar { length: 10_485_761 },
"c.length",
"outside",
),
] {
refused(validate_column_type("c", &invalid), path, reason);
}
}
#[test]
fn check_constraints() {
let check = |name: &str, expr: &str| CheckConstraint {
name: name.into(),
expr: expr.into(),
};
assert_eq!(
validate_check_constraint("ck", &check("price_nonneg", "price_cents >= 0")),
Ok(())
);
refused(
validate_check_constraint("ck", &check("price-nonneg", "price_cents >= 0")),
"ck.name",
"is not an identifier",
);
refused(
validate_check_constraint("ck", &check(&"n".repeat(64), "a > 0")),
"ck.name",
"is not an identifier",
);
refused(
validate_check_constraint("ck", &check("c", "a > 0; DROP TABLE t")),
"ck.expr",
"contains ';'",
);
refused(
validate_check_constraints("checks", &[check("c", "a > 0"), check("c", "b > 0")]),
"checks",
"check 'c' is listed twice",
);
refused(
validate_check_constraints("checks", &[check("c", "a > 0"), check("d", "(b > 0")]),
"checks[1].expr",
"unbalanced",
);
}
fn index(columns: &[&str]) -> IndexDefinition {
IndexDefinition {
name: "uq_org_email".into(),
columns: columns.iter().map(|c| c.to_string()).collect(),
..Default::default()
}
}
#[test]
fn indexes() {
for valid in [
index(&["email"]),
IndexDefinition {
unique: true,
predicate: Some("deleted_at IS NULL".into()),
..index(&["organisation_id", "email"])
},
IndexDefinition {
method: IndexMethod::Gin,
..index(&["services"])
},
IndexDefinition {
method: IndexMethod::Brin,
..index(&["created_at", "id"])
},
] {
assert_eq!(validate_index("ix", &valid), Ok(()), "{valid:?}");
}
let cases: Vec<(IndexDefinition, &str, &str)> = vec![
(
IndexDefinition {
name: "uq org".into(),
..index(&["a"])
},
"ix.name",
"is not an identifier",
),
(index(&[]), "ix.columns", "is empty"),
(
index(&["a", "b\""]),
"ix.columns[1]",
"is not an identifier",
),
(index(&["a", "a"]), "ix.columns", "listed twice"),
(
IndexDefinition {
unique: true,
method: IndexMethod::Gin,
..index(&["a"])
},
"ix.method",
"a unique index is a btree index, not gin",
),
(
IndexDefinition {
method: IndexMethod::Hash,
..index(&["a", "b"])
},
"ix.method",
"a hash index has one column",
),
(
IndexDefinition {
predicate: Some("a IS NULL; DROP TABLE t".into()),
..index(&["a"])
},
"ix.predicate",
"contains ';'",
),
];
for (invalid, path, reason) in cases {
refused(validate_index("ix", &invalid), path, reason);
}
}
#[test]
fn search_specs() {
let search = |columns: &[&str], config: &str, vector: Option<&str>| SearchSpec {
columns: columns.iter().map(|c| c.to_string()).collect(),
config: config.into(),
vector_column: vector.map(str::to_string),
};
for valid in [
search(&["name", "email"], "simple", None),
search(&[], "english", Some("search_vector")),
search(&["name"], "simple", Some("search_vector")),
] {
assert_eq!(validate_search_spec("se", &valid), Ok(()), "{valid:?}");
}
for (invalid, path, reason) in [
(search(&[], "simple", None), "se", "names neither"),
(
search(&["na me"], "simple", None),
"se.columns[0]",
"is not an identifier",
),
(
search(&["name", "name"], "simple", None),
"se.columns",
"listed twice",
),
(
search(&["name"], "English'", None),
"se.config",
"is not a text search configuration name",
),
(
search(&["name"], "simple", Some("v;")),
"se.vector_column",
"is not an identifier",
),
] {
refused(validate_search_spec("se", &invalid), path, reason);
}
}
#[test]
fn audit_redaction_rules() {
let rules =
|rules: &[&str]| -> Vec<String> { rules.iter().map(|r| r.to_string()).collect() };
assert_eq!(
validate_audit_redact(
"r",
&rules(&[
"dob",
"bank_*",
"_x*",
"profile.dob",
"contacts[].email",
"meta.card_*",
"tags[]",
"orders[].lines[].sku*"
])
),
Ok(())
);
assert_eq!(validate_audit_redact("r", &[]), Ok(()));
for (invalid, path, reason) in [
(
rules(&["*"]),
"r[0]",
"a rule is a column name or 'prefix*'",
),
(rules(&["dob", "ba*nk"]), "r[1]", "is not an identifier"),
(rules(&["bank_**"]), "r[0]", "is not an identifier"),
(rules(&["dob", "dob"]), "r", "listed twice"),
(rules(&["profile."]), "r[0]", "is not an identifier"),
(rules(&[".dob"]), "r[0]", "is not an identifier"),
(rules(&["a..b"]), "r[0]", "is not an identifier"),
(rules(&["contacts[]email"]), "r[0]", "is not an identifier"),
(
rules(&["contacts[0].email"]),
"r[0]",
"is not an identifier",
),
(rules(&["tags[][]"]), "r[0]", "is not an identifier"),
(rules(&["[]"]), "r[0]", "is not an identifier"),
] {
refused(validate_audit_redact("r", &invalid), path, reason);
}
}
}