use std::collections::HashMap;
use super::builtins::{
self, ParamContext, AGGREGATE_FUNCTIONS, AGGREGATE_REQUIRES_FIELD, AGGREGATE_REQUIRES_TWO_ARGS,
ALERT_PARAMS, LOG_PARAMS, WATERMARK_PARAMS,
};
use super::scope::*;
use super::suggest::{did_you_mean, suggest};
use super::{RelatedSpan, Severity, Validator};
use crate::ast::*;
use crate::span::Span;
pub fn pass1_declarations(v: &mut Validator, program: &Program) {
for stmt in &program.statements {
let span = stmt.span;
match &stmt.node {
Stmt::EventDecl { name, fields, .. } => {
if let Some(prev) = v.symbols.events.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E001",
format!("duplicate event type '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols.events.insert(
name.clone(),
EventInfo {
span,
field_names: fields.iter().map(|f| f.name.clone()).collect(),
},
);
}
}
Stmt::StreamDecl { name, .. } => {
if let Some(prev) = v.symbols.streams.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E002",
format!("duplicate stream '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols.streams.insert(name.clone(), StreamInfo { span });
}
}
Stmt::FnDecl { name, params, .. } => {
if let Some(prev) = v.symbols.functions.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E003",
format!("duplicate function '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols.functions.insert(
name.clone(),
FunctionInfo {
span,
param_count: params.len(),
},
);
}
}
Stmt::ConnectorDecl {
name,
connector_type,
..
} => {
if let Some(prev) = v.symbols.connectors.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E004",
format!("duplicate connector '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols.connectors.insert(
name.clone(),
ConnectorInfo {
span,
connector_type: connector_type.clone(),
},
);
}
}
Stmt::ContextDecl { name, .. } => {
if let Some(prev) = v.symbols.contexts.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E005",
format!("duplicate context '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols
.contexts
.insert(name.clone(), ContextInfo { span });
}
}
Stmt::PatternDecl { name, .. } => {
if let Some(prev) = v.symbols.patterns.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E006",
format!("duplicate pattern '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols
.patterns
.insert(name.clone(), PatternInfo { span });
}
}
Stmt::VarDecl { name, mutable, .. } => {
v.symbols.variables.insert(
name.clone(),
VarInfo {
span,
mutable: *mutable,
},
);
}
Stmt::ConstDecl { name, .. } => {
v.symbols.variables.insert(
name.clone(),
VarInfo {
span,
mutable: false,
},
);
}
Stmt::TypeDecl { name, fields, .. } => {
if let Some(prev) = v.symbols.types.get(name) {
v.emit_with_related(
Severity::Error,
span,
"E007",
format!("duplicate type alias '{name}'"),
vec![RelatedSpan {
span: prev.span,
message: "previously declared here".to_string(),
}],
);
} else {
v.symbols.types.insert(
name.clone(),
TypeInfo {
span,
fields: fields
.iter()
.map(|f| (f.name.clone(), f.ty.clone()))
.collect(),
},
);
}
}
_ => {}
}
}
}
pub fn pass2_semantic(v: &mut Validator, program: &Program) {
for stmt in &program.statements {
let span = stmt.span;
match &stmt.node {
Stmt::StreamDecl {
source,
ops,
op_spans,
..
} => {
check_stream_source(v, source, span);
check_stream_ops(v, ops, op_spans, source, span);
}
Stmt::ConnectorDecl {
connector_type,
params,
..
} => {
if !builtins::is_known_connector_type(connector_type) {
let known: Vec<&str> = builtins::KNOWN_CONNECTOR_TYPES.to_vec();
let suggestion = did_you_mean(connector_type, &known);
v.emit_with_hint(
Severity::Error,
span,
"E008",
format!("unknown connector type '{connector_type}'"),
format!(
"known types: {}{}",
builtins::KNOWN_CONNECTOR_TYPES.join(", "),
suggestion
),
);
} else {
check_connector_params(
v,
params,
connector_type,
ParamContext::Both,
"connector declaration",
span,
);
}
}
Stmt::PatternDecl { expr, .. } => {
check_sase_pattern_refs(v, expr, span);
}
Stmt::Assignment { name, value } => {
check_assignment(v, name, span);
check_expr_functions(v, value, span);
}
Stmt::VarDecl { value, .. } | Stmt::ConstDecl { value, .. } => {
check_expr_functions(v, value, span);
}
_ => {}
}
}
}
fn check_assignment(v: &mut Validator, name: &str, span: Span) {
if let Some(var_info) = v.symbols.variables.get(name) {
if !var_info.mutable {
let decl_snippet = v
.snippet(var_info.span)
.unwrap_or("")
.lines()
.next()
.unwrap_or("");
let context = if decl_snippet.is_empty() {
String::new()
} else {
format!(" (from: {})", decl_snippet.trim())
};
v.emit_with_related(
Severity::Error,
span,
"E040",
format!("cannot assign to immutable variable '{name}'{context}"),
vec![RelatedSpan {
span: var_info.span,
message: "declared as immutable here — use 'var' instead of 'let'".to_string(),
}],
);
}
}
}
fn check_stream_source(v: &mut Validator, source: &StreamSource, span: Span) {
match source {
StreamSource::Ident(name) => {
check_source_name(v, name, span);
}
StreamSource::IdentWithAlias { name, .. }
| StreamSource::IdentWithFilterAndAlias { name, .. }
| StreamSource::AllWithAlias { name, .. } => {
check_source_name(v, name, span);
}
StreamSource::FromConnector {
connector_name,
params,
..
} => {
if !v.symbols.connectors.contains_key(connector_name) {
let suggestion = did_you_mean(connector_name, &v.symbols.connector_names());
v.emit_with_hint(
Severity::Error,
span,
"E030",
format!("undefined connector '{connector_name}'"),
format!("declare it with: connector {connector_name} = type (...){suggestion}"),
);
} else {
let connector_type = v.symbols.connectors[connector_name].connector_type.clone();
check_connector_params(
v,
params,
&connector_type,
ParamContext::Source,
".from()",
span,
);
}
}
StreamSource::Merge(inline_streams) => {
for s in inline_streams {
check_source_name(v, &s.source, span);
}
}
StreamSource::Join(clauses) => {
for c in clauses {
check_source_name(v, &c.source, span);
}
}
StreamSource::Sequence(seq) => {
for step in &seq.steps {
check_source_name(v, &step.event_type, span);
}
}
StreamSource::Timer(_) => {}
}
}
fn check_source_name(v: &mut Validator, name: &str, span: Span) {
if !v.symbols.events.contains_key(name)
&& !v.symbols.streams.contains_key(name)
&& !v.symbols.patterns.contains_key(name)
{
let suggestion = did_you_mean(name, &v.symbols.source_names());
v.emit_with_hint(
Severity::Error,
span,
"E033",
format!("undefined event type or stream '{name}'"),
format!("declare it with: event {name} {{ ... }} or stream {name} = ...{suggestion}"),
);
}
}
fn check_stream_ops(
v: &mut Validator,
ops: &[StreamOp],
op_spans: &[Span],
source: &StreamSource,
span: Span,
) {
let mut seen_aggregate = false;
let mut seen_window = false;
let mut in_sequence = is_sequence_source(source);
let mut alias_to_event: HashMap<String, String> = HashMap::new();
match source {
StreamSource::Ident(name) if v.symbols.events.contains_key(name) => {
alias_to_event.insert(name.clone(), name.clone());
}
StreamSource::Ident(_) => {}
StreamSource::IdentWithAlias { name, alias } => {
alias_to_event.insert(alias.clone(), name.clone());
}
StreamSource::IdentWithFilterAndAlias { name, alias, .. } => {
if let Some(a) = alias {
alias_to_event.insert(a.clone(), name.clone());
} else {
alias_to_event.insert(name.clone(), name.clone());
}
}
StreamSource::AllWithAlias { name, alias } => {
if let Some(a) = alias {
alias_to_event.insert(a.clone(), name.clone());
} else {
alias_to_event.insert(name.clone(), name.clone());
}
}
StreamSource::FromConnector { event_type, .. } => {
alias_to_event.insert(event_type.clone(), event_type.clone());
}
StreamSource::Merge(inline_streams) => {
for s in inline_streams {
alias_to_event.insert(s.name.clone(), s.source.clone());
if s.name != s.source {
alias_to_event.insert(s.source.clone(), s.source.clone());
}
}
}
StreamSource::Join(clauses) => {
for c in clauses {
alias_to_event.insert(c.source.clone(), c.source.clone());
}
}
StreamSource::Sequence(seq) => {
for step in &seq.steps {
alias_to_event.insert(step.alias.clone(), step.event_type.clone());
}
}
_ => {}
}
for op in ops {
if let StreamOp::FollowedBy(clause) = op {
if let Some(alias) = &clause.alias {
alias_to_event.insert(alias.clone(), clause.event_type.clone());
} else {
alias_to_event.insert(clause.event_type.clone(), clause.event_type.clone());
}
}
}
let mut bare_fields: Vec<String> = Vec::new();
for event_name in alias_to_event.values() {
if let Some(fields) = v.symbols.event_field_names(event_name) {
for f in fields {
if !bare_fields.contains(f) {
bare_fields.push(f.clone());
}
}
}
}
let has_forecast = ops.iter().any(|op| matches!(op, StreamOp::Forecast(_)));
let has_enrich = ops.iter().any(|op| matches!(op, StreamOp::Enrich(_)));
for (op_idx, op) in ops.iter().enumerate() {
let op_span = op_spans.get(op_idx).copied().unwrap_or(span);
match op {
StreamOp::Map(_) => {
v.emit_with_hint(
Severity::Error,
op_span,
"E090",
".map() is not implemented".to_string(),
"use .select() with expressions instead".to_string(),
);
}
StreamOp::Filter(_) => {
v.emit_with_hint(
Severity::Error,
op_span,
"E090",
".filter() is not implemented".to_string(),
"use .where() instead".to_string(),
);
}
StreamOp::Concurrent(ref args) => {
for arg in args {
if !crate::validate::builtins::CONCURRENT_PARAMS.contains(&arg.name.as_str()) {
v.emit(
Severity::Error,
op_span,
"E091",
format!(
".concurrent() unknown parameter '{}'; expected one of: {}",
arg.name,
crate::validate::builtins::CONCURRENT_PARAMS.join(", ")
),
);
}
}
for arg in args {
if arg.name == "workers" {
if let crate::ast::Expr::Int(n) = &arg.value {
if *n < 1 || *n > 128 {
v.emit(
Severity::Error,
op_span,
"E091",
format!(
".concurrent(workers: {n}) out of range; must be 1–128"
),
);
}
}
}
}
let mut found_concurrent = false;
for sop in ops {
if std::ptr::eq(sop, op) {
found_concurrent = true;
continue;
}
if found_concurrent {
let op_name = match sop {
StreamOp::Window { .. } => Some("Window"),
StreamOp::Aggregate(_) => Some("Aggregate"),
StreamOp::FollowedBy(_) => Some("Sequence"),
StreamOp::Forecast(_) => Some("Forecast"),
StreamOp::TrendAggregate(_) => Some("TrendAggregate"),
StreamOp::Distinct(_) => Some("Distinct"),
_ => None,
};
if let Some(name) = op_name {
v.emit(
Severity::Error,
op_span,
"E091",
format!(
".concurrent() cannot be followed by stateful .{}() operator",
name.to_lowercase()
),
);
break;
}
}
}
}
StreamOp::OnError(_) => {
v.emit_with_hint(
Severity::Error,
op_span,
"E090",
".on_error() is not yet implemented".to_string(),
"handle errors in your .where() or .select() logic".to_string(),
);
}
StreamOp::Collect => {
v.emit(
Severity::Error,
op_span,
"E090",
".collect() is not yet implemented".to_string(),
);
}
StreamOp::Fork(_) => {
v.emit(
Severity::Error,
op_span,
"E090",
".fork() is not yet implemented".to_string(),
);
}
StreamOp::Any(_) => {
v.emit(
Severity::Error,
op_span,
"E090",
".any() is not yet implemented".to_string(),
);
}
StreamOp::All => {
v.emit(
Severity::Error,
op_span,
"E090",
".all() is not yet implemented".to_string(),
);
}
StreamOp::First => {
v.emit(
Severity::Error,
op_span,
"E090",
".first() is not yet implemented".to_string(),
);
}
StreamOp::Distinct(_) => {
v.emit(
Severity::Error,
op_span,
"E090",
".distinct() is not yet implemented".to_string(),
);
}
StreamOp::OrderBy(_) => {
v.emit(
Severity::Error,
op_span,
"E090",
".order_by() is not yet implemented".to_string(),
);
}
StreamOp::Limit(_) => {
v.emit(
Severity::Error,
op_span,
"E090",
".limit() is not yet implemented".to_string(),
);
}
StreamOp::ToExpr(_) => {
v.emit_with_hint(
Severity::Error,
op_span,
"E090",
".to(expr) is not supported".to_string(),
"use .to(ConnectorName, ...) with a declared connector".to_string(),
);
}
StreamOp::Having(expr) => {
if !seen_aggregate {
v.emit_with_hint(
Severity::Error,
op_span,
"E010",
".having() used without a prior .aggregate()".to_string(),
"add .aggregate(...) before .having()".to_string(),
);
}
check_boolean_expr(v, expr, ".having()", op_span);
check_expr_field_refs(v, expr, &alias_to_event, op_span);
}
StreamOp::Aggregate(items) => {
if seen_aggregate {
v.emit(
Severity::Error,
op_span,
"E011",
"duplicate .aggregate() — only one aggregation per stream is allowed"
.to_string(),
);
}
if !seen_window {
v.emit_with_hint(
Severity::Warning,
op_span,
"W001",
".aggregate() without a prior .window()".to_string(),
"results will accumulate indefinitely; add .window() for bounded aggregation".to_string(),
);
}
seen_aggregate = true;
check_aggregate_items(v, items, op_span);
for item in items {
if !bare_fields.contains(&item.alias) {
bare_fields.push(item.alias.clone());
}
}
}
StreamOp::Window(_) => {
if seen_window {
v.emit(
Severity::Error,
op_span,
"E012",
"duplicate .window() — only one window per stream is allowed".to_string(),
);
}
seen_window = true;
}
StreamOp::PartitionBy(expr) => {
if seen_window {
v.emit_with_hint(
Severity::Warning,
op_span,
"W002",
".partition_by() after .window() — partitioning should come before windowing".to_string(),
"move .partition_by() before .window() for correct behavior".to_string(),
);
}
check_bare_ident_refs(
v,
expr,
&bare_fields,
&alias_to_event,
has_forecast,
has_enrich,
".partition_by()",
op_span,
);
}
StreamOp::Within(expr) => {
if !in_sequence {
v.emit_with_hint(
Severity::Error,
op_span,
"E020",
".within() used outside a sequence context".to_string(),
".within() requires a sequence source or -> (followed_by) operators"
.to_string(),
);
}
check_duration_expr(v, expr, ".within()", op_span);
}
StreamOp::FollowedBy(clause) | StreamOp::Not(clause) => {
check_source_name(v, &clause.event_type, op_span);
in_sequence = true;
}
StreamOp::Log(args) => {
check_named_params(v, args, LOG_PARAMS, ".log()", op_span);
}
StreamOp::Alert(args) => {
check_named_params(v, args, ALERT_PARAMS, ".alert()", op_span);
}
StreamOp::Watermark(args) => {
check_named_params(v, args, WATERMARK_PARAMS, ".watermark()", op_span);
}
StreamOp::To {
connector_name,
params,
} => {
if !v.symbols.connectors.contains_key(connector_name) {
let suggestion = did_you_mean(connector_name, &v.symbols.connector_names());
let available = v
.symbols
.connectors
.values()
.map(|c| c.connector_type.as_str())
.collect::<Vec<_>>();
let avail_hint = if available.is_empty() {
String::new()
} else {
format!(" (declared connector types: {})", available.join(", "))
};
v.emit_with_hint(
Severity::Error,
op_span,
"E030",
format!("undefined connector '{connector_name}'"),
format!(
"declare it with: connector {connector_name} = type (...){suggestion}{avail_hint}"
),
);
} else {
let connector_type =
v.symbols.connectors[connector_name].connector_type.clone();
check_connector_params(
v,
params,
&connector_type,
ParamContext::Sink,
".to()",
op_span,
);
}
}
StreamOp::Context(name) => {
if !v.symbols.contexts.contains_key(name) {
let suggestion = did_you_mean(name, &v.symbols.context_names());
v.emit_with_hint(
Severity::Error,
op_span,
"E031",
format!("undefined context '{name}'"),
format!("declare it with: context {name} (cores: [0, 1]){suggestion}"),
);
}
}
StreamOp::Where(expr) => {
check_boolean_expr(v, expr, ".where()", op_span);
check_expr_field_refs(v, expr, &alias_to_event, op_span);
check_bare_ident_refs(
v,
expr,
&bare_fields,
&alias_to_event,
has_forecast,
has_enrich,
".where()",
op_span,
);
}
StreamOp::AllowedLateness(expr) => {
check_duration_expr(v, expr, ".allowed_lateness()", op_span);
}
StreamOp::Emit {
output_type,
fields,
..
} => {
for field in fields {
check_expr_field_refs(v, &field.value, &alias_to_event, op_span);
}
if let Some(type_name) = output_type {
if let Some(event_info) = v.symbols.events.get(type_name) {
let _ = &event_info.field_names; } else if !v.symbols.is_declared(type_name) {
let suggestion = did_you_mean(type_name, &v.symbols.all_names());
v.emit_with_hint(
Severity::Error,
op_span,
"E034",
format!(".emit as '{type_name}' references an undeclared type"),
format!("declare it with: event {type_name} {{ ... }}{suggestion}"),
);
}
}
}
StreamOp::Enrich(spec) => {
if !v.symbols.connectors.contains_key(&spec.connector_name) {
let suggestion =
did_you_mean(&spec.connector_name, &v.symbols.connector_names());
v.emit_with_hint(
Severity::Error,
op_span,
"E030",
format!("undefined connector '{}'", spec.connector_name),
format!(
"declare it with: connector {} = type (...){}",
spec.connector_name, suggestion
),
);
} else {
let connector_type = &v.symbols.connectors[&spec.connector_name].connector_type;
if !builtins::ENRICH_COMPATIBLE_TYPES.contains(&connector_type.as_str()) {
v.emit_with_hint(
Severity::Error,
op_span,
"E032",
format!(
".enrich() is not compatible with '{}' connector type '{}'",
spec.connector_name, connector_type
),
format!(
".enrich() requires a request-response connector ({})",
builtins::ENRICH_COMPATIBLE_TYPES.join(", ")
),
);
}
}
if spec.fields.is_empty() {
v.emit_with_hint(
Severity::Warning,
op_span,
"W032",
".enrich() has no fields specified".to_string(),
"add fields: [field1, field2] to extract data from the enrichment response"
.to_string(),
);
}
}
StreamOp::TrendAggregate(items) => {
for item in items {
if !bare_fields.contains(&item.alias) {
bare_fields.push(item.alias.clone());
}
}
}
StreamOp::Tap(_)
| StreamOp::Print(_)
| StreamOp::Select(_)
| StreamOp::Pattern(_)
| StreamOp::Process(_)
| StreamOp::On(_)
| StreamOp::Score(_)
| StreamOp::Forecast(_)
| StreamOp::SelectionMode(_)
| StreamOp::EmissionMode(_) => {}
}
}
}
const fn is_sequence_source(source: &StreamSource) -> bool {
matches!(source, StreamSource::Sequence(_))
}
fn collect_member_chain(expr: &Expr) -> Option<(String, Vec<String>)> {
match expr {
Expr::Member {
expr: inner,
member,
} => {
if let Expr::Ident(root) = inner.as_ref() {
Some((root.clone(), vec![member.clone()]))
} else if let Some((root, mut chain)) = collect_member_chain(inner) {
chain.push(member.clone());
Some((root, chain))
} else {
None
}
}
_ => None,
}
}
fn check_expr_field_refs(
v: &mut Validator,
expr: &Expr,
alias_to_event: &HashMap<String, String>,
span: Span,
) {
match expr {
Expr::Member {
expr: inner,
member,
} => {
if let Some((root, chain)) = collect_member_chain(expr) {
if let Some(event_name) = alias_to_event.get(&root) {
if chain[0] != "LEN" {
if let Some(fields) = v.symbols.event_field_names(event_name) {
if !fields.is_empty() && !fields.iter().any(|f| f == &chain[0]) {
let suggestion = did_you_mean(
&chain[0],
&fields.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
);
v.emit_with_hint(
Severity::Warning,
span,
"W034",
format!(
"reference to undeclared field '{}' on event '{event_name}'",
chain[0]
),
format!(
"declared fields: {}{}",
fields.join(", "),
suggestion
),
);
}
}
}
let mut current_type_name: Option<String> = None;
if let Some(event_info) = v.symbols.events.get(event_name) {
let _ = event_info;
}
if let Some(type_info) = v.symbols.types.get(event_name) {
for (fname, fty) in &type_info.fields {
if fname == &chain[0] {
if let crate::types::Type::Named(ref n) = fty {
current_type_name = Some(n.clone());
}
break;
}
}
}
for field in chain.iter().skip(1) {
let type_name = match current_type_name.take() {
Some(n) => n,
None => break,
};
let resolved = v.symbols.types.get(&type_name).map(|ti| {
let names: Vec<String> =
ti.fields.iter().map(|(n, _)| n.clone()).collect();
let next = ti.fields.iter().find_map(|(n, t)| {
if n == field {
if let crate::types::Type::Named(ref tn) = t {
Some(tn.clone())
} else {
None
}
} else {
None
}
});
(names, next)
});
match resolved {
Some((type_fields, next_type)) => {
if !type_fields.is_empty()
&& !type_fields.iter().any(|f| f == field)
{
let suggestion = did_you_mean(
field,
&type_fields.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
);
v.emit_with_hint(
Severity::Warning,
span,
"W034",
format!(
"reference to undeclared field '{field}' on type '{type_name}'"
),
format!(
"declared fields: {}{}",
type_fields.join(", "),
suggestion
),
);
}
current_type_name = next_type;
}
None => break, }
}
}
} else if let Expr::Ident(name) = inner.as_ref() {
if let Some(event_name) = alias_to_event.get(name) {
if let Some(fields) = v.symbols.event_field_names(event_name) {
if !fields.is_empty() && !fields.iter().any(|f| f == member) {
let suggestion = did_you_mean(
member,
&fields.iter().map(|s| s.as_str()).collect::<Vec<_>>(),
);
v.emit_with_hint(
Severity::Warning,
span,
"W034",
format!(
"reference to undeclared field '{member}' on event '{event_name}'"
),
format!("declared fields: {}{}", fields.join(", "), suggestion),
);
}
}
}
}
check_expr_field_refs(v, inner, alias_to_event, span);
}
Expr::Binary { left, right, .. } => {
check_expr_field_refs(v, left, alias_to_event, span);
check_expr_field_refs(v, right, alias_to_event, span);
}
Expr::Unary { expr: inner, .. } => {
check_expr_field_refs(v, inner, alias_to_event, span);
}
Expr::Call { func, args } => {
check_expr_field_refs(v, func, alias_to_event, span);
for arg in args {
match arg {
Arg::Positional(e) | Arg::Named(_, e) => {
check_expr_field_refs(v, e, alias_to_event, span);
}
}
}
}
Expr::OptionalMember { expr: inner, .. } => {
check_expr_field_refs(v, inner, alias_to_event, span);
}
Expr::Index { expr: e, index } => {
check_expr_field_refs(v, e, alias_to_event, span);
check_expr_field_refs(v, index, alias_to_event, span);
}
Expr::If {
cond,
then_branch,
else_branch,
} => {
check_expr_field_refs(v, cond, alias_to_event, span);
check_expr_field_refs(v, then_branch, alias_to_event, span);
check_expr_field_refs(v, else_branch, alias_to_event, span);
}
Expr::Coalesce { expr: e, default } => {
check_expr_field_refs(v, e, alias_to_event, span);
check_expr_field_refs(v, default, alias_to_event, span);
}
Expr::Array(elems) => {
for e in elems {
check_expr_field_refs(v, e, alias_to_event, span);
}
}
_ => {}
}
}
#[allow(clippy::too_many_arguments)]
fn check_bare_ident_refs(
v: &mut Validator,
expr: &Expr,
bare_fields: &[String],
alias_to_event: &HashMap<String, String>,
has_forecast: bool,
has_enrich: bool,
context: &str,
span: Span,
) {
if bare_fields.is_empty() && !has_forecast && !has_enrich {
return;
}
match expr {
Expr::Ident(name) => {
if alias_to_event.contains_key(name) {
return;
}
if bare_fields.iter().any(|f| f == name) {
return;
}
if v.symbols.variables.contains_key(name) {
return;
}
if builtins::is_known_function(name) {
return;
}
if matches!(name.as_str(), "true" | "false" | "null") {
return;
}
if has_forecast && builtins::FORECAST_BUILTIN_VARS.contains(&name.as_str()) {
return;
}
if has_enrich && builtins::ENRICH_BUILTIN_VARS.contains(&name.as_str()) {
return;
}
let mut candidates: Vec<&str> = bare_fields.iter().map(|s| s.as_str()).collect();
for alias in alias_to_event.keys() {
candidates.push(alias);
}
if has_forecast {
candidates.extend(builtins::FORECAST_BUILTIN_VARS);
}
if has_enrich {
candidates.extend(builtins::ENRICH_BUILTIN_VARS);
}
let suggestion = did_you_mean(name, &candidates);
v.emit_with_hint(
Severity::Warning,
span,
"W035",
format!("unknown field '{name}' in {context}"),
format!("available fields: {}{}", candidates.join(", "), suggestion),
);
}
Expr::Member { .. } => {}
Expr::Binary { left, right, .. } => {
check_bare_ident_refs(
v,
left,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
check_bare_ident_refs(
v,
right,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
}
Expr::Unary { expr: inner, .. } => {
check_bare_ident_refs(
v,
inner,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
}
Expr::Call { args, .. } => {
for arg in args {
match arg {
Arg::Positional(e) | Arg::Named(_, e) => {
check_bare_ident_refs(
v,
e,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
}
}
}
}
Expr::If {
cond,
then_branch,
else_branch,
} => {
check_bare_ident_refs(
v,
cond,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
check_bare_ident_refs(
v,
then_branch,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
check_bare_ident_refs(
v,
else_branch,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
}
Expr::Index { expr: e, index } => {
check_bare_ident_refs(
v,
e,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
check_bare_ident_refs(
v,
index,
bare_fields,
alias_to_event,
has_forecast,
has_enrich,
context,
span,
);
}
_ => {}
}
}
fn check_boolean_expr(v: &mut Validator, expr: &Expr, context: &str, span: Span) {
match expr {
Expr::Int(_)
| Expr::Float(_)
| Expr::Str(_)
| Expr::Array(_)
| Expr::Map(_)
| Expr::Null
| Expr::Duration(_)
| Expr::Timestamp(_) => {
v.emit_with_hint(
Severity::Error,
span,
"E060",
format!(
"{} condition must be a boolean expression, got {} literal",
context,
literal_type_name(expr)
),
"use a comparison like field > value or a boolean expression".to_string(),
);
}
Expr::Ident(name) if !matches!(name.as_str(), "true" | "false") => {
v.emit_with_hint(
Severity::Warning,
span,
"W061",
format!(
"{context} condition is a bare identifier '{name}', expected a boolean expression"
),
format!(
"use a comparison like {name} > value or {name} == value"
),
);
}
Expr::Member {
expr: inner,
member,
} => {
if let Expr::Ident(obj) = inner.as_ref() {
v.emit_with_hint(
Severity::Warning,
span,
"W061",
format!(
"{context} condition is a field access '{obj}.{member}', expected a boolean expression"
),
format!(
"use a comparison like {obj}.{member} > value or {obj}.{member} == value"
),
);
}
}
Expr::Binary { op, .. }
if matches!(
op,
BinOp::Add | BinOp::Sub | BinOp::Mul | BinOp::Div | BinOp::Mod | BinOp::Pow
) =>
{
v.emit_with_hint(
Severity::Warning,
span,
"W060",
format!(
"{} condition is an arithmetic expression ({}), expected boolean",
context,
op.as_str()
),
"use a comparison operator (==, !=, <, >, <=, >=)".to_string(),
);
}
_ => {} }
}
fn check_duration_expr(v: &mut Validator, expr: &Expr, context: &str, span: Span) {
match expr {
Expr::Duration(_) | Expr::Ident(_) | Expr::Member { .. } | Expr::Call { .. } => {}
Expr::Int(_) => {} Expr::Str(_) | Expr::Bool(_) | Expr::Float(_) | Expr::Array(_) | Expr::Null => {
v.emit_with_hint(
Severity::Error,
span,
"E061",
format!(
"{} must be a duration, got {} literal",
context,
literal_type_name(expr)
),
"use a duration like 5s, 1m, 1h".to_string(),
);
}
_ => {} }
}
const fn literal_type_name(expr: &Expr) -> &'static str {
match expr {
Expr::Int(_) => "integer",
Expr::Float(_) => "float",
Expr::Str(_) => "string",
Expr::Bool(_) => "boolean",
Expr::Null => "null",
Expr::Duration(_) => "duration",
Expr::Timestamp(_) => "timestamp",
Expr::Array(_) => "array",
Expr::Map(_) => "map",
_ => "expression",
}
}
fn check_connector_params(
v: &mut Validator,
params: &[ConnectorParam],
connector_type: &str,
ctx: builtins::ParamContext,
op_name: &str,
span: Span,
) {
let schema = match builtins::connector_params_for_type(connector_type) {
Some(s) => s,
None => return, };
let valid_names: Vec<&str> = schema
.iter()
.filter(|p| p.valid_in(ctx))
.map(|p| p.name)
.collect();
for param in params {
let def = schema.iter().find(|d| d.name == param.name);
match def {
None => {
let suggestion = did_you_mean(¶m.name, &valid_names);
v.emit_with_hint(
Severity::Warning,
span,
"W080",
format!(
"unknown parameter '{}' for {} connector in {}",
param.name, connector_type, op_name
),
format!("valid parameters: {}{}", valid_names.join(", "), suggestion),
);
}
Some(def) => {
if !def.valid_in(ctx) {
let ctx_name = match ctx {
builtins::ParamContext::Source => "source (.from())",
builtins::ParamContext::Sink => "sink (.to())",
builtins::ParamContext::Both => "both",
};
v.emit_with_hint(
Severity::Warning,
span,
"W080",
format!(
"parameter '{}' is not valid in {} context",
param.name, ctx_name
),
format!(
"'{}' is only valid for {}",
param.name,
match def.context {
builtins::ParamContext::Source => ".from() (source)",
builtins::ParamContext::Sink => ".to() (sink)",
builtins::ParamContext::Both => "both",
}
),
);
}
let type_ok = match def.param_type {
builtins::ParamType::Str => matches!(
param.value,
crate::ast::ConfigValue::Str(_)
| crate::ast::ConfigValue::Ident(_)
| crate::ast::ConfigValue::Concat(_)
),
builtins::ParamType::Int => {
matches!(param.value, crate::ast::ConfigValue::Int(_))
}
builtins::ParamType::Bool => {
matches!(param.value, crate::ast::ConfigValue::Bool(_))
}
builtins::ParamType::StrArray => {
matches!(param.value, crate::ast::ConfigValue::Array(_))
}
};
if !type_ok {
let expected = match def.param_type {
builtins::ParamType::Str => "string",
builtins::ParamType::Int => "integer",
builtins::ParamType::Bool => "boolean",
builtins::ParamType::StrArray => "array of strings",
};
v.emit_with_hint(
Severity::Warning,
span,
"W081",
format!("parameter '{}' expects {} value", param.name, expected),
format!("{}: {}", def.name, def.description),
);
}
}
}
}
for def in schema.iter().filter(|p| p.required && p.valid_in(ctx)) {
if !params.iter().any(|p| p.name == def.name) {
v.emit_with_hint(
Severity::Error,
span,
"E009",
format!(
"missing required parameter '{}' for {} {}",
def.name, connector_type, op_name
),
def.description.to_string(),
);
}
}
}
fn check_named_params(
v: &mut Validator,
args: &[NamedArg],
valid: &[&str],
context: &str,
span: Span,
) {
for arg in args {
if !valid.contains(&arg.name.as_str()) {
let suggestion = did_you_mean(&arg.name, valid);
v.emit_with_hint(
Severity::Error,
span,
"E080",
format!("unknown parameter '{}' for {}", arg.name, context),
format!("valid parameters: {}{}", valid.join(", "), suggestion),
);
}
}
}
fn check_aggregate_items(v: &mut Validator, items: &[AggItem], span: Span) {
for item in items {
match &item.expr {
Expr::Call { func, args } => {
if let Some(func_name) = extract_ident(func) {
if !builtins::is_aggregate_function(&func_name) {
let suggestion = did_you_mean(&func_name, AGGREGATE_FUNCTIONS);
v.emit_with_hint(
Severity::Error,
span,
"E070",
format!(
"unknown aggregate function '{}' in alias '{}'",
func_name, item.alias
),
format!(
"known aggregate functions: {}{}",
AGGREGATE_FUNCTIONS.join(", "),
suggestion
),
);
continue;
}
if AGGREGATE_REQUIRES_FIELD.contains(&func_name.as_str()) && args.is_empty() {
v.emit_with_hint(
Severity::Error,
span,
"E071",
format!("aggregate function '{func_name}' requires a field argument"),
format!("usage: {func_name}(field_name)"),
);
}
if AGGREGATE_REQUIRES_TWO_ARGS.contains(&func_name.as_str()) && args.len() < 2 {
v.emit_with_hint(
Severity::Error,
span,
"E072",
format!(
"aggregate function '{func_name}' requires two arguments: field and period"
),
format!("usage: {func_name}(field_name, period)"),
);
}
}
}
Expr::Ident(name) => {
v.emit_with_hint(
Severity::Error,
span,
"E073",
format!(
"bare field reference '{name}' in aggregate without an aggregate function"
),
format!(
"wrap in an aggregate function, e.g. last({name}), first({name}), or sum({name})"
),
);
}
_ => {
}
}
}
}
fn extract_ident(expr: &Expr) -> Option<String> {
match expr {
Expr::Ident(name) => Some(name.clone()),
_ => None,
}
}
fn check_sase_pattern_refs(v: &mut Validator, expr: &SasePatternExpr, span: Span) {
let SasePatternExpr::Seq(items) = expr;
for item in items {
let name = item
.event_type
.strip_prefix('!')
.unwrap_or(&item.event_type);
check_source_name(v, name, span);
}
}
fn check_function_call(v: &mut Validator, name: &str, args_len: usize, span: Span) {
if let Some(func_info) = v.symbols.functions.get(name) {
if args_len != func_info.param_count {
v.emit_with_related(
Severity::Error,
span,
"E051",
format!(
"function '{}' expects {} argument(s), but {} provided",
name, func_info.param_count, args_len
),
vec![RelatedSpan {
span: func_info.span,
message: "function declared here".to_string(),
}],
);
}
return;
}
if builtins::is_known_function(name) {
return;
}
let mut candidates: Vec<&str> = builtins::BUILTIN_FUNCTIONS.to_vec();
candidates.extend(builtins::AGGREGATE_FUNCTIONS);
candidates.extend(v.symbols.function_names());
let suggestion = suggest(name, &candidates);
let hint = match suggestion {
Some(s) => format!("did you mean '{s}'?"),
None => "check the function name or declare it with fn".to_string(),
};
v.emit_with_hint(
Severity::Error,
span,
"E050",
format!("unknown function '{name}'"),
hint,
);
}
pub fn check_expr_functions(v: &mut Validator, expr: &Expr, span: Span) {
match expr {
Expr::Call { func, args } => {
if let Some(name) = extract_ident(func) {
check_function_call(v, &name, count_positional_args(args), span);
}
for arg in args {
match arg {
Arg::Positional(e) | Arg::Named(_, e) => check_expr_functions(v, e, span),
}
}
}
Expr::Binary { left, right, .. } => {
check_expr_functions(v, left, span);
check_expr_functions(v, right, span);
}
Expr::Unary { expr: inner, .. } => {
check_expr_functions(v, inner, span);
}
Expr::Member { expr: inner, .. } | Expr::OptionalMember { expr: inner, .. } => {
check_expr_functions(v, inner, span);
}
Expr::Index { expr: e, index } => {
check_expr_functions(v, e, span);
check_expr_functions(v, index, span);
}
Expr::Slice {
expr: e,
start,
end,
} => {
check_expr_functions(v, e, span);
if let Some(s) = start {
check_expr_functions(v, s, span);
}
if let Some(e) = end {
check_expr_functions(v, e, span);
}
}
Expr::If {
cond,
then_branch,
else_branch,
} => {
check_expr_functions(v, cond, span);
check_expr_functions(v, then_branch, span);
check_expr_functions(v, else_branch, span);
}
Expr::Coalesce { expr: e, default } => {
check_expr_functions(v, e, span);
check_expr_functions(v, default, span);
}
Expr::Array(elems) => {
for e in elems {
check_expr_functions(v, e, span);
}
}
Expr::Map(entries) => {
for (_, e) in entries {
check_expr_functions(v, e, span);
}
}
Expr::Lambda { body, .. } => {
check_expr_functions(v, body, span);
}
Expr::Range { start, end, .. } => {
check_expr_functions(v, start, span);
check_expr_functions(v, end, span);
}
Expr::Block { stmts, result } => {
for (_, _, val, _) in stmts {
check_expr_functions(v, val, span);
}
check_expr_functions(v, result, span);
}
Expr::Null
| Expr::Bool(_)
| Expr::Int(_)
| Expr::Float(_)
| Expr::Str(_)
| Expr::Duration(_)
| Expr::Timestamp(_)
| Expr::Ident(_) => {}
}
}
fn count_positional_args(args: &[Arg]) -> usize {
args.iter()
.filter(|a| matches!(a, Arg::Positional(_)))
.count()
}