bacnet-server 0.15.0

use super::*;
use crate::event_enrollment::{
    CommittedEventEnrollmentDelivery, CommittedEventEnrollmentResult,
    EventEnrollmentReliabilityCause, EventEnrollmentReliabilityResult, EventEnrollmentTransition,
};
use crate::server::event_notification_payload::CommittedNotificationPayload;
use crate::server::event_timestamp::SampledEventClock;
use bacnet_objects::event::{
    EventStateChange, EventTransitionCommit, EventTransitionCommitError, TransitionOutcome,
};
use bacnet_objects::traits::BACnetObject;
use bacnet_services::alarm_event::NotificationParameters;
use bacnet_types::enums::{EventState, EventType};
use bacnet_types::primitives::{BACnetTimeStamp, Date, StatusFlags, Time};
use std::sync::atomic::{AtomicUsize, Ordering};

#[derive(Clone)]
enum IndexedHistoryRead {
    Value(PropertyValue),
    Missing,
    Unreadable,
}

struct AtomicHistoryObject {
    oid: ObjectIdentifier,
    event_type: Option<PropertyValue>,
    timestamps: [IndexedHistoryRead; 3],
    messages: [IndexedHistoryRead; 3],
    status_flags: Option<PropertyValue>,
    commits: Arc<AtomicUsize>,
}

impl AtomicHistoryObject {
    fn indexed(
        values: &[IndexedHistoryRead; 3],
        array_index: Option<u32>,
    ) -> Result<PropertyValue, Error> {
        let Some(index @ 1..=3) = array_index else {
            return Err(Error::protocol(
                ErrorClass::PROPERTY,
                ErrorCode::INVALID_ARRAY_INDEX,
            ));
        };
        match &values[index as usize - 1] {
            IndexedHistoryRead::Value(value) => Ok(value.clone()),
            IndexedHistoryRead::Missing => Err(Error::protocol(
                ErrorClass::PROPERTY,
                ErrorCode::UNKNOWN_PROPERTY,
            )),
            IndexedHistoryRead::Unreadable => Err(Error::protocol(
                ErrorClass::PROPERTY,
                ErrorCode::READ_ACCESS_DENIED,
            )),
        }
    }
}

impl BACnetObject for AtomicHistoryObject {
    fn object_identifier(&self) -> ObjectIdentifier {
        self.oid
    }

    fn object_name(&self) -> &str {
        "atomic-history"
    }

    fn read_property(
        &self,
        property: PropertyIdentifier,
        array_index: Option<u32>,
    ) -> Result<PropertyValue, Error> {
        match property {
            p if p == PropertyIdentifier::OBJECT_IDENTIFIER => {
                Ok(PropertyValue::ObjectIdentifier(self.oid))
            }
            p if p == PropertyIdentifier::OBJECT_NAME => {
                Ok(PropertyValue::CharacterString(self.object_name().into()))
            }
            p if p == PropertyIdentifier::OBJECT_TYPE => {
                Ok(PropertyValue::Enumerated(ObjectType::ANALOG_INPUT.to_raw()))
            }
            p if p == PropertyIdentifier::NOTIFICATION_CLASS => Ok(PropertyValue::Unsigned(0)),
            p if p == PropertyIdentifier::NOTIFY_TYPE => {
                Ok(PropertyValue::Enumerated(NotifyType::ALARM.to_raw()))
            }
            p if p == PropertyIdentifier::EVENT_TYPE => self.event_type.clone().ok_or(
                Error::protocol(ErrorClass::PROPERTY, ErrorCode::UNKNOWN_PROPERTY),
            ),
            p if p == PropertyIdentifier::EVENT_TIME_STAMPS => {
                Self::indexed(&self.timestamps, array_index)
            }
            p if p == PropertyIdentifier::EVENT_MESSAGE_TEXTS => {
                Self::indexed(&self.messages, array_index)
            }
            p if p == PropertyIdentifier::PRESENT_VALUE => Ok(PropertyValue::Real(50.0)),
            p if p == PropertyIdentifier::STATUS_FLAGS => self.status_flags.clone().ok_or(
                Error::protocol(ErrorClass::PROPERTY, ErrorCode::UNKNOWN_PROPERTY),
            ),
            p if p == PropertyIdentifier::RELIABILITY => Ok(PropertyValue::Enumerated(0)),
            p if p == PropertyIdentifier::HIGH_LIMIT => Ok(PropertyValue::Real(80.0)),
            p if p == PropertyIdentifier::LOW_LIMIT => Ok(PropertyValue::Real(20.0)),
            p if p == PropertyIdentifier::DEADBAND => Ok(PropertyValue::Real(2.0)),
            _ => Err(Error::protocol(
                ErrorClass::PROPERTY,
                ErrorCode::UNKNOWN_PROPERTY,
            )),
        }
    }

