use std::collections::{HashMap, HashSet};
use bacnet_objects::database::ObjectDatabase;
use bacnet_objects::event::{
EventStateChange, EventTransition, EventTransitionCommit, EventTransitionCommitError,
};
use bacnet_objects::event_enrollment::EventEnrollmentReliabilityCommit;
use bacnet_objects::event_enrollment::{EventEnrollmentEvalState, EventEnrollmentMonitoredSource};
use bacnet_types::enums::{EventState, EventType, PropertyIdentifier, Reliability};
use bacnet_types::primitives::{ObjectIdentifier, PropertyValue};
use super::EventEnrollmentTransition;
use crate::server::event_notification_payload::{
CapturedReferencedValue, CommittedNotificationPayload, EventEnrollmentProjectionSnapshot,
project_event_enrollment_payload,
};
use crate::server::event_timestamp::{
SampledEventClock, confirm_event_timestamp, stage_event_timestamp,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventEnrollmentEvaluationStage {
Evaluation,
EvaluationSource,
EvaluationState,
Reliability,
EventTransition,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventEnrollmentEvaluationOutcome {
NoTransition,
CancellationCommitted,
ObservationUnavailable,
Rejected,
LandedAfterError,
}
impl EventEnrollmentEvaluationOutcome {
pub fn is_failure(self) -> bool {
matches!(self, Self::Rejected | Self::LandedAfterError)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EventEnrollmentEvaluationDiagnostic {
pub enrollment_oid: ObjectIdentifier,
pub stage: EventEnrollmentEvaluationStage,
pub outcome: EventEnrollmentEvaluationOutcome,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventEnrollmentReliabilityCause {
Configuration,
MonitoredObject,
FaultAlgorithm,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EventEnrollmentReliabilityResult {
pub enrollment_oid: ObjectIdentifier,
pub monitored_oid: Option<ObjectIdentifier>,
pub previous_reliability: Reliability,
pub new_reliability: Reliability,
pub state_change: Option<EventStateChange>,
pub distribute: bool,
pub cause: EventEnrollmentReliabilityCause,
}
impl EventEnrollmentReliabilityResult {
pub fn event_type(&self, configured_algorithm: EventType) -> Option<EventType> {
self.state_change
.as_ref()
.map(|change| change.event_type(configured_algorithm))
}
}
#[derive(Debug, Clone, Default, PartialEq)]
pub struct EventEnrollmentEvaluationReport {
pub transitions: Vec<EventEnrollmentTransition>,
pub reliability_results: Vec<EventEnrollmentReliabilityResult>,
pub diagnostics: Vec<EventEnrollmentEvaluationDiagnostic>,
}
pub(crate) enum CommittedEventEnrollmentResult {
Normal(EventEnrollmentTransition),
Reliability(EventEnrollmentReliabilityResult),
}
pub(crate) struct CommittedEventEnrollmentDelivery {
pub(crate) result: CommittedEventEnrollmentResult,
pub(crate) ack_required: bool,
pub(crate) recipient_clock: SampledEventClock,
pub(crate) event_type: EventType,
pub(crate) event_values: CommittedNotificationPayload,
}
pub(crate) struct EventEnrollmentEvaluationBatch {
pub(crate) report: EventEnrollmentEvaluationReport,
pub(crate) deliveries: Vec<CommittedEventEnrollmentDelivery>,
}
pub(crate) fn log_evaluation_report(report: &EventEnrollmentEvaluationReport) {
for result in &report.reliability_results {
tracing::debug!(
enrollment = %result.enrollment_oid,
monitored = ?result.monitored_oid,
previous = ?result.previous_reliability,
new = ?result.new_reliability,
state_change = ?result.state_change,
distribute = result.distribute,
cause = ?result.cause,
"Event enrollment: reliability committed"
);
}
for diagnostic in report
.diagnostics
.iter()
.filter(|diagnostic| diagnostic.outcome.is_failure())
{
tracing::warn!(
enrollment = %diagnostic.enrollment_oid,
stage = ?diagnostic.stage,
outcome = ?diagnostic.outcome,
"Event enrollment: internal commit failed"
);
}
}
#[derive(Default)]
pub(super) struct EnrollmentUpdate {
eval_state: Option<EventEnrollmentEvalState>,
eval_source: Option<Option<EventEnrollmentMonitoredSource>>,
clears_invalidation: bool,
fired: Option<FiredTransition>,
reliability: Option<ReliabilityUpdate>,
observation_unavailable: bool,
canceled: bool,
}
pub(super) struct FiredTransition {
pub(super) monitored_oid: ObjectIdentifier,
pub(super) event_type: EventType,
pub(super) from: EventState,
pub(super) to: EventState,
pub(super) distribute: bool,
pub(super) ack_required: bool,
pub(super) projection: EventEnrollmentProjectionSnapshot,
}
pub(super) struct ReliabilityUpdate {
pub(super) monitored_oid: Option<ObjectIdentifier>,
pub(super) previous: Reliability,
pub(super) desired: Reliability,
pub(super) from: EventState,
pub(super) to: EventState,
pub(super) distribute: bool,
pub(super) ack_required: bool,
pub(super) cause: EventEnrollmentReliabilityCause,
pub(super) referenced_value: CapturedReferencedValue,
}
impl EnrollmentUpdate {
pub(super) fn set_eval_state(&mut self, state: EventEnrollmentEvalState) {
self.eval_state = Some(state);
}
pub(super) fn reset_eval_state(&mut self) {
self.eval_state = Some(EventEnrollmentEvalState::default());
self.clears_invalidation = true;
}
pub(super) fn cancel_pending(&mut self, state: EventEnrollmentEvalState) {
self.eval_state = Some(state);
self.canceled = true;
}
pub(super) fn set_eval_source(&mut self, source: Option<EventEnrollmentMonitoredSource>) {
self.eval_source = Some(source);
}
pub(super) fn reset_eval_source_for_observation_gap(
&mut self,
previous: Option<Option<EventEnrollmentMonitoredSource>>,
) {
self.eval_source = match previous {
Some(Some(_)) => Some(None),
Some(None) | None => None,
};
}
pub(super) fn fire(&mut self, fired: FiredTransition) {
debug_assert!(
self.fired.is_none(),
"one enrollment fires at most once per pass"
);
self.fired = Some(fired);
}
pub(super) fn set_reliability(&mut self, reliability: ReliabilityUpdate) {
debug_assert!(
self.reliability.is_none() && self.fired.is_none(),
"one enrollment commits one transition family per pass"
);
self.reliability = Some(reliability);
}
pub(super) fn observation_unavailable(&mut self) {
self.fired = None;
self.reliability = None;
self.canceled = false;
self.observation_unavailable = true;
}
}
fn diagnostic(
enrollment_oid: ObjectIdentifier,
stage: EventEnrollmentEvaluationStage,
outcome: EventEnrollmentEvaluationOutcome,
) -> EventEnrollmentEvaluationDiagnostic {
EventEnrollmentEvaluationDiagnostic {
enrollment_oid,
stage,
outcome,
}
}
fn event_state_landed(db: &ObjectDatabase, oid: &ObjectIdentifier, state: EventState) -> bool {
matches!(
db.get(oid)
.and_then(|object| object.read_property(PropertyIdentifier::EVENT_STATE, None).ok()),
Some(PropertyValue::Enumerated(raw)) if raw == state.to_raw()
)
}
fn reliability_or_state_landed(
db: &ObjectDatabase,
oid: &ObjectIdentifier,
update: &ReliabilityUpdate,
) -> bool {
let Some(object) = db.get(oid) else {
return false;
};
let reliability_landed = update.previous != update.desired
&& matches!(
object.read_property(PropertyIdentifier::RELIABILITY, None),
Ok(PropertyValue::Enumerated(raw)) if raw == update.desired.to_raw()
);
let state_landed = update.from != update.to
&& matches!(
object.read_property(PropertyIdentifier::EVENT_STATE, None),
Ok(PropertyValue::Enumerated(raw)) if raw == update.to.to_raw()
);
reliability_landed || state_landed
}
fn clear_source_ownership(
db: &mut ObjectDatabase,
oid: ObjectIdentifier,
database_eval_sources: &HashSet<ObjectIdentifier>,
) {
if database_eval_sources.contains(&oid) {
db.set_enrollment_eval_source(oid, None);
} else if let Some(object) = db.get_mut(&oid) {
let _ = object.set_enrollment_eval_source_internal(None);
}
}
#[cfg(test)]
pub(super) fn apply_updates(
db: &mut ObjectDatabase,
oids: &[ObjectIdentifier],
updates: HashMap<ObjectIdentifier, EnrollmentUpdate>,
database_eval_sources: &HashSet<ObjectIdentifier>,
) -> EventEnrollmentEvaluationReport {
apply_updates_for_delivery(db, oids, updates, database_eval_sources).report
}
pub(super) fn apply_updates_for_delivery(
db: &mut ObjectDatabase,
oids: &[ObjectIdentifier],
mut updates: HashMap<ObjectIdentifier, EnrollmentUpdate>,
database_eval_sources: &HashSet<ObjectIdentifier>,
) -> EventEnrollmentEvaluationBatch {
let mut report = EventEnrollmentEvaluationReport::default();
let mut deliveries = Vec::new();
for &oid in oids {
let Some(update) = updates.remove(&oid) else {
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::Evaluation,
EventEnrollmentEvaluationOutcome::NoTransition,
));
continue;
};
let mut source_failed = false;
let mut state_failed = false;
if let Some(source) = update.eval_source {
if database_eval_sources.contains(&oid) {
db.set_enrollment_eval_source(oid, source);
} else {
let failed = db.get_mut(&oid).is_none_or(|object| {
object.set_enrollment_eval_source_internal(source).is_err()
});
if failed {
source_failed = true;
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::EvaluationSource,
EventEnrollmentEvaluationOutcome::Rejected,
));
}
}
}
let mut state_committed = false;
if source_failed {
state_failed = db.get_mut(&oid).is_none_or(|object| {
object
.set_enrollment_eval_state_internal(EventEnrollmentEvalState::default())
.is_err()
});
if state_failed {
db.set_enrollment_eval_state_invalidated(oid, true);
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::EvaluationState,
EventEnrollmentEvaluationOutcome::Rejected,
));
} else {
state_committed = true;
}
} else if let Some(state) = update.eval_state {
state_failed = db
.get_mut(&oid)
.is_none_or(|object| object.set_enrollment_eval_state_internal(state).is_err());
if state_failed {
clear_source_ownership(db, oid, database_eval_sources);
db.set_enrollment_eval_state_invalidated(oid, true);
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::EvaluationState,
EventEnrollmentEvaluationOutcome::Rejected,
));
} else {
state_committed = true;
if update.clears_invalidation {
db.set_enrollment_eval_state_invalidated(oid, false);
}
}
}
if update.observation_unavailable {
if source_failed || state_failed {
db.set_enrollment_eval_state_invalidated(oid, true);
}
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::Reliability,
EventEnrollmentEvaluationOutcome::ObservationUnavailable,
));
continue;
}
if state_failed {
continue;
}
if let Some(reliability) = update.reliability {
let coordinate = EventTransition::for_target_state(reliability.to);
let staged_timestamp = stage_event_timestamp(db);
let change = EventStateChange {
from: reliability.from,
to: reliability.to,
};
let transition = EventTransitionCommit {
change: change.clone(),
coordinate,
ack_required: reliability.ack_required,
timestamp: staged_timestamp.sample.timestamp.clone(),
message_text: Some(String::new()),
};
let commit = EventEnrollmentReliabilityCommit {
reliability: reliability.desired,
transition: Some(transition),
};
let commit_result = db
.get_mut(&oid)
.map_or(Err(EventTransitionCommitError::Unsupported), |object| {
object.commit_event_enrollment_reliability_internal(commit)
});
if commit_result.is_err() {
let outcome = if reliability_or_state_landed(db, &oid, &reliability) {
EventEnrollmentEvaluationOutcome::LandedAfterError
} else {
EventEnrollmentEvaluationOutcome::Rejected
};
clear_source_ownership(db, oid, database_eval_sources);
db.set_enrollment_eval_state_invalidated(oid, true);
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::Reliability,
outcome,
));
continue;
}
let recipient_clock = confirm_event_timestamp(db, staged_timestamp).clock;
let result = EventEnrollmentReliabilityResult {
enrollment_oid: oid,
monitored_oid: reliability.monitored_oid,
previous_reliability: reliability.previous,
new_reliability: reliability.desired,
state_change: Some(change),
distribute: reliability.distribute,
cause: reliability.cause,
};
report.reliability_results.push(result.clone());
let event_type = EventType::CHANGE_OF_RELIABILITY;
if let Some(event_values) = project_event_enrollment_payload(
db,
oid,
reliability.monitored_oid,
result
.state_change
.as_ref()
.expect("committed Reliability delivery has a state change"),
event_type,
None,
Some(&reliability.referenced_value),
) {
deliveries.push(CommittedEventEnrollmentDelivery {
result: CommittedEventEnrollmentResult::Reliability(result),
ack_required: reliability.ack_required,
recipient_clock,
event_type,
event_values,
});
} else {
tracing::debug!(
enrollment = %oid,
"Committed Event Enrollment reliability payload rejected; suppressing distribution"
);
}
continue;
}
let Some(fired) = update.fired else {
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::EvaluationState,
if update.canceled && state_committed {
EventEnrollmentEvaluationOutcome::CancellationCommitted
} else {
EventEnrollmentEvaluationOutcome::NoTransition
},
));
continue;
};
if source_failed && fired.from == fired.to {
continue;
}
let coordinate = EventTransition::for_target_state(fired.to);
let staged_timestamp = stage_event_timestamp(db);
let change = EventStateChange {
from: fired.from,
to: fired.to,
};
let commit = EventTransitionCommit {
change: change.clone(),
coordinate,
ack_required: fired.ack_required,
timestamp: staged_timestamp.sample.timestamp.clone(),
message_text: Some(String::new()),
};
let commit_result = db
.get_mut(&oid)
.map_or(Err(EventTransitionCommitError::Unsupported), |object| {
object.commit_event_transition_internal(commit)
});
if commit_result.is_err() {
let outcome = if fired.from != fired.to && event_state_landed(db, &oid, fired.to) {
EventEnrollmentEvaluationOutcome::LandedAfterError
} else {
EventEnrollmentEvaluationOutcome::Rejected
};
clear_source_ownership(db, oid, database_eval_sources);
db.set_enrollment_eval_state_invalidated(oid, true);
report.diagnostics.push(diagnostic(
oid,
EventEnrollmentEvaluationStage::EventTransition,
outcome,
));
continue;
}
let recipient_clock = confirm_event_timestamp(db, staged_timestamp).clock;
let event_type = change.event_type(fired.event_type);
let result = EventEnrollmentTransition {
enrollment_oid: oid,
monitored_oid: fired.monitored_oid,
change,
event_type,
distribute: fired.distribute,
};
report.transitions.push(result.clone());
if let Some(event_values) = project_event_enrollment_payload(
db,
oid,
Some(fired.monitored_oid),
&result.change,
event_type,
Some(&fired.projection),
None,
) {
deliveries.push(CommittedEventEnrollmentDelivery {
result: CommittedEventEnrollmentResult::Normal(result),
ack_required: fired.ack_required,
recipient_clock,
event_type,
event_values,
});
} else {
tracing::debug!(
enrollment = %oid,
?event_type,
"Committed Event Enrollment payload rejected; suppressing distribution"
);
}
}
EventEnrollmentEvaluationBatch { report, deliveries }
}