use crate::access_control::{
AccessLevel, CommandToken, DirectCreationToken, EventReadToken, ExternalEventToken,
RoleAccessMapping, RoleStatus, TokenStatus,
};
use crate::encryption::DataKey;
use crate::error::SkiljRejection;
use crate::shared::{
CommandDecision, Filter, FilterOperator, Metadata, SensitiveField, Tag, TagMapping,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BoundedContextStatus {
Active,
Archived,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BoundedContext {
pub name: String,
pub status: BoundedContextStatus,
pub created_at: chrono::DateTime<chrono::Utc>,
pub created_by: crate::bootstrap::ContextCreator,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MissedOccurrencePolicy {
Skip,
FireOnce,
ReplayBacklog,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventType {
pub bounded_context: BoundedContext,
pub name: String,
pub schema: String,
pub schema_version: i64,
pub tag_mappings: Vec<TagMapping>,
pub sensitive_fields: Vec<SensitiveField>,
pub external_creation_allowed: bool,
pub direct_creation_allowed: bool,
pub system_triggered_allowed: bool,
pub system_triggered_schedule: Option<String>,
pub missed_occurrence_policy: Option<MissedOccurrencePolicy>,
pub schedule_position: Option<chrono::DateTime<chrono::Utc>>,
pub last_fired_at: Option<chrono::DateTime<chrono::Utc>>,
pub event_read_allowed: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandType {
pub bounded_context: BoundedContext,
pub name: String,
pub schema: String,
pub schema_version: i64,
pub tag_mappings: Vec<TagMapping>,
pub sensitive_fields: Vec<SensitiveField>,
pub rest_trigger_allowed: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EncryptionKeyStatus {
Active,
Destroyed,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EncryptionKey {
pub bounded_context: BoundedContext,
pub subject_key: String,
pub subject_value: String,
pub status: EncryptionKeyStatus,
pub created_at: chrono::DateTime<chrono::Utc>,
pub destroyed_at: Option<chrono::DateTime<chrono::Utc>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProtectedPayload {
pub payload: String,
pub encryption_keys: Vec<EncryptionKey>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Command {
pub id: String,
pub bounded_context: BoundedContext,
pub command_type: CommandType,
pub payload: String,
pub metadata: Metadata,
pub encryption_keys: Vec<EncryptionKey>,
pub consistency_tags: Vec<Tag>,
pub consistency_boundary: Option<i64>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EventOrigin {
CommandTriggered {
command: Box<Command>,
},
ExternalTriggered {
source_content: String,
source_context: Option<String>,
},
DirectlyCreated,
SystemTriggered,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Event {
pub bounded_context: BoundedContext,
pub event_type: EventType,
pub payload: String,
pub metadata: Metadata,
pub sequence: i64,
pub tags: Vec<Tag>,
pub encryption_keys: Vec<EncryptionKey>,
pub origin: EventOrigin,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AllEventsSubscription {
pub bounded_context: BoundedContext,
pub access_mapping: RoleAccessMapping,
pub created_at: chrono::DateTime<chrono::Utc>,
pub from_sequence: i64,
pub event_types: Vec<EventType>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventTypeSubscription {
pub bounded_context: BoundedContext,
pub access_mapping: RoleAccessMapping,
pub created_at: chrono::DateTime<chrono::Utc>,
pub from_sequence: i64,
pub event_type: EventType,
pub filters: Vec<Filter>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Subscription {
AllEventsSubscription(Box<AllEventsSubscription>),
EventTypeSubscription(Box<EventTypeSubscription>),
}
impl Subscription {
fn bounded_context(&self) -> &BoundedContext {
match self {
Subscription::AllEventsSubscription(s) => &s.bounded_context,
Subscription::EventTypeSubscription(s) => &s.bounded_context,
}
}
fn access_mapping(&self) -> &RoleAccessMapping {
match self {
Subscription::AllEventsSubscription(s) => &s.access_mapping,
Subscription::EventTypeSubscription(s) => &s.access_mapping,
}
}
fn starting_sequence(&self) -> i64 {
match self {
Subscription::AllEventsSubscription(s) => s.from_sequence,
Subscription::EventTypeSubscription(s) => s.from_sequence,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AckMode {
AutoAdvance,
ManualAck,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReadCursor {
pub token: EventReadToken,
pub ack_mode: AckMode,
pub sequence: i64,
pub updated_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum Error {
#[error("this bounded context is archived and accepts no new writes")]
BoundedContextArchived,
#[error("this schema change is not backwards compatible")]
SchemaIncompatible,
#[error(
"a scheduled EventType must declare both system_triggered_schedule and \
missed_occurrence_policy - neither has a default"
)]
MissingScheduleOrPolicy,
#[error("this filter is invalid for its field's declared type")]
InvalidFilter,
#[error("this schema is not well-formed JSON Schema")]
InvalidSchema,
#[error("this tag mapping names a field the schema doesn't declare")]
InvalidTagMapping,
#[error("a TagMapping and a SensitiveField may not name the same field")]
SensitiveFieldTagOverlap,
#[error("this sensitive field names a field the schema doesn't declare")]
InvalidSensitiveField,
#[error("a re-registration may not drop a tag mapping key already in use")]
TagMappingKeyDropped,
#[error("this EncryptionKey is not active")]
EncryptionKeyNotActive,
#[error("this named EventType belongs to a different bounded context than the grant's")]
EventTypeNotInBoundedContext,
#[error("this named CommandType belongs to a different bounded context than the grant's")]
CommandTypeNotInBoundedContext,
#[error("this triggered_event belongs to a different bounded context than the grant's")]
TriggeredEventNotInBoundedContext,
#[error("this EventReadToken's cursor mode doesn't match what was requested")]
CursorAckModeMismatch,
#[error("acknowledging requires a manual_ack cursor")]
NotManualAckCursor,
#[error("this EventType is not opted into event reads")]
EventReadNotAllowed,
#[error("this EventReadToken has no read cursor yet - consume before acknowledging")]
NoReadCursor,
#[error("an acknowledgement may not move a cursor backwards")]
AcknowledgementRegresses,
#[error("this EventType is not opted into external event ingestion")]
ExternalCreationNotAllowed,
#[error("this EventType is not opted into direct event creation")]
DirectCreationNotAllowed,
#[error("this CommandType is not opted into REST triggering")]
RestTriggerNotAllowed,
#[error("this payload does not validate against its type's registered schema")]
PayloadDoesNotMatchSchema,
#[error("decide() named an event type this bounded context has not registered: {0}")]
UnregisteredEventType(String),
#[error("stored payload did not decode into its expected type: {0}")]
PayloadDecodeFailed(String),
}
impl SkiljRejection for Error {
fn code(&self) -> &str {
match self {
Error::BoundedContextArchived => "bounded_context_archived",
Error::SchemaIncompatible => "schema_incompatible",
Error::MissingScheduleOrPolicy => "missing_schedule_or_policy",
Error::InvalidFilter => "invalid_filter",
Error::InvalidSchema => "invalid_schema",
Error::InvalidTagMapping => "invalid_tag_mapping",
Error::SensitiveFieldTagOverlap => "sensitive_field_tag_overlap",
Error::InvalidSensitiveField => "invalid_sensitive_field",
Error::TagMappingKeyDropped => "tag_mapping_key_dropped",
Error::EventTypeNotInBoundedContext => "event_type_not_in_bounded_context",
Error::CommandTypeNotInBoundedContext => "command_type_not_in_bounded_context",
Error::TriggeredEventNotInBoundedContext => "triggered_event_not_in_bounded_context",
Error::EncryptionKeyNotActive => "encryption_key_not_active",
Error::CursorAckModeMismatch => "cursor_ack_mode_mismatch",
Error::NotManualAckCursor => "not_manual_ack_cursor",
Error::EventReadNotAllowed => "event_read_not_allowed",
Error::NoReadCursor => "no_read_cursor",
Error::AcknowledgementRegresses => "acknowledgement_regresses",
Error::ExternalCreationNotAllowed => "external_creation_not_allowed",
Error::DirectCreationNotAllowed => "direct_creation_not_allowed",
Error::RestTriggerNotAllowed => "rest_trigger_not_allowed",
Error::PayloadDoesNotMatchSchema => "payload_does_not_match_schema",
Error::UnregisteredEventType(_) => "unregistered_event_type",
Error::PayloadDecodeFailed(_) => "payload_decode_failed",
}
}
fn message(&self) -> String {
self.to_string()
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum FieldKind {
Scalar {
json_type: &'static str,
format: Option<String>,
},
ListOfScalar,
Other,
}
fn json_type_str(type_value: &serde_json::Value) -> Option<&str> {
match type_value {
serde_json::Value::String(s) => Some(s.as_str()),
serde_json::Value::Array(arr) => {
arr.iter().find_map(|v| v.as_str().filter(|s| *s != "null"))
}
_ => None,
}
}
fn classify(
schema: &serde_json::Value,
definitions: Option<&serde_json::Map<String, serde_json::Value>>,
) -> FieldKind {
if let Some(def_name) = schema
.get("$ref")
.and_then(|v| v.as_str())
.and_then(|r| r.rsplit('/').next())
{
return definitions
.and_then(|defs| defs.get(def_name))
.map(|resolved| classify(resolved, definitions))
.unwrap_or(FieldKind::Other);
}
match schema.get("type").and_then(json_type_str) {
Some(t @ ("string" | "integer" | "number" | "boolean")) => FieldKind::Scalar {
json_type: match t {
"string" => "string",
"integer" => "integer",
"number" => "number",
_ => "boolean",
},
format: schema
.get("format")
.and_then(|f| f.as_str())
.map(String::from),
},
Some("array") => match schema
.get("items")
.and_then(|i| i.get("type"))
.and_then(json_type_str)
{
Some("string" | "integer" | "number" | "boolean") => FieldKind::ListOfScalar,
_ => FieldKind::Other, },
_ => FieldKind::Other, }
}
fn resolve_field_kind(
properties: &serde_json::Map<String, serde_json::Value>,
definitions: Option<&serde_json::Map<String, serde_json::Value>>,
field: &str,
) -> Option<FieldKind> {
match classify(resolve_field(properties, definitions, field)?, definitions) {
FieldKind::Other => None,
kind => Some(kind),
}
}
fn filter_operator_is_valid(kind: &FieldKind, operator: FilterOperator) -> bool {
match kind {
FieldKind::Scalar {
json_type: "string",
format,
} => {
matches!(
operator,
FilterOperator::Equals | FilterOperator::Contains | FilterOperator::IsLike
) || (matches!(
format.as_deref(),
Some("date-time" | "date" | "partial-date-time")
) && matches!(
operator,
FilterOperator::GreaterThan | FilterOperator::LessThan
))
}
FieldKind::Scalar {
json_type: "integer" | "number",
..
} => matches!(
operator,
FilterOperator::Equals | FilterOperator::GreaterThan | FilterOperator::LessThan
),
FieldKind::Scalar {
json_type: "boolean",
..
} => operator == FilterOperator::Equals,
FieldKind::Scalar { .. } => false, FieldKind::ListOfScalar => operator == FilterOperator::Contains,
FieldKind::Other => false,
}
}
pub fn valid_filters(event_type: &EventType, filters: &[Filter]) -> bool {
if filters.is_empty() {
return true;
}
let Some(properties) = schema_properties(&event_type.schema) else {
return false;
};
let definitions = schema_definitions(&event_type.schema);
filters.iter().all(|f| {
resolve_field_kind(&properties, definitions.as_ref(), &f.field)
.is_some_and(|kind| filter_operator_is_valid(&kind, f.operator))
})
}
fn schema_properties(schema: &str) -> Option<serde_json::Map<String, serde_json::Value>> {
let parsed: serde_json::Value = serde_json::from_str(schema).ok()?;
parsed.get("properties")?.as_object().cloned()
}
fn schema_definitions(schema: &str) -> Option<serde_json::Map<String, serde_json::Value>> {
let parsed: serde_json::Value = serde_json::from_str(schema).ok()?;
parsed.get("definitions")?.as_object().cloned()
}
fn schema_required(schema: &str) -> Vec<String> {
serde_json::from_str::<serde_json::Value>(schema)
.ok()
.and_then(|v| v.get("required").cloned())
.and_then(|v| v.as_array().cloned())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(String::from))
.collect()
})
.unwrap_or_default()
}
fn resolve_field<'a>(
properties: &'a serde_json::Map<String, serde_json::Value>,
definitions: Option<&'a serde_json::Map<String, serde_json::Value>>,
field: &str,
) -> Option<&'a serde_json::Value> {
match field.split_once('.') {
None => properties.get(field),
Some((outer, inner)) => {
let def_name = properties
.get(outer)?
.get("$ref")?
.as_str()?
.rsplit('/')
.next()?;
definitions?.get(def_name)?.get("properties")?.get(inner)
}
}
}
fn payload_field_value<'a>(
payload: &'a serde_json::Value,
field: &str,
) -> Option<&'a serde_json::Value> {
match field.split_once('.') {
None => payload.get(field),
Some((outer, inner)) => payload.get(outer)?.get(inner),
}
}
fn payload_field_value_mut<'a>(
payload: &'a mut serde_json::Value,
field: &str,
) -> Option<&'a mut serde_json::Value> {
match field.split_once('.') {
None => payload.get_mut(field),
Some((outer, inner)) => payload.get_mut(outer)?.get_mut(inner),
}
}
fn json_scalar_to_string(value: &serde_json::Value) -> Option<String> {
match value {
serde_json::Value::String(s) => Some(s.clone()),
serde_json::Value::Number(_) | serde_json::Value::Bool(_) => Some(value.to_string()),
_ => None,
}
}
pub fn valid_schema(schema: &str) -> bool {
let Ok(schema_value) = serde_json::from_str::<serde_json::Value>(schema) else {
return false;
};
jsonschema::validator_for(&schema_value).is_ok()
}
pub fn valid_tag_mappings(schema: &str, tag_mappings: &[TagMapping]) -> bool {
if tag_mappings.is_empty() {
return true;
}
let Some(properties) = schema_properties(schema) else {
return false;
};
let definitions = schema_definitions(schema);
tag_mappings
.iter()
.all(|m| resolve_field_kind(&properties, definitions.as_ref(), &m.field).is_some())
}
pub fn valid_sensitive_fields(schema: &str, sensitive_fields: &[SensitiveField]) -> bool {
if sensitive_fields.is_empty() {
return true;
}
let Some(properties) = schema_properties(schema) else {
return false;
};
let definitions = schema_definitions(schema);
sensitive_fields.iter().all(|s| {
matches!(
resolve_field_kind(&properties, definitions.as_ref(), &s.field),
Some(FieldKind::Scalar {
json_type: "string",
..
})
) && resolve_field_kind(&properties, definitions.as_ref(), &s.subject_field).is_some()
})
}
pub fn valid_payload(schema: &str, payload: &str) -> bool {
let Ok(schema_value) = serde_json::from_str::<serde_json::Value>(schema) else {
return false;
};
let Ok(payload_value) = serde_json::from_str::<serde_json::Value>(payload) else {
return false;
};
let Ok(validator) = jsonschema::validator_for(&schema_value) else {
return false;
};
validator.is_valid(&payload_value)
}
pub fn schema_is_backwards_compatible(existing_schema: &str, schema: &str) -> bool {
if existing_schema == schema {
return true;
}
let (Some(existing_properties), Some(new_properties)) = (
schema_properties(existing_schema),
schema_properties(schema),
) else {
return false;
};
let existing_required = schema_required(existing_schema);
let new_required = schema_required(schema);
for (field, existing_def) in &existing_properties {
let Some(new_def) = new_properties.get(field) else {
return false; };
if existing_def.get("type") != new_def.get("type") {
return false; }
let was_required = existing_required.iter().any(|f| f == field);
let now_required = new_required.iter().any(|f| f == field);
if !was_required && now_required {
return false; }
}
for field in new_properties.keys() {
if !existing_properties.contains_key(field) && new_required.iter().any(|f| f == field) {
return false; }
}
true
}
fn like_matches(text: &str, pattern: &str) -> bool {
let text: Vec<char> = text.chars().collect();
let pattern: Vec<char> = pattern.chars().collect();
let mut dp = vec![vec![false; pattern.len() + 1]; text.len() + 1];
dp[0][0] = true;
for j in 1..=pattern.len() {
if pattern[j - 1] == '%' {
dp[0][j] = dp[0][j - 1];
}
}
for i in 1..=text.len() {
for j in 1..=pattern.len() {
dp[i][j] = match pattern[j - 1] {
'%' => dp[i - 1][j] || dp[i][j - 1],
'_' => dp[i - 1][j - 1],
c => dp[i - 1][j - 1] && text[i - 1] == c,
};
}
}
dp[text.len()][pattern.len()]
}
fn string_ordering(payload: &str, filter_value: &str, operator: FilterOperator) -> bool {
fn cmp<T: PartialOrd>(a: T, b: T, operator: FilterOperator) -> bool {
match operator {
FilterOperator::GreaterThan => a > b,
FilterOperator::LessThan => a < b,
_ => false,
}
}
if let (Ok(a), Ok(b)) = (
chrono::DateTime::parse_from_rfc3339(payload),
chrono::DateTime::parse_from_rfc3339(filter_value),
) {
return cmp(a, b, operator);
}
if let (Ok(a), Ok(b)) = (
chrono::NaiveDate::parse_from_str(payload, "%Y-%m-%d"),
chrono::NaiveDate::parse_from_str(filter_value, "%Y-%m-%d"),
) {
return cmp(a, b, operator);
}
if let (Ok(a), Ok(b)) = (
chrono::NaiveTime::parse_from_str(payload, "%H:%M:%S%.f"),
chrono::NaiveTime::parse_from_str(filter_value, "%H:%M:%S%.f"),
) {
return cmp(a, b, operator);
}
false
}
fn matches_one_filter(payload: &serde_json::Value, filter: &Filter) -> bool {
let Some(value) = payload_field_value(payload, &filter.field) else {
return false;
};
match value {
serde_json::Value::Array(items) => {
filter.operator == FilterOperator::Contains
&& items.iter().any(|item| {
json_scalar_to_string(item).as_deref() == Some(filter.value.as_str())
})
}
serde_json::Value::String(s) => match filter.operator {
FilterOperator::Equals => s == &filter.value,
FilterOperator::Contains => s.contains(filter.value.as_str()),
FilterOperator::IsLike => like_matches(s, &filter.value),
FilterOperator::GreaterThan | FilterOperator::LessThan => {
string_ordering(s, &filter.value, filter.operator)
}
},
serde_json::Value::Number(n) => {
let (Some(a), Ok(b)) = (n.as_f64(), filter.value.parse::<f64>()) else {
return false;
};
match filter.operator {
FilterOperator::Equals => a == b,
FilterOperator::GreaterThan => a > b,
FilterOperator::LessThan => a < b,
FilterOperator::Contains | FilterOperator::IsLike => false,
}
}
serde_json::Value::Bool(b) => {
filter.operator == FilterOperator::Equals && b.to_string() == filter.value
}
serde_json::Value::Null | serde_json::Value::Object(_) => false,
}
}
pub fn matches_filters(event: &Event, filters: &[Filter]) -> bool {
if filters.is_empty() {
return true;
}
let Ok(payload) = serde_json::from_str::<serde_json::Value>(&event.payload) else {
return false;
};
filters.iter().all(|f| matches_one_filter(&payload, f))
}
fn highest_sequence(events: &[Event]) -> Option<i64> {
events.iter().map(|e| e.sequence).max()
}
pub fn protect_sensitive_fields(
sensitive_fields: &[SensitiveField],
payload: &str,
resolve_key: impl Fn(&str, &str) -> (EncryptionKey, DataKey),
) -> ProtectedPayload {
if sensitive_fields.is_empty() {
return ProtectedPayload {
payload: payload.to_string(),
encryption_keys: Vec::new(),
};
}
let mut parsed: serde_json::Value = serde_json::from_str(payload).expect(
"protect_sensitive_fields: payload is always valid JSON by the time this is called",
);
let mut used_keys: Vec<EncryptionKey> = Vec::new();
for sf in sensitive_fields {
let Some(subject_value) =
payload_field_value(&parsed, &sf.subject_field).and_then(json_scalar_to_string)
else {
continue;
};
let Some(plaintext) = payload_field_value(&parsed, &sf.field).map(|v| {
v.as_str()
.map(str::to_string)
.unwrap_or_else(|| v.to_string())
}) else {
continue;
};
let (key, data_key) = resolve_key(&sf.subject_key, &subject_value);
let ciphertext = crate::encryption::encrypt_leaf(&data_key, &plaintext);
if let Some(slot) = payload_field_value_mut(&mut parsed, &sf.field) {
*slot = serde_json::Value::String(ciphertext);
}
if !used_keys.contains(&key) {
used_keys.push(key);
}
}
ProtectedPayload {
payload: serde_json::to_string(&parsed)
.expect("re-serialising a parsed JSON Value is infallible"),
encryption_keys: used_keys,
}
}
pub fn sensitive_field_subjects(
sensitive_fields: &[SensitiveField],
payload: &str,
) -> Vec<(String, String)> {
if sensitive_fields.is_empty() {
return Vec::new();
}
let parsed: serde_json::Value = serde_json::from_str(payload).expect(
"sensitive_field_subjects: payload is always valid JSON by the time this is called",
);
let mut subjects = Vec::new();
for sf in sensitive_fields {
if let Some(subject_value) =
payload_field_value(&parsed, &sf.subject_field).and_then(json_scalar_to_string)
{
let pair = (sf.subject_key.clone(), subject_value);
if !subjects.contains(&pair) {
subjects.push(pair);
}
}
}
subjects
}
fn push_unique_tag(tags: &mut Vec<Tag>, key: &str, value: Option<String>) {
let tag = Tag {
key: key.to_string(),
value,
};
if !tags.contains(&tag) {
tags.push(tag);
}
}
pub fn derive_tags(tag_mappings: &[TagMapping], payload: &str) -> Vec<Tag> {
if tag_mappings.is_empty() {
return Vec::new();
}
let parsed: serde_json::Value =
serde_json::from_str(payload).unwrap_or(serde_json::Value::Null);
let mut tags = Vec::new();
for mapping in tag_mappings {
let value = payload_field_value(&parsed, &mapping.field);
match value.and_then(|v| v.as_array()) {
Some(elements) if !elements.is_empty() => {
for element in elements {
if let Some(scalar) = json_scalar_to_string(element) {
push_unique_tag(&mut tags, &mapping.key, Some(scalar));
}
}
}
_ => {
let scalar = value.and_then(json_scalar_to_string);
push_unique_tag(&mut tags, &mapping.key, scalar);
}
}
}
tags
}
pub fn sensitive_field_is_granted(access_mapping: &RoleAccessMapping, subject_value: &str) -> bool {
access_mapping.can_read_sensitive || access_mapping.role.external_subject == subject_value
}
pub fn render_event(
event: &Event,
access_mapping: &RoleAccessMapping,
resolve_data_key: &impl Fn(&str, &str) -> Option<DataKey>,
) -> String {
decrypt_sensitive_fields(
&event.event_type.sensitive_fields,
&event.payload,
access_mapping,
resolve_data_key,
)
}
pub fn render_command(
command: &Command,
access_mapping: &RoleAccessMapping,
resolve_data_key: &impl Fn(&str, &str) -> Option<DataKey>,
) -> String {
decrypt_sensitive_fields(
&command.command_type.sensitive_fields,
&command.payload,
access_mapping,
resolve_data_key,
)
}
fn decrypt_sensitive_fields(
sensitive_fields: &[SensitiveField],
payload: &str,
access_mapping: &RoleAccessMapping,
resolve_data_key: &impl Fn(&str, &str) -> Option<DataKey>,
) -> String {
if sensitive_fields.is_empty() {
return payload.to_string();
}
let mut parsed: serde_json::Value = serde_json::from_str(payload).expect(
"decrypt_sensitive_fields: payload is always valid JSON by the time this is called",
);
for sf in sensitive_fields {
let Some(subject_value) =
payload_field_value(&parsed, &sf.subject_field).and_then(json_scalar_to_string)
else {
continue;
};
if !sensitive_field_is_granted(access_mapping, &subject_value) {
continue;
}
let Some(data_key) = resolve_data_key(&sf.subject_key, &subject_value) else {
continue;
};
let Some(ciphertext) = payload_field_value(&parsed, &sf.field).and_then(|v| v.as_str())
else {
continue;
};
let Ok(plaintext) = crate::encryption::decrypt_leaf(&data_key, ciphertext) else {
continue;
};
if let Some(slot) = payload_field_value_mut(&mut parsed, &sf.field) {
*slot = serde_json::Value::String(plaintext);
}
}
serde_json::to_string(&parsed).expect("re-serialising a parsed JSON Value is infallible")
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EventTypeRegistration {
Created(EventType),
Updated(EventType),
}
impl EventTypeRegistration {
pub fn event_type(&self) -> &EventType {
match self {
EventTypeRegistration::Created(et) | EventTypeRegistration::Updated(et) => et,
}
}
}
#[allow(clippy::too_many_arguments)]
pub fn register_event_type(
access_mapping: &RoleAccessMapping,
bounded_context: &BoundedContext,
name: String,
schema: String,
tag_mappings: Vec<TagMapping>,
sensitive_fields: Vec<SensitiveField>,
external_creation_allowed: bool,
direct_creation_allowed: bool,
system_triggered_allowed: bool,
system_triggered_schedule: Option<String>,
missed_occurrence_policy: Option<MissedOccurrencePolicy>,
event_read_allowed: bool,
existing: Option<&EventType>,
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<EventTypeRegistration> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if &access_mapping.bounded_context != bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
if bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_schema(&schema) {
return Err(Error::InvalidSchema.into());
}
if !valid_tag_mappings(&schema, &tag_mappings) {
return Err(Error::InvalidTagMapping.into());
}
if !valid_sensitive_fields(&schema, &sensitive_fields) {
return Err(Error::InvalidSensitiveField.into());
}
if tag_mappings
.iter()
.any(|m| sensitive_fields.iter().any(|s| s.field == m.field))
{
return Err(Error::SensitiveFieldTagOverlap.into());
}
if system_triggered_allowed
&& (system_triggered_schedule.is_none() || missed_occurrence_policy.is_none())
{
return Err(Error::MissingScheduleOrPolicy.into());
}
let scheduling_newly_enabled =
system_triggered_allowed && existing.is_none_or(|e| !e.system_triggered_allowed);
let Some(existing) = existing else {
return Ok(EventTypeRegistration::Created(EventType {
bounded_context: bounded_context.clone(),
name,
schema,
schema_version: 1,
tag_mappings,
sensitive_fields,
external_creation_allowed,
direct_creation_allowed,
system_triggered_allowed,
system_triggered_schedule,
missed_occurrence_policy,
schedule_position: scheduling_newly_enabled.then_some(now),
last_fired_at: None,
event_read_allowed,
}));
};
if !schema_is_backwards_compatible(&existing.schema, &schema) {
return Err(Error::SchemaIncompatible.into());
}
if !existing
.tag_mappings
.iter()
.all(|m| tag_mappings.iter().any(|n| n.key == m.key))
{
return Err(Error::TagMappingKeyDropped.into());
}
let schema_changed = existing.schema != schema;
Ok(EventTypeRegistration::Updated(EventType {
bounded_context: bounded_context.clone(),
name,
schema,
schema_version: if schema_changed {
existing.schema_version + 1
} else {
existing.schema_version
},
tag_mappings,
sensitive_fields,
external_creation_allowed,
direct_creation_allowed,
system_triggered_allowed,
system_triggered_schedule,
missed_occurrence_policy,
schedule_position: if scheduling_newly_enabled {
Some(now)
} else {
existing.schedule_position
},
last_fired_at: existing.last_fired_at,
event_read_allowed,
}))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CommandTypeRegistration {
Created(CommandType),
Updated(CommandType),
}
impl CommandTypeRegistration {
pub fn command_type(&self) -> &CommandType {
match self {
CommandTypeRegistration::Created(ct) | CommandTypeRegistration::Updated(ct) => ct,
}
}
}
#[allow(clippy::too_many_arguments)]
pub fn register_command_type(
access_mapping: &RoleAccessMapping,
bounded_context: &BoundedContext,
name: String,
schema: String,
tag_mappings: Vec<TagMapping>,
sensitive_fields: Vec<SensitiveField>,
rest_trigger_allowed: bool,
existing: Option<&CommandType>,
) -> crate::error::Result<CommandTypeRegistration> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if &access_mapping.bounded_context != bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
if bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_schema(&schema) {
return Err(Error::InvalidSchema.into());
}
if !valid_tag_mappings(&schema, &tag_mappings) {
return Err(Error::InvalidTagMapping.into());
}
if !valid_sensitive_fields(&schema, &sensitive_fields) {
return Err(Error::InvalidSensitiveField.into());
}
if tag_mappings
.iter()
.any(|m| sensitive_fields.iter().any(|s| s.field == m.field))
{
return Err(Error::SensitiveFieldTagOverlap.into());
}
let Some(existing) = existing else {
return Ok(CommandTypeRegistration::Created(CommandType {
bounded_context: bounded_context.clone(),
name,
schema,
schema_version: 1,
tag_mappings,
sensitive_fields,
rest_trigger_allowed,
}));
};
if !schema_is_backwards_compatible(&existing.schema, &schema) {
return Err(Error::SchemaIncompatible.into());
}
if !existing
.tag_mappings
.iter()
.all(|m| tag_mappings.iter().any(|n| n.key == m.key))
{
return Err(Error::TagMappingKeyDropped.into());
}
let schema_changed = existing.schema != schema;
Ok(CommandTypeRegistration::Updated(CommandType {
bounded_context: bounded_context.clone(),
name,
schema,
schema_version: if schema_changed {
existing.schema_version + 1
} else {
existing.schema_version
},
tag_mappings,
sensitive_fields,
rest_trigger_allowed,
}))
}
pub fn forget_subject(
access_mapping: &RoleAccessMapping,
key: &EncryptionKey,
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<EncryptionKey> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if access_mapping.bounded_context != key.bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
if key.status != EncryptionKeyStatus::Active {
return Err(Error::EncryptionKeyNotActive.into());
}
Ok(EncryptionKey {
status: EncryptionKeyStatus::Destroyed,
destroyed_at: Some(now),
..key.clone()
})
}
pub fn query_events(
access_mapping: &RoleAccessMapping,
event_types: &[EventType],
tags: Option<&[Tag]>,
after_sequence: Option<i64>,
bounded_context_events: &[Event],
resolve_data_key: impl Fn(&str, &str) -> Option<DataKey>,
) -> crate::error::Result<Vec<(i64, String)>> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if !event_types
.iter()
.all(|et| et.bounded_context == access_mapping.bounded_context)
{
return Err(Error::EventTypeNotInBoundedContext.into());
}
let after = after_sequence.unwrap_or(-1);
Ok(bounded_context_events
.iter()
.filter(|e| e.bounded_context == access_mapping.bounded_context)
.filter(|e| event_types.is_empty() || event_types.contains(&e.event_type))
.filter(|e| tags.is_none_or(|wanted| wanted.iter().any(|t| e.tags.contains(t))))
.filter(|e| e.sequence > after)
.map(|e| {
(
e.sequence,
render_event(e, access_mapping, &resolve_data_key),
)
})
.collect())
}
pub fn count_events(
access_mapping: &RoleAccessMapping,
event_types: &[EventType],
tags: Option<&[Tag]>,
bounded_context_events: &[Event],
) -> crate::error::Result<i64> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if !event_types
.iter()
.all(|et| et.bounded_context == access_mapping.bounded_context)
{
return Err(Error::EventTypeNotInBoundedContext.into());
}
Ok(bounded_context_events
.iter()
.filter(|e| e.bounded_context == access_mapping.bounded_context)
.filter(|e| event_types.is_empty() || event_types.contains(&e.event_type))
.filter(|e| tags.is_none_or(|wanted| wanted.iter().any(|t| e.tags.contains(t))))
.count() as i64)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventInspected {
pub event: Event,
pub rendered_payload: String,
}
pub fn inspect_event(
access_mapping: &RoleAccessMapping,
event: &Event,
resolve_data_key: impl Fn(&str, &str) -> Option<DataKey>,
) -> crate::error::Result<EventInspected> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if access_mapping.bounded_context != event.bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
Ok(EventInspected {
event: event.clone(),
rendered_payload: render_event(event, access_mapping, &resolve_data_key),
})
}
pub fn fetch_commands(
access_mapping: &RoleAccessMapping,
command_types: &[CommandType],
after: Option<chrono::DateTime<chrono::Utc>>,
before: Option<chrono::DateTime<chrono::Utc>>,
triggered_event: Option<&Event>,
bounded_context_commands: &[Command],
resolve_data_key: impl Fn(&str, &str) -> Option<DataKey>,
) -> crate::error::Result<Vec<String>> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.level != AccessLevel::Admin {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if !command_types
.iter()
.all(|ct| ct.bounded_context == access_mapping.bounded_context)
{
return Err(Error::CommandTypeNotInBoundedContext.into());
}
if triggered_event.is_some_and(|te| te.bounded_context != access_mapping.bounded_context) {
return Err(Error::TriggeredEventNotInBoundedContext.into());
}
Ok(bounded_context_commands
.iter()
.filter(|c| c.bounded_context == access_mapping.bounded_context)
.filter(|c| command_types.is_empty() || command_types.contains(&c.command_type))
.filter(|c| after.is_none_or(|a| c.metadata.created_at >= a))
.filter(|c| before.is_none_or(|b| c.metadata.created_at <= b))
.filter(|c| {
triggered_event.is_none_or(|te| match &te.origin {
EventOrigin::CommandTriggered { command } => command.id == c.id,
_ => false,
})
})
.map(|c| render_command(c, access_mapping, &resolve_data_key))
.collect())
}
pub fn create_all_events_subscription(
access_mapping: &RoleAccessMapping,
event_types: Vec<EventType>,
from_sequence: Option<i64>,
bounded_context_events: &[Event],
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<AllEventsSubscription> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !event_types
.iter()
.all(|et| et.bounded_context == access_mapping.bounded_context)
{
return Err(Error::EventTypeNotInBoundedContext.into());
}
let starting_point = from_sequence.unwrap_or_else(|| {
bounded_context_events
.iter()
.filter(|e| e.bounded_context == access_mapping.bounded_context)
.map(|e| e.sequence)
.max()
.unwrap_or(-1)
});
Ok(AllEventsSubscription {
bounded_context: access_mapping.bounded_context.clone(),
access_mapping: access_mapping.clone(),
created_at: now,
from_sequence: starting_point,
event_types,
})
}
pub fn create_event_type_subscription(
access_mapping: &RoleAccessMapping,
event_type: &EventType,
filters: Vec<Filter>,
from_sequence: Option<i64>,
bounded_context_events: &[Event],
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<EventTypeSubscription> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if access_mapping.bounded_context != event_type.bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
if event_type.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_filters(event_type, &filters) {
return Err(Error::InvalidFilter.into());
}
let starting_point = from_sequence.unwrap_or_else(|| {
bounded_context_events
.iter()
.filter(|e| e.bounded_context == event_type.bounded_context)
.map(|e| e.sequence)
.max()
.unwrap_or(-1)
});
Ok(EventTypeSubscription {
bounded_context: event_type.bounded_context.clone(),
access_mapping: access_mapping.clone(),
created_at: now,
from_sequence: starting_point,
event_type: event_type.clone(),
filters,
})
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventDelivered {
pub subscription: Subscription,
pub event: Event,
pub rendered_payload: String,
}
pub fn deliver_to_subscriptions(
event: &Event,
subscriptions: &[Subscription],
resolve_data_key: impl Fn(&str, &str) -> Option<DataKey>,
) -> Vec<EventDelivered> {
subscriptions
.iter()
.filter(|s| {
s.bounded_context() == &event.bounded_context
&& s.starting_sequence() < event.sequence
&& s.access_mapping().status == RoleStatus::Active
})
.filter(|s| match s {
Subscription::AllEventsSubscription(a) => {
a.event_types.is_empty() || a.event_types.contains(&event.event_type)
}
Subscription::EventTypeSubscription(e) => {
e.event_type == event.event_type && matches_filters(event, &e.filters)
}
})
.map(|s| EventDelivered {
subscription: s.clone(),
event: event.clone(),
rendered_payload: render_event(event, s.access_mapping(), &resolve_data_key),
})
.collect()
}
#[derive(Clone)]
pub struct EventBroadcaster {
sender: tokio::sync::broadcast::Sender<Event>,
instance_id: String,
}
impl EventBroadcaster {
pub fn new(capacity: usize) -> Self {
let (sender, _receiver) = tokio::sync::broadcast::channel(capacity);
Self {
sender,
instance_id: crate::shared::generate_token_id(),
}
}
pub fn instance_id(&self) -> &str {
&self.instance_id
}
pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Event> {
self.sender.subscribe()
}
pub fn publish(&self, event: &Event) {
let _ = self.sender.send(event.clone());
}
}
pub fn fetch_events(
token: &EventReadToken,
events: &[Event],
filters: &[Filter],
after_sequence: Option<i64>,
) -> crate::error::Result<Vec<Event>> {
if token.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let read_type = &token.event_type;
if !read_type.event_read_allowed {
return Err(Error::EventReadNotAllowed.into());
}
if !valid_filters(read_type, filters) {
return Err(Error::InvalidFilter.into());
}
let after = after_sequence.unwrap_or(-1);
Ok(events
.iter()
.filter(|e| e.bounded_context == read_type.bounded_context)
.filter(|e| &e.event_type == read_type)
.filter(|e| e.sequence > after)
.filter(|e| matches_filters(e, filters))
.cloned()
.collect())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CursorUpdate {
Created(Box<ReadCursor>),
Advanced {
sequence: i64,
updated_at: chrono::DateTime<chrono::Utc>,
},
Unchanged,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ConsumeEventsResult {
pub served: Vec<Event>,
pub cursor_update: CursorUpdate,
}
pub fn consume_events(
token: &EventReadToken,
existing_cursor: Option<&ReadCursor>,
ack_mode: Option<AckMode>,
events: &[Event],
filters: &[Filter],
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<ConsumeEventsResult> {
if token.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let read_type = &token.event_type;
if !read_type.event_read_allowed {
return Err(Error::EventReadNotAllowed.into());
}
if !valid_filters(read_type, filters) {
return Err(Error::InvalidFilter.into());
}
let is_new = existing_cursor.is_none();
if is_new && ack_mode.is_none() {
return Err(Error::CursorAckModeMismatch.into());
}
if let (Some(cursor), Some(requested)) = (existing_cursor, ack_mode) {
if requested != cursor.ack_mode {
return Err(Error::CursorAckModeMismatch.into());
}
}
let mode = ack_mode
.or_else(|| existing_cursor.map(|c| c.ack_mode))
.expect("is_new implies ack_mode.is_some(), checked above");
let position = existing_cursor.map(|c| c.sequence).unwrap_or(-1);
let served: Vec<Event> = events
.iter()
.filter(|e| e.bounded_context == read_type.bounded_context)
.filter(|e| &e.event_type == read_type)
.filter(|e| e.sequence > position)
.filter(|e| matches_filters(e, filters))
.cloned()
.collect();
let next_position = match mode {
AckMode::AutoAdvance => highest_sequence(&served).unwrap_or(position),
AckMode::ManualAck => position,
};
let cursor_update = if is_new {
CursorUpdate::Created(Box::new(ReadCursor {
token: token.clone(),
ack_mode: mode,
sequence: next_position,
updated_at: now,
}))
} else if mode == AckMode::AutoAdvance {
CursorUpdate::Advanced {
sequence: next_position,
updated_at: now,
}
} else {
CursorUpdate::Unchanged
};
Ok(ConsumeEventsResult {
served,
cursor_update,
})
}
pub fn acknowledge_events(
token: &EventReadToken,
cursor: Option<&ReadCursor>,
sequence: i64,
now: chrono::DateTime<chrono::Utc>,
) -> crate::error::Result<(i64, chrono::DateTime<chrono::Utc>)> {
if token.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let cursor = cursor.ok_or(Error::NoReadCursor)?;
if cursor.ack_mode != AckMode::ManualAck {
return Err(Error::NotManualAckCursor.into());
}
if sequence < cursor.sequence {
return Err(Error::AcknowledgementRegresses.into());
}
Ok((sequence, now))
}
pub fn create_external_event(
adapter: &ExternalEventToken,
payload: String,
source_content: String,
source_context: Option<String>,
next_sequence: i64,
now: chrono::DateTime<chrono::Utc>,
resolve_key: impl Fn(&str, &str) -> (EncryptionKey, DataKey),
) -> crate::error::Result<Event> {
if adapter.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let event_type = &adapter.event_type;
if !event_type.external_creation_allowed {
return Err(Error::ExternalCreationNotAllowed.into());
}
if event_type.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_payload(&event_type.schema, &payload) {
return Err(Error::PayloadDoesNotMatchSchema.into());
}
let protected = protect_sensitive_fields(&event_type.sensitive_fields, &payload, resolve_key);
Ok(Event {
bounded_context: event_type.bounded_context.clone(),
event_type: event_type.clone(),
payload: protected.payload,
metadata: Metadata {
r#type: event_type.name.clone(),
version: event_type.schema_version,
client_id: adapter.id.clone(),
created_at: now,
},
sequence: next_sequence,
tags: derive_tags(&event_type.tag_mappings, &payload),
encryption_keys: protected.encryption_keys,
origin: EventOrigin::ExternalTriggered {
source_content,
source_context,
},
})
}
pub fn create_direct_event(
adapter: &DirectCreationToken,
payload: String,
next_sequence: i64,
now: chrono::DateTime<chrono::Utc>,
resolve_key: impl Fn(&str, &str) -> (EncryptionKey, DataKey),
) -> crate::error::Result<Event> {
if adapter.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let event_type = &adapter.event_type;
if !event_type.direct_creation_allowed {
return Err(Error::DirectCreationNotAllowed.into());
}
if event_type.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_payload(&event_type.schema, &payload) {
return Err(Error::PayloadDoesNotMatchSchema.into());
}
let protected = protect_sensitive_fields(&event_type.sensitive_fields, &payload, resolve_key);
Ok(Event {
bounded_context: event_type.bounded_context.clone(),
event_type: event_type.clone(),
payload: protected.payload,
metadata: Metadata {
r#type: event_type.name.clone(),
version: event_type.schema_version,
client_id: adapter.id.clone(),
created_at: now,
},
sequence: next_sequence,
tags: derive_tags(&event_type.tag_mappings, &payload),
encryption_keys: protected.encryption_keys,
origin: EventOrigin::DirectlyCreated,
})
}
pub fn next_occurrence_after(
schedule: &str,
instant: chrono::DateTime<chrono::Utc>,
) -> Option<chrono::DateTime<chrono::Utc>> {
use std::str::FromStr;
cron::Schedule::from_str(schedule)
.ok()?
.after(&instant)
.next()
}
#[allow(clippy::too_many_arguments)]
pub fn create_system_event(
event_type: &EventType,
occurrence_at: chrono::DateTime<chrono::Utc>,
now: chrono::DateTime<chrono::Utc>,
next_sequence: i64,
resolve_payload: impl FnOnce() -> String,
resolve_key: impl Fn(&str, &str) -> (EncryptionKey, DataKey),
) -> Option<(Event, chrono::DateTime<chrono::Utc>)> {
if !event_type.system_triggered_allowed {
return None;
}
if event_type.bounded_context.status != BoundedContextStatus::Active {
return None;
}
let schedule = event_type.system_triggered_schedule.as_deref()?;
let policy = event_type.missed_occurrence_policy?;
let position = event_type.schedule_position?;
if occurrence_at > now || occurrence_at <= position {
return None;
}
let no_gap = next_occurrence_after(schedule, position) == Some(occurrence_at);
let nothing_later_is_due =
next_occurrence_after(schedule, occurrence_at).is_none_or(|next| next > now);
let eligible = match policy {
MissedOccurrencePolicy::ReplayBacklog => true,
MissedOccurrencePolicy::FireOnce => nothing_later_is_due,
MissedOccurrencePolicy::Skip => no_gap && nothing_later_is_due,
};
if !eligible {
return None;
}
let payload = resolve_payload();
let protected = protect_sensitive_fields(&event_type.sensitive_fields, &payload, resolve_key);
let event = Event {
bounded_context: event_type.bounded_context.clone(),
event_type: event_type.clone(),
payload: protected.payload,
metadata: Metadata {
r#type: event_type.name.clone(),
version: event_type.schema_version,
client_id: "system".to_string(),
created_at: now,
},
sequence: next_sequence,
tags: derive_tags(&event_type.tag_mappings, &payload),
encryption_keys: protected.encryption_keys,
origin: EventOrigin::SystemTriggered,
};
Some((event, occurrence_at))
}
pub fn skip_missed_occurrences(
event_type: &EventType,
now: chrono::DateTime<chrono::Utc>,
) -> Option<chrono::DateTime<chrono::Utc>> {
if !event_type.system_triggered_allowed {
return None;
}
if event_type.missed_occurrence_policy != Some(MissedOccurrencePolicy::Skip) {
return None;
}
if event_type.bounded_context.status != BoundedContextStatus::Active {
return None;
}
let position = event_type.schedule_position?;
if position >= now {
return None;
}
Some(now)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CommandAuthorised {
pub command_type: CommandType,
pub payload: String,
pub client_id: String,
}
pub fn authorise_command_trigger(
token: &CommandToken,
payload: String,
) -> crate::error::Result<CommandAuthorised> {
if token.status != TokenStatus::Active {
return Err(crate::access_control::Error::TokenNotActive.into());
}
let command_type = &token.command_type;
if !command_type.rest_trigger_allowed {
return Err(Error::RestTriggerNotAllowed.into());
}
if command_type.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_payload(&command_type.schema, &payload) {
return Err(Error::PayloadDoesNotMatchSchema.into());
}
Ok(CommandAuthorised {
command_type: command_type.clone(),
payload,
client_id: token.id.clone(),
})
}
pub fn authorise_command_submission(
access_mapping: &RoleAccessMapping,
command_type: &CommandType,
payload: String,
) -> crate::error::Result<CommandAuthorised> {
if access_mapping.status != RoleStatus::Active {
return Err(crate::access_control::Error::GrantNotActive.into());
}
if !matches!(
access_mapping.level,
AccessLevel::Write | AccessLevel::Admin
) {
return Err(crate::access_control::Error::InsufficientAccessLevel.into());
}
if access_mapping.bounded_context != command_type.bounded_context {
return Err(crate::access_control::Error::GrantBoundedContextMismatch.into());
}
if command_type.bounded_context.status != BoundedContextStatus::Active {
return Err(Error::BoundedContextArchived.into());
}
if !valid_payload(&command_type.schema, &payload) {
return Err(Error::PayloadDoesNotMatchSchema.into());
}
Ok(CommandAuthorised {
command_type: command_type.clone(),
payload,
client_id: access_mapping.role.id.clone(),
})
}
pub fn consistency_boundary_and_matching_events(
bounded_context_events: &[Event],
consistency_tags: &[Tag],
) -> (Option<i64>, Vec<Event>) {
let matching: Vec<Event> = bounded_context_events
.iter()
.filter(|e| consistency_tags.iter().any(|t| e.tags.contains(t)))
.cloned()
.collect();
let boundary = highest_sequence(&matching);
(boundary, matching)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProcessCommandResult {
pub command: Command,
pub events: Vec<Event>,
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
skip_all,
fields(
bounded_context = %command_type.bounded_context.name,
command_type = %command_type.name,
)
)]
pub fn process_command(
id: String,
command_type: &CommandType,
payload: &str,
client_id: &str,
bounded_context_events: &[Event],
decision: CommandDecision,
resolve_event_type: impl Fn(&str) -> Option<EventType>,
mut next_sequence: impl FnMut() -> i64,
now: chrono::DateTime<chrono::Utc>,
resolve_key: impl Fn(&str, &str) -> (EncryptionKey, DataKey),
) -> crate::error::Result<ProcessCommandResult> {
let event_specs = match decision {
CommandDecision::Accepted { events } => events,
CommandDecision::Rejected { reason, kind } => {
return Err(crate::error::Error::CommandRejected { reason, kind });
}
};
let consistency_tags = derive_tags(&command_type.tag_mappings, payload);
let (consistency_boundary, _matching_events) =
consistency_boundary_and_matching_events(bounded_context_events, &consistency_tags);
let protected = protect_sensitive_fields(&command_type.sensitive_fields, payload, &resolve_key);
let command = Command {
id,
bounded_context: command_type.bounded_context.clone(),
command_type: command_type.clone(),
payload: protected.payload,
metadata: Metadata {
r#type: command_type.name.clone(),
version: command_type.schema_version,
client_id: client_id.to_string(),
created_at: now,
},
encryption_keys: protected.encryption_keys,
consistency_tags,
consistency_boundary,
};
let mut events = Vec::with_capacity(event_specs.len());
for spec in event_specs {
let event_type = resolve_event_type(&spec.event_type)
.ok_or_else(|| Error::UnregisteredEventType(spec.event_type.clone()))?;
let spec_payload = spec.payload.to_string();
let event_protected =
protect_sensitive_fields(&event_type.sensitive_fields, &spec_payload, &resolve_key);
events.push(Event {
bounded_context: command_type.bounded_context.clone(),
event_type: event_type.clone(),
payload: event_protected.payload,
metadata: Metadata {
r#type: event_type.name.clone(),
version: event_type.schema_version,
client_id: client_id.to_string(),
created_at: now,
},
sequence: next_sequence(),
tags: derive_tags(&event_type.tag_mappings, &spec_payload),
encryption_keys: event_protected.encryption_keys,
origin: EventOrigin::CommandTriggered {
command: Box::new(command.clone()),
},
});
}
Ok(ProcessCommandResult { command, events })
}