    fn write_property(
        &mut self,
        _property: PropertyIdentifier,
        _array_index: Option<u32>,
        _value: PropertyValue,
        _priority: Option<u8>,
    ) -> Result<(), Error> {
        Err(Error::protocol(
            ErrorClass::PROPERTY,
            ErrorCode::WRITE_ACCESS_DENIED,
        ))
    }

    fn property_list(&self) -> std::borrow::Cow<'static, [PropertyIdentifier]> {
        std::borrow::Cow::Borrowed(&[
            PropertyIdentifier::OBJECT_IDENTIFIER,
            PropertyIdentifier::OBJECT_NAME,
            PropertyIdentifier::OBJECT_TYPE,
            PropertyIdentifier::NOTIFICATION_CLASS,
            PropertyIdentifier::NOTIFY_TYPE,
            PropertyIdentifier::EVENT_TIME_STAMPS,
            PropertyIdentifier::EVENT_MESSAGE_TEXTS,
            PropertyIdentifier::PRESENT_VALUE,
            PropertyIdentifier::STATUS_FLAGS,
            PropertyIdentifier::RELIABILITY,
            PropertyIdentifier::HIGH_LIMIT,
            PropertyIdentifier::LOW_LIMIT,
            PropertyIdentifier::DEADBAND,
        ])
    }

    fn commit_event_transition_internal(
        &mut self,
        _commit: EventTransitionCommit,
    ) -> Result<(), EventTransitionCommitError> {
        self.commits.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }
}

fn timestamp_read(timestamp: BACnetTimeStamp) -> IndexedHistoryRead {
    let mut encoded = BytesMut::new();
    bacnet_encoding::primitives::encode_timestamp_choice(&mut encoded, &timestamp).unwrap();
    IndexedHistoryRead::Value(PropertyValue::ApplicationData(encoded.to_vec()))
}

fn empty_message_read() -> IndexedHistoryRead {
    IndexedHistoryRead::Value(PropertyValue::CharacterString(String::new()))
}

fn atomic_history_database(
    timestamps: [IndexedHistoryRead; 3],
    messages: [IndexedHistoryRead; 3],
    clocked: bool,
) -> (Arc<RwLock<ObjectDatabase>>, ObjectIdentifier) {
    let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 77).unwrap();
    let mut db = if clocked {
        clocked_test_database()
    } else {
        ObjectDatabase::new()
    };
    db.add(Box::new(AtomicHistoryObject {
        oid,
        event_type: Some(PropertyValue::Enumerated(EventType::OUT_OF_RANGE.to_raw())),
        timestamps,
        messages,
        status_flags: Some(PropertyValue::BitString {
            unused_bits: 4,
            data: vec![0],
        }),
        commits: Arc::new(AtomicUsize::new(0)),
    }))
    .unwrap();
    db.add(Box::new(
        DeviceObject::new(DeviceConfig {
            identity: bacnet_objects::DeviceIdentity::default().with_name("History Device"),
            ..DeviceConfig::default()
        })
        .unwrap(),
    ))
    .unwrap();
    db.add(Box::new(notification_class_0_broadcasting()))
        .unwrap();
    (Arc::new(RwLock::new(db)), oid)
}

