use serde::{Deserialize, Serialize};
use serde_json::Value;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(untagged)]
pub enum EventPattern {
Specific(String),
Wildcard,
}
impl EventPattern {
pub fn matches(&self, event_type_id: &str) -> bool {
match self {
EventPattern::Specific(pattern) => pattern == event_type_id,
EventPattern::Wildcard => true,
}
}
pub fn wildcard() -> Self {
EventPattern::Wildcard
}
pub fn specific(event_type_id: impl Into<String>) -> Self {
EventPattern::Specific(event_type_id.into())
}
}
impl From<String> for EventPattern {
fn from(s: String) -> Self {
if s == "*" || s == "all" {
EventPattern::Wildcard
} else {
EventPattern::Specific(s)
}
}
}
impl From<&str> for EventPattern {
fn from(s: &str) -> Self {
if s == "*" || s == "all" {
EventPattern::Wildcard
} else {
EventPattern::Specific(s.to_string())
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)]
pub enum EventHandlerType {
Llm {
instruction: String,
},
Script {
language: String,
code: String,
#[serde(default)]
resident: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
scope: Option<String>,
},
Static {
actions: Vec<serde_json::Value>,
},
Manual {
#[serde(default = "default_manual_timeout_secs")]
timeout_secs: u64,
},
}
pub const DEFAULT_MANUAL_TIMEOUT_SECS: u64 = 300;
fn default_manual_timeout_secs() -> u64 {
DEFAULT_MANUAL_TIMEOUT_SECS
}
impl EventHandlerType {
pub fn script(language: impl Into<String>, code: impl Into<String>) -> Self {
EventHandlerType::Script {
language: language.into(),
code: code.into(),
resident: false,
scope: None,
}
}
pub fn script_resident(
language: impl Into<String>,
code: impl Into<String>,
scope: Option<String>,
) -> Self {
EventHandlerType::Script {
language: language.into(),
code: code.into(),
resident: true,
scope,
}
}
pub fn static_response(actions: Vec<serde_json::Value>) -> Self {
EventHandlerType::Static { actions }
}
pub fn llm(instruction: impl Into<String>) -> Self {
EventHandlerType::Llm {
instruction: instruction.into(),
}
}
pub fn manual(timeout_secs: u64) -> Self {
EventHandlerType::Manual { timeout_secs }
}
pub fn validate(&self) -> Result<(), InterpolationError> {
if let EventHandlerType::Script { code, .. } = self {
if code.len() > super::types::ScriptSource::MAX_CODE_BYTES {
return Err(budget_error());
}
}
match self {
EventHandlerType::Static { actions } => {
for action in actions {
validate_event_references(action)?;
}
Ok(())
}
EventHandlerType::Script {
language,
code,
resident,
..
} if !*resident && defines_handle(language, code) => {
Err(InterpolationError::script_contract(language.clone()))
}
_ => Ok(()),
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct EventHandler {
pub event_pattern: EventPattern,
pub handler: EventHandlerType,
}
impl EventHandler {
pub fn new(event_pattern: EventPattern, handler: EventHandlerType) -> Self {
Self {
event_pattern,
handler,
}
}
pub fn matches(&self, event_type_id: &str) -> bool {
self.event_pattern.matches(event_type_id)
}
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct EventHandlerConfig {
pub handlers: Vec<EventHandler>,
}
impl EventHandlerConfig {
pub fn new() -> Self {
Self {
handlers: Vec::new(),
}
}
pub fn add_handler(&mut self, handler: EventHandler) {
self.handlers.push(handler);
}
pub fn find_handler(&self, event_type_id: &str) -> Option<&EventHandlerType> {
self.handlers
.iter()
.find(|h| h.matches(event_type_id))
.map(|h| &h.handler)
}
pub fn is_empty(&self) -> bool {
self.handlers.is_empty()
}
pub fn len(&self) -> usize {
self.handlers.len()
}
}
const REF_OPEN: &str = "{{";
const REF_CLOSE: &str = "}}";
const REF_ROOT: &str = "event";
const REF_ROOT_DOT: &str = "event.";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct InterpolationError {
pub reference: String,
pub detail: String,
pub kind: ErrorKind,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ErrorKind {
Interpolation,
ScriptContract,
}
impl InterpolationError {
fn new(reference: impl Into<String>, detail: impl Into<String>) -> Self {
Self {
reference: reference.into(),
detail: detail.into(),
kind: ErrorKind::Interpolation,
}
}
fn script_contract(language: impl Into<String>) -> Self {
Self {
reference: language.into(),
detail: String::new(),
kind: ErrorKind::ScriptContract,
}
}
}
impl std::fmt::Display for InterpolationError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self.kind {
ErrorKind::Interpolation => write!(
f,
"static handler reference `{}` could not be resolved: {}",
self.reference, self.detail
),
ErrorKind::ScriptContract => write!(
f,
"this {} script defines `handle(...)`, which is the RESIDENT entry point, \
but the handler is not marked `resident: true`. As written the script \
reads nothing from stdin and prints nothing, so the handler produces no \
actions and the server fails closed -- with no hint of why. Add \
`\"resident\": true`, or rewrite the script to read the event from stdin \
and print its actions.",
self.reference
),
}
}
}
impl std::error::Error for InterpolationError {}
struct FoundRef<'a> {
start: usize,
end: usize,
raw: &'a str,
inner: &'a str,
}
fn find_reference(s: &str, from: usize) -> Option<FoundRef<'_>> {
let mut cursor = from;
while cursor < s.len() {
let open = cursor + s[cursor..].find(REF_OPEN)?;
let after_open = open + REF_OPEN.len();
let Some(rel_close) = s[after_open..].find(REF_CLOSE) else {
return None;
};
let close = after_open + rel_close;
let inner = s[after_open..close].trim();
if inner == REF_ROOT || inner.starts_with(REF_ROOT_DOT) {
return Some(FoundRef {
start: open,
end: close + REF_CLOSE.len(),
raw: &s[open..close + REF_CLOSE.len()],
inner,
});
}
cursor = open + 1;
}
None
}
fn parse_path<'a>(found: &FoundRef<'a>) -> Result<Vec<&'a str>, InterpolationError> {
if found.inner == REF_ROOT {
return Ok(Vec::new());
}
let rest = &found.inner[REF_ROOT_DOT.len()..];
if rest.is_empty() {
return Err(InterpolationError::new(
found.raw,
format!(
"the path after `{}.` is empty; write `{{{{event.field}}}}` or `{{{{event}}}}` for the whole payload",
REF_ROOT
),
));
}
let mut segments = Vec::new();
for segment in rest.split('.') {
let segment = segment.trim();
if segment.is_empty() {
return Err(InterpolationError::new(
found.raw,
"it contains an empty path segment (a doubled or trailing `.`)",
));
}
segments.push(segment);
}
Ok(segments)
}
fn describe_available(value: &Value) -> String {
match value {
Value::Object(map) if map.is_empty() => "this object is empty".to_string(),
Value::Object(map) => format!(
"available fields: {}",
map.keys()
.take(16)
.map(|key| key.chars().take(64).collect::<String>())
.collect::<Vec<_>>()
.join(", ")
),
Value::Array(items) if items.is_empty() => "this array is empty".to_string(),
Value::Array(items) => format!("available indices: 0..{}", items.len() - 1),
other => format!("it is a {} and has no fields", json_type_name(other)),
}
}
fn json_type_name(value: &Value) -> &'static str {
match value {
Value::Null => "null",
Value::Bool(_) => "boolean",
Value::Number(_) => "number",
Value::String(_) => "string",
Value::Array(_) => "array",
Value::Object(_) => "object",
}
}
fn resolve<'a>(
event: &'a Value,
segments: &[&str],
raw: &str,
) -> Result<&'a Value, InterpolationError> {
let mut current = event;
let mut walked = REF_ROOT.to_string();
for segment in segments {
let next = match current {
Value::Object(map) => map.get(*segment),
Value::Array(items) => segment.parse::<usize>().ok().and_then(|i| items.get(i)),
_ => None,
};
current = next.ok_or_else(|| {
InterpolationError::new(
raw,
format!(
"`{}` has no `{}` ({})",
walked,
segment,
describe_available(current)
),
)
})?;
walked.push('.');
walked.push_str(segment);
}
Ok(current)
}
pub const MAX_INTERPOLATION_BYTES: usize = 8 * 1024 * 1024;
pub const MAX_INTERPOLATION_NODES: usize = 65_536;
pub const MAX_INTERPOLATION_DEPTH: usize = 64;
fn budget_error() -> InterpolationError {
InterpolationError::new(
"event",
"static handler exceeds the 8 MiB, 65536-node or 64-level interpolation budget",
)
}
fn check_tree(value: &Value) -> Result<(), InterpolationError> {
if crate::utils::json_budget::within_budget(
value,
MAX_INTERPOLATION_BYTES,
MAX_INTERPOLATION_NODES,
MAX_INTERPOLATION_DEPTH,
) {
Ok(())
} else {
Err(budget_error())
}
}
struct ExpansionBudget {
bytes: usize,
nodes: usize,
}
impl ExpansionBudget {
fn append(&mut self, out: &mut String, text: &str) -> Result<(), InterpolationError> {
if text.len() > MAX_INTERPOLATION_BYTES.saturating_sub(self.bytes) {
return Err(budget_error());
}
self.bytes += text.len();
out.push_str(text);
Ok(())
}
fn visit(&mut self) -> Result<(), InterpolationError> {
self.nodes += 1;
if self.nodes > MAX_INTERPOLATION_NODES {
return Err(budget_error());
}
Ok(())
}
}
fn bounded_display(
value: &Value,
budget: &mut ExpansionBudget,
) -> Result<String, InterpolationError> {
check_tree(value)?;
if let Value::String(text) = value {
let mut result = String::new();
budget.append(&mut result, text)?;
return Ok(result);
}
struct Writer<'a> {
data: Vec<u8>,
budget: &'a mut ExpansionBudget,
}
impl std::io::Write for Writer<'_> {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
if bytes.len() > MAX_INTERPOLATION_BYTES.saturating_sub(self.budget.bytes) {
return Err(std::io::Error::other("interpolation byte cap"));
}
self.budget.bytes += bytes.len();
self.data.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
let mut writer = Writer {
data: Vec::new(),
budget,
};
serde_json::to_writer(&mut writer, value).map_err(|_| budget_error())?;
String::from_utf8(writer.data).map_err(|_| budget_error())
}
fn clone_bounded(value: &Value, budget: &mut ExpansionBudget) -> Result<Value, InterpolationError> {
check_tree(value)?;
fn copy(value: &Value, budget: &mut ExpansionBudget) -> Result<Value, InterpolationError> {
budget.visit()?;
match value {
Value::String(text) => {
let mut out = String::new();
budget.append(&mut out, text)?;
Ok(Value::String(out))
}
Value::Array(items) => items
.iter()
.map(|v| copy(v, budget))
.collect::<Result<Vec<_>, _>>()
.map(Value::Array),
Value::Object(map) => {
let mut out = serde_json::Map::new();
for (key, value) in map {
let mut owned = String::new();
budget.append(&mut owned, key)?;
out.insert(owned, copy(value, budget)?);
}
Ok(Value::Object(out))
}
other => Ok(other.clone()),
}
}
copy(value, budget)
}
fn missing_event_error(raw: &str) -> InterpolationError {
InterpolationError::new(
raw,
"this event carries no structured data, so nothing can be substituted",
)
}
fn interpolate_string(
s: &str,
event: Option<&Value>,
budget: &mut ExpansionBudget,
) -> Result<Value, InterpolationError> {
let Some(first) = find_reference(s, 0) else {
let mut out = String::new();
budget.append(&mut out, s)?;
return Ok(Value::String(out));
};
if first.start == 0 && first.end == s.len() {
let segments = parse_path(&first)?;
let event = event.ok_or_else(|| missing_event_error(first.raw))?;
return clone_bounded(resolve(event, &segments, first.raw)?, budget);
}
let mut out = String::new();
let mut cursor = 0usize;
let mut found = Some(first);
while let Some(f) = found {
budget.append(&mut out, &s[cursor..f.start])?;
let segments = parse_path(&f)?;
let event = event.ok_or_else(|| missing_event_error(f.raw))?;
out.push_str(&bounded_display(resolve(event, &segments, f.raw)?, budget)?);
cursor = f.end;
found = find_reference(s, cursor);
}
budget.append(&mut out, &s[cursor..])?;
Ok(Value::String(out))
}
pub fn interpolate_value(
value: &Value,
event_data: Option<&Value>,
) -> Result<Value, InterpolationError> {
check_tree(value)?;
let result = interpolate_inner(
value,
event_data,
&mut ExpansionBudget { bytes: 0, nodes: 0 },
)?;
if check_tree(&result).is_err() {
crate::utils::json_budget::drop_iteratively(result);
return Err(budget_error());
}
Ok(result)
}
fn interpolate_inner(
value: &Value,
event_data: Option<&Value>,
budget: &mut ExpansionBudget,
) -> Result<Value, InterpolationError> {
budget.visit()?;
match value {
Value::String(s) => interpolate_string(s, event_data, budget),
Value::Array(items) => items
.iter()
.map(|item| interpolate_inner(item, event_data, budget))
.collect::<Result<Vec<_>, _>>()
.map(Value::Array),
Value::Object(map) => {
let mut out = serde_json::Map::new();
for (key, val) in map {
let key = match interpolate_string(key, event_data, budget)? {
Value::String(s) => s,
other => bounded_display(&other, budget)?,
};
if out.contains_key(&key) {
return Err(InterpolationError::new(key, "interpolated keys collide"));
}
out.insert(key, interpolate_inner(val, event_data, budget)?);
}
Ok(Value::Object(out))
}
other => Ok(other.clone()),
}
}
pub fn interpolate_actions(
actions: &[Value],
event_data: Option<&Value>,
) -> Result<Vec<Value>, InterpolationError> {
if actions.len() > MAX_INTERPOLATION_NODES {
return Err(budget_error());
}
let mut budget = ExpansionBudget { bytes: 0, nodes: 0 };
let expanded = actions
.iter()
.map(|action| {
check_tree(action)?;
interpolate_inner(action, event_data, &mut budget)
})
.collect::<Result<Vec<_>, _>>()?;
let combined = Value::Array(expanded);
if check_tree(&combined).is_err() {
crate::utils::json_budget::drop_iteratively(combined);
return Err(budget_error());
}
let Value::Array(expanded) = combined else {
unreachable!()
};
Ok(expanded)
}
pub fn contains_event_reference(value: &Value) -> bool {
if check_tree(value).is_err() {
return false;
}
let mut pending = vec![value];
while let Some(value) = pending.pop() {
match value {
Value::String(s) if find_reference(s, 0).is_some() => return true,
Value::Array(items) => pending.extend(items),
Value::Object(map) => {
if map.keys().any(|key| find_reference(key, 0).is_some()) {
return true;
}
pending.extend(map.values());
}
_ => {}
}
}
false
}
fn defines_handle(language: &str, code: &str) -> bool {
let language = language.to_ascii_lowercase();
let python = matches!(language.as_str(), "python" | "python3");
let perl = language == "perl";
let javascript = matches!(language.as_str(), "javascript" | "js" | "node");
if !python && !perl && !javascript && language != "go" {
return false;
}
let bytes = code.as_bytes();
let mut tokens: Vec<&str> = Vec::new();
let mut i = 0;
while i < bytes.len() {
if bytes[i].is_ascii_whitespace() {
i += 1;
continue;
}
if ((python || perl) && bytes[i] == b'#')
|| (!python && !perl && bytes[i..].starts_with(b"//"))
{
while i < bytes.len() && bytes[i] != b'\n' {
i += 1;
}
continue;
}
if !python && !perl && bytes[i..].starts_with(b"/*") {
i += 2;
while i < bytes.len() && !bytes[i..].starts_with(b"*/") {
i += 1;
}
i = (i + 2).min(bytes.len());
continue;
}
if perl && bytes[i] == b'q' {
let mut delimiter = i + 1;
if bytes
.get(delimiter)
.is_some_and(|b| matches!(*b, b'q' | b'w' | b'x' | b'r'))
{
delimiter += 1;
}
while bytes.get(delimiter).is_some_and(u8::is_ascii_whitespace) {
delimiter += 1;
}
if let Some(&open) = bytes
.get(delimiter)
.filter(|b| !b.is_ascii_alphanumeric() && **b != b'_')
{
let close = match open {
b'{' => b'}',
b'[' => b']',
b'(' => b')',
b'<' => b'>',
other => other,
};
let mut depth = 1;
i = delimiter + 1;
while i < bytes.len() && depth > 0 {
if bytes[i] == b'\\' {
i = (i + 2).min(bytes.len());
continue;
}
if bytes[i] == close {
depth -= 1;
} else if open != close && bytes[i] == open {
depth += 1;
}
i += 1;
}
tokens.push("<literal>");
continue;
}
}
if matches!(bytes[i], b'\'' | b'"' | b'`') {
let quote = bytes[i];
let triple = python && bytes[i..].starts_with(&[quote; 3]);
let size = if triple { 3 } else { 1 };
i += size;
while i < bytes.len() {
if bytes[i] == b'\\' {
i = (i + 2).min(bytes.len());
continue;
}
if bytes[i] == quote && (!triple || bytes[i..].starts_with(&[quote; 3])) {
i += size;
break;
}
i += 1;
}
tokens.push("<literal>");
continue;
}
if javascript
&& bytes[i] == b'/'
&& tokens
.last()
.is_none_or(|t| matches!(*t, "=" | "(" | "," | ":" | "return"))
{
i += 1;
let mut class = false;
while i < bytes.len() {
match bytes[i] {
b'\\' => {
i = (i + 2).min(bytes.len());
continue;
}
b'[' => class = true,
b']' => class = false,
b'/' if !class => {
i += 1;
break;
}
b'\n' => break,
_ => {}
}
i += 1;
}
tokens.push("<literal>");
continue;
}
let start = i;
if bytes[i].is_ascii_alphanumeric() || bytes[i] == b'_' || bytes[i] == b'$' {
i += 1;
while i < bytes.len()
&& (bytes[i].is_ascii_alphanumeric() || matches!(bytes[i], b'_' | b'$'))
{
i += 1;
}
} else {
i += code[i..].chars().next().unwrap().len_utf8();
}
tokens.push(&code[start..i]);
}
for (index, token) in tokens.iter().enumerate() {
if *token != "handle" {
continue;
}
let before = index.checked_sub(1).and_then(|n| tokens.get(n)).copied();
let after = tokens.get(index + 1).copied();
if python && before == Some("def") && after == Some("(") {
return true;
}
if perl && before == Some("sub") && matches!(after, Some("{" | "(" | ":")) {
return true;
}
if language == "go" && before == Some("func") && after == Some("(") {
return true;
}
if javascript {
if before == Some("function") && after == Some("(") {
return true;
}
if after != Some("=") {
continue;
}
let mut next = index + 2;
if tokens.get(next) == Some(&"async") {
next += 1;
}
if tokens.get(next) == Some(&"function") {
return true;
}
if tokens.get(next) == Some(&"(") {
let mut depth = 0;
while let Some(token) = tokens.get(next) {
match *token {
"(" => depth += 1,
")" => depth -= 1,
_ => {}
}
next += 1;
if depth == 0 {
break;
}
}
} else {
next += 1;
} if tokens.get(next..next.saturating_add(2)) == Some(&["=", ">"][..]) {
return true;
}
}
}
false
}
pub fn validate_event_references(value: &Value) -> Result<(), InterpolationError> {
check_tree(value)?;
fn check_string(s: &str) -> Result<(), InterpolationError> {
let mut cursor = 0;
while let Some(found) = find_reference(s, cursor) {
parse_path(&found)?;
cursor = found.end;
}
Ok(())
}
let mut pending = vec![value];
while let Some(value) = pending.pop() {
match value {
Value::String(s) => check_string(s)?,
Value::Array(items) => pending.extend(items),
Value::Object(map) => {
for key in map.keys() {
check_string(key)?;
}
pending.extend(map.values());
}
_ => {}
}
}
Ok(())
}