use crate::operation::{Operation, OperationType};
use crate::state_trigger::{StateTrigger, StateTriggerChangeKind, StateTriggerPredicate};
use serde::de::Deserializer;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
pub const TRIGGER_MATCH_CONTRACT_ID: &str = "feltdb.trigger.match";
pub const TRIGGER_MATCH_CONTRACT_VERSION: u32 = 1;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContractChangeKind {
Insert,
Update,
Delete,
}
impl ContractChangeKind {
pub fn from_operation_type(value: &OperationType) -> Self {
match value {
OperationType::Insert => Self::Insert,
OperationType::Update => Self::Update,
OperationType::Delete => Self::Delete,
}
}
pub fn from_trigger_change_kind(value: StateTriggerChangeKind) -> Self {
match value {
StateTriggerChangeKind::Insert => Self::Insert,
StateTriggerChangeKind::Update => Self::Update,
StateTriggerChangeKind::Delete => Self::Delete,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContractChange {
pub change_kind: ContractChangeKind,
pub collection: String,
#[serde(
default,
deserialize_with = "deserialize_present",
skip_serializing_if = "Option::is_none"
)]
pub value: Option<Value>,
}
impl ContractChange {
pub fn from_operation(operation: &Operation) -> Self {
Self {
change_kind: ContractChangeKind::from_operation_type(&operation.op_type),
collection: operation.capability.clone(),
value: operation.value.clone(),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContractTrigger {
pub change_kind: ContractChangeKind,
pub enabled: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub predicate: Option<ContractPredicate>,
pub source_collection: String,
}
impl ContractTrigger {
pub fn from_state_trigger(trigger: &StateTrigger) -> Self {
Self {
change_kind: ContractChangeKind::from_trigger_change_kind(trigger.change_kind),
enabled: trigger.enabled,
predicate: trigger.predicate.as_ref().map(ContractPredicate::from),
source_collection: trigger.source_collection.clone(),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContractPredicate {
pub equals: Value,
pub field: String,
}
impl From<&StateTriggerPredicate> for ContractPredicate {
fn from(value: &StateTriggerPredicate) -> Self {
Self {
equals: value.equals.clone(),
field: value.field.clone(),
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TriggerMatchInput {
pub change: ContractChange,
pub contract: String,
pub trigger: ContractTrigger,
pub version: u32,
}
impl TriggerMatchInput {
pub fn new(change: ContractChange, trigger: ContractTrigger) -> Self {
Self {
change,
contract: TRIGGER_MATCH_CONTRACT_ID.to_string(),
trigger,
version: TRIGGER_MATCH_CONTRACT_VERSION,
}
}
pub fn for_operation(trigger: &StateTrigger, operation: &Operation) -> Self {
Self::new(
ContractChange::from_operation(operation),
ContractTrigger::from_state_trigger(trigger),
)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MatchReason {
Matched,
TriggerDisabled,
CollectionMismatch,
ChangeKindMismatch,
ChangeValueAbsent,
PredicateFieldAbsent,
PredicateValueMismatch,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MatchOutcome {
pub matched: bool,
pub reason: MatchReason,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum ContractErrorCode {
ContractInputInvalid,
ContractIdMismatch,
ContractVersionUnsupported,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ContractErrorBody {
pub code: ContractErrorCode,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TriggerMatchOutput {
pub contract: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<ContractErrorBody>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub outcome: Option<MatchOutcome>,
pub version: u32,
}
impl TriggerMatchOutput {
fn decision(matched: bool, reason: MatchReason) -> Self {
Self {
contract: TRIGGER_MATCH_CONTRACT_ID.to_string(),
error: None,
outcome: Some(MatchOutcome { matched, reason }),
version: TRIGGER_MATCH_CONTRACT_VERSION,
}
}
fn failure(code: ContractErrorCode) -> Self {
Self {
contract: TRIGGER_MATCH_CONTRACT_ID.to_string(),
error: Some(ContractErrorBody { code }),
outcome: None,
version: TRIGGER_MATCH_CONTRACT_VERSION,
}
}
pub fn matched(&self) -> bool {
self.outcome.is_some_and(|outcome| outcome.matched)
}
}
pub fn evaluate_trigger(input: &TriggerMatchInput) -> TriggerMatchOutput {
let change = &input.change;
let trigger = &input.trigger;
if !trigger.enabled {
return TriggerMatchOutput::decision(false, MatchReason::TriggerDisabled);
}
if change.collection != trigger.source_collection {
return TriggerMatchOutput::decision(false, MatchReason::CollectionMismatch);
}
if change.change_kind != trigger.change_kind {
return TriggerMatchOutput::decision(false, MatchReason::ChangeKindMismatch);
}
let Some(predicate) = trigger.predicate.as_ref() else {
return TriggerMatchOutput::decision(true, MatchReason::Matched);
};
let Some(value) = change.value.as_ref() else {
return TriggerMatchOutput::decision(false, MatchReason::ChangeValueAbsent);
};
let Some(resolved) = resolve_field(value, &predicate.field) else {
return TriggerMatchOutput::decision(false, MatchReason::PredicateFieldAbsent);
};
if resolved != &predicate.equals {
return TriggerMatchOutput::decision(false, MatchReason::PredicateValueMismatch);
}
TriggerMatchOutput::decision(true, MatchReason::Matched)
}
fn resolve_field<'a>(value: &'a Value, path: &str) -> Option<&'a Value> {
if path.is_empty() {
return Some(value);
}
let mut current = value;
for segment in path.split('.') {
current = current.as_object()?.get(segment)?;
}
Some(current)
}
pub fn canonical_input_bytes(input: &TriggerMatchInput) -> Vec<u8> {
serde_json::to_vec(input).expect("contract input is always serializable")
}
pub fn canonical_output_bytes(output: &TriggerMatchOutput) -> Vec<u8> {
serde_json::to_vec(output).expect("contract output is always serializable")
}
pub fn input_digest(input: &TriggerMatchInput) -> String {
format!("sha256:{:x}", Sha256::digest(canonical_input_bytes(input)))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ContractRuntime {
Native,
Wasm,
}
impl ContractRuntime {
pub fn as_str(self) -> &'static str {
match self {
Self::Native => "native",
Self::Wasm => "wasm",
}
}
}
pub fn evaluate_canonical(input: &[u8]) -> Vec<u8> {
canonical_output_bytes(&evaluate_canonical_output(input))
}
pub fn evaluate_canonical_output(input: &[u8]) -> TriggerMatchOutput {
let Ok(parsed) = serde_json::from_slice::<TriggerMatchInput>(input) else {
return TriggerMatchOutput::failure(ContractErrorCode::ContractInputInvalid);
};
if parsed.contract != TRIGGER_MATCH_CONTRACT_ID {
return TriggerMatchOutput::failure(ContractErrorCode::ContractIdMismatch);
}
if parsed.version != TRIGGER_MATCH_CONTRACT_VERSION {
return TriggerMatchOutput::failure(ContractErrorCode::ContractVersionUnsupported);
}
evaluate_trigger(&parsed)
}
fn deserialize_present<'de, D>(deserializer: D) -> Result<Option<Value>, D::Error>
where
D: Deserializer<'de>,
{
Value::deserialize(deserializer).map(Some)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::operation::Operation;
use serde_json::json;
fn change(kind: ContractChangeKind, collection: &str, value: Option<Value>) -> ContractChange {
ContractChange {
change_kind: kind,
collection: collection.into(),
value,
}
}
fn trigger(predicate: Option<ContractPredicate>) -> ContractTrigger {
ContractTrigger {
change_kind: ContractChangeKind::Insert,
enabled: true,
predicate,
source_collection: "orders".into(),
}
}
fn decide(change: ContractChange, trigger: ContractTrigger) -> (bool, MatchReason) {
let output = evaluate_trigger(&TriggerMatchInput::new(change, trigger));
let outcome = output.outcome.expect("a typed input always decides");
(outcome.matched, outcome.reason)
}
#[test]
fn checks_run_in_a_fixed_order() {
let mut disabled = trigger(Some(ContractPredicate {
equals: json!("paid"),
field: "status".into(),
}));
disabled.enabled = false;
assert_eq!(
decide(
change(ContractChangeKind::Delete, "customers", None),
disabled
),
(false, MatchReason::TriggerDisabled)
);
assert_eq!(
decide(
change(ContractChangeKind::Delete, "customers", None),
trigger(None)
),
(false, MatchReason::CollectionMismatch)
);
assert_eq!(
decide(
change(ContractChangeKind::Delete, "orders", None),
trigger(None)
),
(false, MatchReason::ChangeKindMismatch)
);
}
#[test]
fn absent_and_null_record_values_are_different_inputs() {
let whole_record = || {
Some(ContractPredicate {
equals: Value::Null,
field: String::new(),
})
};
assert_eq!(
decide(
change(ContractChangeKind::Insert, "orders", Some(Value::Null)),
trigger(whole_record())
),
(true, MatchReason::Matched)
);
assert_eq!(
decide(
change(ContractChangeKind::Insert, "orders", None),
trigger(whole_record())
),
(false, MatchReason::ChangeValueAbsent)
);
}
#[test]
fn a_missing_field_is_not_a_null_field() {
let expects_null = || {
Some(ContractPredicate {
equals: Value::Null,
field: "cancelled_at".into(),
})
};
assert_eq!(
decide(
change(
ContractChangeKind::Insert,
"orders",
Some(json!({"cancelled_at": null}))
),
trigger(expects_null())
),
(true, MatchReason::Matched)
);
assert_eq!(
decide(
change(ContractChangeKind::Insert, "orders", Some(json!({}))),
trigger(expects_null())
),
(false, MatchReason::PredicateFieldAbsent)
);
}
#[test]
fn field_paths_descend_through_objects_only() {
let value = json!({"customer": {"country": "NL"}, "tags": ["a"]});
assert_eq!(
resolve_field(&value, "customer.country"),
Some(&json!("NL"))
);
assert_eq!(resolve_field(&value, ""), Some(&value));
assert_eq!(resolve_field(&value, "tags.0"), None);
assert_eq!(resolve_field(&value, "customer.missing"), None);
}
#[test]
fn canonical_encoding_sorts_members() {
let bytes = canonical_input_bytes(&TriggerMatchInput::new(
change(
ContractChangeKind::Insert,
"orders",
Some(json!({"zeta": 1, "alpha": 2})),
),
trigger(None),
));
assert_eq!(
String::from_utf8(bytes).unwrap(),
r#"{"change":{"change_kind":"insert","collection":"orders","value":{"alpha":2,"zeta":1}},"contract":"feltdb.trigger.match","trigger":{"change_kind":"insert","enabled":true,"source_collection":"orders"},"version":1}"#
);
}
#[test]
fn canonical_encoding_omits_an_absent_value_and_keeps_a_null_one() {
let with_null = canonical_input_bytes(&TriggerMatchInput::new(
change(ContractChangeKind::Insert, "orders", Some(Value::Null)),
trigger(None),
));
assert!(String::from_utf8_lossy(&with_null).contains(r#""value":null"#));
let without = canonical_input_bytes(&TriggerMatchInput::new(
change(ContractChangeKind::Delete, "orders", None),
trigger(None),
));
assert!(!String::from_utf8_lossy(&without).contains("value"));
}
#[test]
fn canonical_round_trip_is_stable() {
let input = TriggerMatchInput::new(
change(
ContractChangeKind::Insert,
"orders",
Some(json!({"status": "paid", "total": 42})),
),
trigger(Some(ContractPredicate {
equals: json!("paid"),
field: "status".into(),
})),
);
let once = canonical_input_bytes(&input);
let parsed: TriggerMatchInput = serde_json::from_slice(&once).unwrap();
assert_eq!(parsed, input);
assert_eq!(canonical_input_bytes(&parsed), once);
assert_eq!(input_digest(&parsed), input_digest(&input));
}
#[test]
fn unevaluatable_inputs_are_decisions_rather_than_failures() {
for (bytes, code) in [
(
&b"not json"[..],
ContractErrorCode::ContractInputInvalid,
),
(
br#"{"change":{"change_kind":"insert","collection":"o"},"contract":"other","trigger":{"change_kind":"insert","enabled":true,"source_collection":"o"},"version":1}"#,
ContractErrorCode::ContractIdMismatch,
),
(
br#"{"change":{"change_kind":"insert","collection":"o"},"contract":"feltdb.trigger.match","trigger":{"change_kind":"insert","enabled":true,"source_collection":"o"},"version":9}"#,
ContractErrorCode::ContractVersionUnsupported,
),
] {
let output = evaluate_canonical_output(bytes);
assert_eq!(output.error.map(|error| error.code), Some(code));
assert!(!output.matched(), "an unevaluatable input never matches");
}
}
#[test]
fn the_contract_reads_only_what_the_decision_depends_on() {
let one = Operation::insert(
1,
"node-a".into(),
1,
"orders:1".into(),
json!({"status": "paid"}),
"Order".into(),
"orders".into(),
);
let mut two = Operation::insert(
999,
"node-b".into(),
77,
"orders:2".into(),
json!({"status": "paid"}),
"OtherOrder".into(),
"orders".into(),
);
two.timestamp_ms = one.timestamp_ms + 1_000_000;
two.content_hash = "different".into();
assert_eq!(
ContractChange::from_operation(&one),
ContractChange::from_operation(&two)
);
}
#[test]
fn the_native_contract_decides_production_matching() {
use crate::authorization::{
Grant, GrantConstraints, GrantSigner, GrantStatus, ResourceScope, ResourceScopeKind,
Subject,
};
use crate::state_trigger::{StateTrigger, StateTriggerChangeKind, StateTriggerPredicate};
use std::collections::BTreeSet;
let subject = Subject::Agent("a".into());
let signed = GrantSigner::new("k", vec![7; 32])
.unwrap()
.sign(Grant {
contract_version: 1,
id: "g".into(),
issuer: Subject::Human("o".into()),
subject: subject.clone(),
tenant_id: "t".into(),
application_id: "app".into(),
capabilities: ["workload:execute".into()].into(),
resources: vec![ResourceScope {
kind: ResourceScopeKind::Workload,
uri: "flow://app/workloads/*".into(),
}],
constraints: GrantConstraints::default(),
issued_at: 1,
expires_at: 9999,
parent_grant: None,
revision: 1,
status: GrantStatus::Active,
})
.unwrap();
let state_trigger = StateTrigger {
trigger_id: "t1".into(),
source_collection: "orders".into(),
change_kind: StateTriggerChangeKind::Insert,
predicate: Some(StateTriggerPredicate {
field: "status".into(),
equals: json!("paid"),
}),
target_capability: "identity".into(),
tenant_id: "t".into(),
application_id: "app".into(),
environment: "production".into(),
definition_revision: "r".into(),
requested_by: subject,
authorization_snapshot: signed,
capability_snapshot: BTreeSet::new(),
connection_scopes: BTreeSet::new(),
priority: 0,
retry_policy: Default::default(),
timeouts: Default::default(),
not_before: None,
enabled: true,
};
for (value, expected) in [
(json!({"status": "paid"}), true),
(json!({"status": "draft"}), false),
] {
let operation = Operation::insert(
1,
"node".into(),
1,
"orders:1".into(),
value,
"Order".into(),
"orders".into(),
);
assert_eq!(state_trigger.matches(&operation), expected);
assert_eq!(
evaluate_trigger(&state_trigger.contract_input(&operation)).matched(),
expected
);
let bytes = canonical_input_bytes(&state_trigger.contract_input(&operation));
assert_eq!(evaluate_canonical_output(&bytes).matched(), expected);
}
}
}