async fn commit_and_capture_history_notification(
    timestamps: [IndexedHistoryRead; 3],
    messages: [IndexedHistoryRead; 3],
    change: EventStateChange,
    clocked: bool,
) -> (Arc<RwLock<ObjectDatabase>>, Vec<bytes::Bytes>) {
    let (db, oid) = atomic_history_database(timestamps, messages, clocked);
    let committed = {
        let mut guard = db.write().await;
        BACnetServer::<TestTransport>::commit_intrinsic_transition(
            &mut guard,
            &oid,
            TransitionOutcome {
                change,
                event_type: EventType::OUT_OF_RANGE,
                distribute: true,
            },
        )
    };

    let (transport, sent) = recording_transport();
    if let Some(committed) = committed {
        let network = Arc::new(NetworkLayer::new(transport));
        BACnetServer::<TestTransport>::build_and_send_event_notification_with_bindings(
            &crate::server::event_delivery::EventDelivery {
                db: &db,
                network: &network,
                comm_state: &Arc::new(CommState::default()),
                learned_routers: &Arc::new(Mutex::new(LearnedRouterCache::new())),
                notification_transactions: &NotificationTransactions::new(),
                device_bindings: &Arc::new(RwLock::new(
                    crate::server::device_bindings::DeviceBindingTable::new(),
                )),
                suppressions: &Default::default(),
                retry_timeout_ms: 1000,
                local_apdu_capacity: 1476,
            },
            &oid,
            committed,
        )
        .await;
    }

    (db, sent.npdus())
}

fn repeated_timestamp_reads(timestamp: BACnetTimeStamp) -> [IndexedHistoryRead; 3] {
    std::array::from_fn(|_| timestamp_read(timestamp.clone()))
}

fn empty_message_reads() -> [IndexedHistoryRead; 3] {
    std::array::from_fn(|_| empty_message_read())
}

fn committed_enrollment_normal(
    oid: ObjectIdentifier,
    change: EventStateChange,
) -> CommittedEventEnrollmentDelivery {
    CommittedEventEnrollmentDelivery {
        result: CommittedEventEnrollmentResult::Normal(EventEnrollmentTransition {
            enrollment_oid: oid,
            monitored_oid: oid,
            change,
            event_type: EventType::OUT_OF_RANGE,
            distribute: true,
        }),
        ack_required: true,
        recipient_clock: SampledEventClock::Unavailable,
        event_type: EventType::OUT_OF_RANGE,
        event_values: CommittedNotificationPayload::for_test(NotificationParameters::OutOfRange {
            exceeding_value: 1.0,
            status_flags: StatusFlags::empty(),
            deadband: 0.0,
            exceeded_limit: 1.0,
        }),
    }
}

fn committed_enrollment_reliability(oid: ObjectIdentifier) -> CommittedEventEnrollmentDelivery {
    CommittedEventEnrollmentDelivery {
        result: CommittedEventEnrollmentResult::Reliability(EventEnrollmentReliabilityResult {
            enrollment_oid: oid,
            monitored_oid: Some(oid),
            previous_reliability: bacnet_types::enums::Reliability::NO_FAULT_DETECTED,
            new_reliability: bacnet_types::enums::Reliability::OVER_RANGE,
            state_change: Some(EventStateChange {
                from: EventState::NORMAL,
                to: EventState::FAULT,
            }),
            distribute: true,
            cause: EventEnrollmentReliabilityCause::FaultAlgorithm,
        }),
        ack_required: true,
        recipient_clock: SampledEventClock::Unavailable,
        event_type: EventType::CHANGE_OF_RELIABILITY,
        event_values: CommittedNotificationPayload::for_test(
            NotificationParameters::ChangeOfReliability {
                reliability: bacnet_types::enums::Reliability::OVER_RANGE,
                status_flags: StatusFlags::empty(),
                property_values: Vec::new(),
            },
        ),
    }
}

/// An Event Enrollment's committed Event_Message_Texts entry is its
/// notification's message text: none for the empty entry it commits by
/// default, the configured text otherwise (#1507). An object that doesn't
/// serve the property sends no text, while a malformed entry or a refused
/// read suppresses the notification.
#[test]
fn event_enrollment_projection_reads_committed_message_without_mutation() {
    let configured = IndexedHistoryRead::Value(PropertyValue::CharacterString("high".into()));
    for (instance, (message, expected)) in [
        (empty_message_read(), Some(None)),
        (configured, Some(Some("high".to_owned()))),
        (IndexedHistoryRead::Missing, Some(None)),
        (IndexedHistoryRead::Value(PropertyValue::Unsigned(7)), None),
        (IndexedHistoryRead::Unreadable, None),
    ]
    .into_iter()
    .enumerate()
    {
        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 77 + instance as u32).unwrap();
        let mut db = ObjectDatabase::new();
        let mut messages = empty_message_reads();
        messages[0] = message;
        db.add(Box::new(AtomicHistoryObject {
            oid,
            event_type: Some(PropertyValue::Enumerated(EventType::OUT_OF_RANGE.to_raw())),
            timestamps: repeated_timestamp_reads(BACnetTimeStamp::SequenceNumber(19)),
            messages,
            status_flags: Some(PropertyValue::BitString {
                unused_bits: 4,
                data: vec![0],
            }),
            commits: Arc::new(AtomicUsize::new(0)),
        }))
        .unwrap();

        let before = db.reserve_event_sequence_number().number();
        let resolved =
            crate::server::event_notifications::resolve_committed_event_enrollment_transition(
                &db,
                committed_enrollment_normal(
                    oid,
                    EventStateChange {
                        from: EventState::NORMAL,
                        to: EventState::HIGH_LIMIT,
                    },
                ),
            );
        let message_text = resolved.map(|(_, _, transition)| match transition.history_source {
            crate::server::event_notifications::NotificationHistorySource::Committed {
                snapshot,
                ..
            } => snapshot.message_text,
            _ => panic!("an enrollment projects its committed history"),
        });
        assert_eq!(message_text, expected, "{instance}");
        assert_eq!(db.reserve_event_sequence_number().number(), before);
    }
}

#[test]
fn malformed_or_unreadable_event_enrollment_history_fails_closed_without_mutation() {
    let failures = [
        IndexedHistoryRead::Missing,
        IndexedHistoryRead::Unreadable,
        IndexedHistoryRead::Value(PropertyValue::Unsigned(7)),
        IndexedHistoryRead::Value(PropertyValue::ApplicationData(vec![0x19])),
    ];

    for (instance, failure) in failures.into_iter().enumerate() {
        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 80 + instance as u32).unwrap();
        let mut db = ObjectDatabase::new();
        let timestamps = std::array::from_fn(|index| {
            if index == 0 {
                failure.clone()
            } else {
                timestamp_read(BACnetTimeStamp::SequenceNumber(index as u16))
            }
        });
        db.add(Box::new(AtomicHistoryObject {
            oid,
            event_type: Some(PropertyValue::Enumerated(EventType::OUT_OF_RANGE.to_raw())),
            timestamps,
            messages: std::array::from_fn(|_| IndexedHistoryRead::Missing),
            status_flags: Some(PropertyValue::BitString {
                unused_bits: 4,
                data: vec![0],
            }),
            commits: Arc::new(AtomicUsize::new(0)),
        }))
        .unwrap();

        assert!(
            crate::server::event_notifications::resolve_committed_event_enrollment_transition(
                &db,
                committed_enrollment_normal(
                    oid,
                    EventStateChange {
                        from: EventState::NORMAL,
                        to: EventState::HIGH_LIMIT,
                    },
                ),
            )
            .is_none()
        );
        assert_eq!(db.reserve_event_sequence_number().number(), 0);
    }
}

#[test]
fn committed_reliability_effective_type_does_not_reread_configured_event_type() {
    for (instance, event_type) in [
        None,
        Some(PropertyValue::Unsigned(2)),
        Some(PropertyValue::Enumerated(999)),
    ]
    .into_iter()
    .enumerate()
    {
        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 90 + instance as u32).unwrap();
        let mut db = ObjectDatabase::new();
        db.add(Box::new(AtomicHistoryObject {
            oid,
            event_type,
            timestamps: repeated_timestamp_reads(BACnetTimeStamp::SequenceNumber(23)),
            messages: empty_message_reads(),
            status_flags: Some(PropertyValue::BitString {
                unused_bits: 4,
                data: vec![0],
            }),
            commits: Arc::new(AtomicUsize::new(0)),
        }))
        .unwrap();

        assert!(
            crate::server::event_notifications::resolve_committed_event_enrollment_transition(
                &db,
                committed_enrollment_reliability(oid),
            )
            .is_some()
        );
        assert_eq!(db.reserve_event_sequence_number().number(), 0);
    }
}

#[test]
fn malformed_or_missing_required_projection_commits_locally_but_cannot_emit() {
    for (instance, status_flags) in [Some(PropertyValue::Unsigned(0)), None]
        .into_iter()
        .enumerate()
    {
        let oid = ObjectIdentifier::new(ObjectType::ANALOG_INPUT, 99 + instance as u32).unwrap();
        let commits = Arc::new(AtomicUsize::new(0));
        let mut db = ObjectDatabase::new();
        db.add(Box::new(AtomicHistoryObject {
            oid,
            event_type: Some(PropertyValue::Enumerated(EventType::OUT_OF_RANGE.to_raw())),
            timestamps: repeated_timestamp_reads(BACnetTimeStamp::SequenceNumber(31)),
            messages: std::array::from_fn(|_| empty_message_read()),
            status_flags,
            commits: Arc::clone(&commits),
        }))
        .unwrap();

        let committed = BACnetServer::<TestTransport>::commit_intrinsic_transition(
            &mut db,
            &oid,
            TransitionOutcome {
                change: EventStateChange {
                    from: EventState::NORMAL,
                    to: EventState::HIGH_LIMIT,
                },
                event_type: EventType::OUT_OF_RANGE,
                distribute: true,
            },
        )
        .expect("the local atomic transition remains committed");
        assert_eq!(commits.load(Ordering::Relaxed), 1);
        assert!(
            committed.event_values.is_none(),
            "malformed or missing required values suppress the frame before encoding"
        );
    }
}

#[tokio::test]
async fn committed_history_preserves_each_timestamp_choice_on_the_wire() {
    let choices = [
        BACnetTimeStamp::Time(Time {
            hour: 4,
            minute: 5,
            second: 6,
            hundredths: 7,
        }),
        BACnetTimeStamp::SequenceNumber(44_444),
        BACnetTimeStamp::DateTime {
            date: Date {
                year: 124,
                month: 2,
                day: 29,
                day_of_week: 4,
            },
            time: Time {
                hour: 12,
                minute: 34,
                second: 56,
                hundredths: 78,
            },
        },
    ];

    for expected in choices {
        let (_, sent) = commit_and_capture_history_notification(
            repeated_timestamp_reads(expected.clone()),
            empty_message_reads(),
            EventStateChange {
                from: EventState::NORMAL,
                to: EventState::HIGH_LIMIT,
            },
            true,
        )
        .await;
        let notification = decode_broadcast_notification(&sent);
        assert_eq!(notification.timestamp, expected);
        assert_eq!(
            notification.message_text, None,
            "empty committed message history must remain absent on the wire"
        );
    }
}

#[tokio::test]
async fn committed_history_selects_only_the_transition_coordinate() {
    let timestamps = [
        timestamp_read(BACnetTimeStamp::SequenceNumber(11)),
        timestamp_read(BACnetTimeStamp::SequenceNumber(22)),
        timestamp_read(BACnetTimeStamp::SequenceNumber(33)),
    ];
    let cases = [
        (
            EventStateChange {
                from: EventState::NORMAL,
                to: EventState::HIGH_LIMIT,
            },
            11,
        ),
        (
            EventStateChange {
                from: EventState::NORMAL,
                to: EventState::FAULT,
            },
            22,
        ),
        (
            EventStateChange {
                from: EventState::HIGH_LIMIT,
                to: EventState::NORMAL,
            },
            33,
        ),
    ];

    for (change, expected) in cases {
        let (_, sent) = commit_and_capture_history_notification(
            timestamps.clone(),
            empty_message_reads(),
            change,
            true,
        )
        .await;
        assert_eq!(
            decode_broadcast_notification(&sent).timestamp,
            BACnetTimeStamp::SequenceNumber(expected)
        );
    }
}

#[tokio::test]
async fn committed_history_timestamp_is_distinct_from_the_staged_device_clock_sample() {
    let expected = BACnetTimeStamp::SequenceNumber(54_321);
    let (_, sent) = commit_and_capture_history_notification(
        repeated_timestamp_reads(expected.clone()),
        empty_message_reads(),
        EventStateChange {
            from: EventState::NORMAL,
            to: EventState::HIGH_LIMIT,
        },
        true,
    )
    .await;

    assert_eq!(
        decode_broadcast_notification(&sent).timestamp,
        expected,
        "the stored coordinate, not the staged Device DateTime, is wire authority"
    );
}

#[tokio::test]
async fn nonempty_committed_message_is_captured_with_its_timestamp() {
    let messages = std::array::from_fn(|index| {
        IndexedHistoryRead::Value(PropertyValue::CharacterString(format!("message-{index}")))
    });
    let (_, sent) = commit_and_capture_history_notification(
        repeated_timestamp_reads(BACnetTimeStamp::SequenceNumber(9)),
        messages,
        EventStateChange {
            from: EventState::NORMAL,
            to: EventState::FAULT,
        },
        true,
    )
    .await;

    assert_eq!(
        decode_broadcast_notification(&sent).message_text,
        Some("message-1".into())
    );
}

#[tokio::test]
async fn invalid_committed_history_suppresses_emission_after_consuming_sequence() {
    let mut trailing = BytesMut::new();
    bacnet_encoding::primitives::encode_timestamp_choice(
        &mut trailing,
        &BACnetTimeStamp::SequenceNumber(7),
    )
    .unwrap();
    trailing.extend_from_slice(&[0]);

    let failures = [
        IndexedHistoryRead::Value(PropertyValue::ApplicationData(vec![0x19])),
        IndexedHistoryRead::Value(PropertyValue::Unsigned(7)),
        IndexedHistoryRead::Missing,
        IndexedHistoryRead::Unreadable,
        IndexedHistoryRead::Value(PropertyValue::ApplicationData(trailing.to_vec())),
    ];

    for failure in failures {
        let timestamps = std::array::from_fn(|index| {
            if index == 0 {
                failure.clone()
            } else {
                timestamp_read(BACnetTimeStamp::SequenceNumber(index as u16))
            }
        });
        let (db, sent) = commit_and_capture_history_notification(
            timestamps,
            empty_message_reads(),
            EventStateChange {
                from: EventState::NORMAL,
                to: EventState::HIGH_LIMIT,
            },
            false,
        )
        .await;

        assert!(
            sent.is_empty(),
            "invalid committed history must fail closed"
        );
        assert_eq!(
            db.write().await.reserve_event_sequence_number().number(),
            1,
            "a successful commit permanently consumes its staged sequence"
        );
    }
}

#[tokio::test]
async fn invalid_committed_message_history_suppresses_emission_after_consuming_sequence() {
    let failures = [
        IndexedHistoryRead::Value(PropertyValue::Unsigned(7)),
        IndexedHistoryRead::Missing,
        IndexedHistoryRead::Unreadable,
    ];

    for failure in failures {
        let messages = std::array::from_fn(|index| {
            if index == 0 {
                failure.clone()
            } else {
                empty_message_read()
            }
        });
        let (db, sent) = commit_and_capture_history_notification(
            repeated_timestamp_reads(BACnetTimeStamp::SequenceNumber(17)),
            messages,
            EventStateChange {
                from: EventState::NORMAL,
                to: EventState::HIGH_LIMIT,
            },
            false,
        )
        .await;

        assert!(
            sent.is_empty(),
            "invalid committed message history must fail closed"
        );
        assert_eq!(
            db.write().await.reserve_event_sequence_number().number(),
            1,
            "a successful commit permanently consumes its staged sequence"
        );
    }
}