bacnet-objects 0.13.0

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use bacnet_types::constructed::{
    AuditPropertyReference, BACnetAuditLogDatum, BACnetAuditNotification, BACnetRecipient,
};
use bacnet_types::enums::{AuditOperation, ObjectType, PropertyIdentifier};
use bacnet_types::error::Error;
use bacnet_types::primitives::{BACnetTimeStamp, Date, ObjectIdentifier, Time};

use crate::clock::{ClockFrame, ClockReader};
use crate::traits::BACnetObject;

use super::{
    AuditLogNotificationSink, AuditLogObject, AuditLogPersistence, AuditLogSnapshot,
    CompletedAuditReceipt, ConfirmedAuditNotificationOutcome,
};

#[derive(Default)]
struct MemoryPersistence {
    snapshot: Mutex<Option<AuditLogSnapshot>>,
    commits: AtomicUsize,
    fail: AtomicBool,
}

impl AuditLogPersistence for MemoryPersistence {
    fn load(&self, _expected_object: ObjectIdentifier) -> Result<Option<AuditLogSnapshot>, Error> {
        Ok(self.snapshot.lock().unwrap().clone())
    }

    fn commit(&self, snapshot: &AuditLogSnapshot) -> Result<(), Error> {
        if self.fail.load(Ordering::Acquire) {
            return Err(Error::Transport(std::io::Error::other(
                "injected commit failure",
            )));
        }
        *self.snapshot.lock().unwrap() = Some(snapshot.clone());
        self.commits.fetch_add(1, Ordering::AcqRel);
        Ok(())
    }
}

struct FixedClock(Option<ClockFrame>);

impl ClockReader for FixedClock {
    fn read_clock(&self) -> Option<ClockFrame> {
        self.0
    }
}

fn oid(object_type: ObjectType, instance: u32) -> ObjectIdentifier {
    ObjectIdentifier::new(object_type, instance).unwrap()
}

fn date() -> Date {
    Date {
        year: 124,
        month: 2,
        day: 29,
        day_of_week: 4,
    }
}

fn time(second: u8) -> Time {
    Time {
        hour: 12,
        minute: 0,
        second,
        hundredths: 0,
    }
}

fn clock() -> Arc<dyn ClockReader> {
    Arc::new(FixedClock(Some(ClockFrame {
        local_date: date(),
        local_time: time(30),
        utc_offset: 0,
        daylight_savings_status: false,
    })))
}

fn invalid_clock_frame() -> ClockFrame {
    ClockFrame {
        local_date: Date {
            year: 124,
            month: 2,
            day: 30,
            day_of_week: 5,
        },
        local_time: time(30),
        utc_offset: 0,
        daylight_savings_status: false,
    }
}

fn notification(
    source: Option<BACnetTimeStamp>,
    target: Option<BACnetTimeStamp>,
) -> BACnetAuditNotification {
    BACnetAuditNotification {
        source_timestamp: source,
        target_timestamp: target,
        source_device: BACnetRecipient::Device(oid(ObjectType::DEVICE, 1)),
        source_object: Some(oid(ObjectType::ANALOG_INPUT, 1)),
        operation: AuditOperation::WRITE,
        source_comment: None,
        target_comment: None,
        invoke_id: Some(7),
        source_user_id: Some(9),
        source_user_role: Some(2),
        target_device: BACnetRecipient::Device(oid(ObjectType::DEVICE, 2)),
        target_object: Some(oid(ObjectType::ANALOG_VALUE, 3)),
        target_property: Some(AuditPropertyReference {
            property_identifier: PropertyIdentifier::PRESENT_VALUE,
            property_array_index: None,
        }),
        target_priority: Some(8),
        target_value: Some(vec![0x21, 0x05]),
        current_value: None,
        result: None,
    }
}

fn log(capacity: u32) -> (AuditLogObject, Arc<MemoryPersistence>) {
    let persistence = Arc::new(MemoryPersistence::default());
    let mut log = AuditLogObject::new(1, "audit", capacity, persistence.clone()).unwrap();
    log.bind_clock_internal(Some(clock()));
    (log, persistence)
}

fn receipt(key: &[u8], completed_at: u64) -> CompletedAuditReceipt {
    CompletedAuditReceipt::new(key.to_vec(), completed_at).unwrap()
}

#[test]
fn forwarding_change_outcome_requires_record_change_and_durable_commit() {
    let (mut log, persistence) = log(10);
    let notification = notification(
        Some(BACnetTimeStamp::Time(time(0))),
        Some(BACnetTimeStamp::Time(time(0))),
    );
    assert!(log
        .store_notifications_with_change(std::slice::from_ref(&notification), 2_000)
        .unwrap());
    assert!(!log
        .store_notifications_with_change(std::slice::from_ref(&notification), 2_000)
        .unwrap());
    assert_eq!(
        log.store_confirmed_notifications_with_change(
            std::slice::from_ref(&notification),
            2_000,
            receipt(b"first", 1),
        )
        .unwrap(),
        (ConfirmedAuditNotificationOutcome::Stored, false),
    );
    assert_eq!(
        log.store_confirmed_notifications_with_change(
            std::slice::from_ref(&notification),
            2_000,
            receipt(b"first", 2),
        )
        .unwrap(),
        (ConfirmedAuditNotificationOutcome::Duplicate, false),
    );
    let before = log.current_snapshot();
    let mut different = notification.clone();
    different.invoke_id = Some(8);
    persistence.fail.store(true, Ordering::Release);
    assert!(log
        .store_confirmed_notifications_with_change(&[different], 2_000, receipt(b"second", 3),)
        .is_err());
    assert_eq!(log.current_snapshot(), before);
}

#[test]
fn complementary_batch_merges_once_and_target_current_value_wins() {
    let (mut log, persistence) = log(10);
    let mut source = notification(Some(BACnetTimeStamp::Time(time(0))), None);
    source.current_value = Some(vec![0x21, 0x01]);
    let mut target = notification(None, Some(BACnetTimeStamp::Time(time(4))));
    target.current_value = Some(vec![0x21, 0x02]);

    log.store_notifications(&[source, target], 2_000).unwrap();

    assert_eq!(persistence.commits.load(Ordering::Acquire), 2); // initialization + batch
    assert_eq!(log.total_record_count(), 1);
    assert_eq!(log.records().len(), 1);
    let result = log.records().front().unwrap();
    assert_eq!(result.sequence_number, 1);
    assert_eq!(result.record.timestamp, (date(), time(30)));
    let BACnetAuditLogDatum::AuditNotification(stored) = &result.record.datum else {
        panic!("expected notification")
    };
    assert!(stored.source_timestamp.is_some());
    assert!(stored.target_timestamp.is_some());
    assert_eq!(stored.current_value, Some(vec![0x21, 0x02]));
}

#[test]
fn completed_match_is_dropped_and_merge_preserves_record_identity() {
    let (mut log, persistence) = log(10);
    let source = notification(Some(BACnetTimeStamp::Time(time(0))), None);
    log.store_notifications(&[source], 2_000).unwrap();
    let identity = log.records().front().unwrap().clone();
    let target = notification(None, Some(BACnetTimeStamp::Time(time(3))));
    log.store_notifications(std::slice::from_ref(&target), 2_000)
        .unwrap();
    assert_eq!(log.total_record_count(), 1);
    assert_eq!(
        log.records().front().unwrap().sequence_number,
        identity.sequence_number
    );
    assert_eq!(
        log.records().front().unwrap().record.timestamp,
        identity.record.timestamp
    );
    let generation = log.generation();
    let commits = persistence.commits.load(Ordering::Acquire);

    log.store_notifications(&[target], 2_000).unwrap();
    assert_eq!(log.generation(), generation);
    assert_eq!(persistence.commits.load(Ordering::Acquire), commits);
    assert_eq!(log.total_record_count(), 1);
}

#[test]
fn confirmed_completed_match_still_commits_receipt_without_record_change() {
    let (mut log, persistence) = log(10);
    let source = notification(Some(BACnetTimeStamp::Time(time(0))), None);
    let target = notification(None, Some(BACnetTimeStamp::Time(time(3))));
    log.store_notifications(&[source, target.clone()], 2_000)
        .unwrap();
    let generation = log.generation();
    let record = log.records().front().unwrap().clone();

    assert_eq!(
        log.store_confirmed_notifications(&[target], 2_000, receipt(b"complete", 100))
            .unwrap(),
        ConfirmedAuditNotificationOutcome::Stored
    );
    assert_eq!(log.generation(), generation + 1);
    assert_eq!(log.total_record_count(), 1);
    assert_eq!(log.records().front(), Some(&record));
    assert_eq!(
        persistence
            .snapshot
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .completed_receipts
            .len(),
        1
    );
}

#[test]
fn confirmed_commit_failure_is_atomic_and_exact_retry_can_succeed() {
    let (mut log, persistence) = log(10);
    let incoming = notification(Some(BACnetTimeStamp::Time(time(0))), None);
    let completed = receipt(b"retry", 100);
    let before = persistence.snapshot.lock().unwrap().clone().unwrap();
    persistence.fail.store(true, Ordering::Release);

    assert!(log
        .store_confirmed_notifications(std::slice::from_ref(&incoming), 2_000, completed.clone(),)
        .is_err());
    assert_eq!(log.generation(), before.generation);
    assert!(log.records().is_empty());
    assert_eq!(persistence.snapshot.lock().unwrap().as_ref(), Some(&before));

    persistence.fail.store(false, Ordering::Release);
    assert_eq!(
        log.store_confirmed_notifications(&[incoming], 2_000, completed.clone())
            .unwrap(),
        ConfirmedAuditNotificationOutcome::Stored
    );
    let accepted = persistence.snapshot.lock().unwrap().clone().unwrap();
    assert_eq!(accepted.completed_receipts, vec![completed.clone()]);
    assert_eq!(
        log.store_confirmed_notifications(
            &[notification(None, Some(BACnetTimeStamp::Time(time(1))))],
            2_000,
            completed,
        )
        .unwrap(),
        ConfirmedAuditNotificationOutcome::Duplicate
    );
    assert_eq!(
        persistence.snapshot.lock().unwrap().as_ref(),
        Some(&accepted)
    );
}

#[test]
fn match_identity_excludes_the_separate_source_and_target_object_parameters() {
    let (mut log, _) = log(10);
    let source = notification(Some(BACnetTimeStamp::Time(time(0))), None);
    let source_object = source.source_object;
    let target_object = source.target_object;
    let mut target = notification(None, Some(BACnetTimeStamp::Time(time(2))));
    target.source_object = Some(oid(ObjectType::ANALOG_INPUT, 99));
    target.target_object = Some(oid(ObjectType::ANALOG_VALUE, 99));

    log.store_notifications(&[source, target], 2_000).unwrap();

    assert_eq!(log.total_record_count(), 1);
    let BACnetAuditLogDatum::AuditNotification(stored) =
        &log.records().front().unwrap().record.datum
    else {
        panic!("expected notification")
    };
    assert_eq!(stored.source_object, source_object);
    assert_eq!(stored.target_object, target_object);
    assert!(stored.source_timestamp.is_some());
    assert!(stored.target_timestamp.is_some());
}

#[test]
fn timestamp_variants_use_configured_window_and_never_cross_compare() {
    let cases = [
        (
            BACnetTimeStamp::DateTime {
                date: date(),
                time: time(0),
            },
            BACnetTimeStamp::DateTime {
                date: date(),
                time: time(5),
            },
            2_500,
            1,
        ),
        (
            BACnetTimeStamp::SequenceNumber(7),
            BACnetTimeStamp::SequenceNumber(7),
            1,
            1,
        ),
        (
            BACnetTimeStamp::SequenceNumber(7),
            BACnetTimeStamp::SequenceNumber(8),
            60_000,
            2,
        ),
        (
            BACnetTimeStamp::Time(time(0)),
            BACnetTimeStamp::SequenceNumber(0),
            60_000,
            2,
        ),
    ];
    for (source_timestamp, target_timestamp, timeout, expected_count) in cases {
        let (mut log, _) = log(10);
        log.store_notifications(
            &[
                notification(Some(source_timestamp), None),
                notification(None, Some(target_timestamp)),
            ],
            timeout,
        )
        .unwrap();
        assert_eq!(log.total_record_count(), expected_count);
    }
}

#[test]
fn forwarding_window_is_inclusive_at_wire_tick_and_timeout_edges() {
    for datetime in [false, true] {
        // Time/DateTime encode hundredths, not milliseconds. Check +/- one
        // wire tick, then +/- one millisecond of APDU_Timeout (a 2 ms window step).
        for (distance, timeout, complements) in [
            (399, 2_000, true),
            (400, 2_000, true),
            (401, 2_000, false),
            (400, 1_999, false),
            (400, 2_001, true),
        ] {
            let stamp = |ticks: u16| {
                let time = Time {
                    second: (ticks / 100) as u8,
                    hundredths: (ticks % 100) as u8,
                    ..time(0)
                };
                if datetime {
                    BACnetTimeStamp::DateTime { date: date(), time }
                } else {
                    BACnetTimeStamp::Time(time)
                }
            };
            let (mut log, persistence) = log(2);
            assert!(log
                .store_notifications_with_change(&[notification(Some(stamp(0)), None)], timeout)
                .unwrap());
            let before = log.records().front().unwrap().clone();
            let target = notification(None, Some(stamp(distance)));
            assert!(log
                .store_notifications_with_change(std::slice::from_ref(&target), timeout)
                .unwrap());
            assert_eq!(log.records().len(), if complements { 1 } else { 2 });
            assert_eq!(log.total_record_count(), if complements { 1 } else { 2 });
            assert_eq!(log.records()[0].sequence_number, before.sequence_number);
            assert_eq!(log.records()[0].record.timestamp, before.record.timestamp);
            assert_eq!(persistence.commits.load(Ordering::Acquire), 3);
            let BACnetAuditLogDatum::AuditNotification(stored) = &log.records()[0].record.datum
            else {
                panic!("expected notification")
            };
            assert_eq!(stored.target_timestamp.is_some(), complements);
            if complements {
                let before = log.current_snapshot();
                assert!(!log
                    .store_notifications_with_change(&[target], timeout)
                    .unwrap());
                assert_eq!(log.current_snapshot(), before);
                assert_eq!(persistence.commits.load(Ordering::Acquire), 3);
            }
        }
    }
}

#[test]
fn forwarding_mixed_timestamp_variants_never_complement_in_either_direction() {
    let variants = [
        BACnetTimeStamp::Time(time(0)),
        BACnetTimeStamp::DateTime {
            date: date(),
            time: time(0),
        },
        BACnetTimeStamp::SequenceNumber(0),
    ];
    for (i, source) in variants.iter().enumerate() {
        for (j, target) in variants.iter().enumerate() {
            if i == j {
                continue;
            }
            let (mut log, persistence) = log(2);
            assert!(log
                .store_notifications_with_change(
                    &[notification(Some(source.clone()), None)],
                    60_000
                )
                .unwrap());
            let first = log.records()[0].clone();
            assert!(log
                .store_notifications_with_change(
                    &[notification(None, Some(target.clone()))],
                    60_000
                )
                .unwrap());
            assert_eq!(log.total_record_count(), 2);
            assert_eq!(log.records().len(), 2);
            assert_eq!(log.records()[0], first);
            assert_eq!(log.records()[1].sequence_number, 2);
            assert_eq!(persistence.commits.load(Ordering::Acquire), 3);
        }
    }
}

#[test]
fn commit_and_clock_failures_leave_memory_and_durable_snapshot_unchanged() {
    let (mut log, persistence) = log(10);
    let before = persistence.snapshot.lock().unwrap().clone().unwrap();
    persistence.fail.store(true, Ordering::Release);
    assert!(log
        .store_notifications(
            &[notification(Some(BACnetTimeStamp::Time(time(0))), None)],
            2_000
        )
        .is_err());
    assert_eq!(log.total_record_count(), 0);
    assert_eq!(*persistence.snapshot.lock().unwrap(), Some(before));

    persistence.fail.store(false, Ordering::Release);
    for frame in [None, Some(invalid_clock_frame())] {
        log.bind_clock_internal(Some(Arc::new(FixedClock(frame))));
        let durable = persistence.snapshot.lock().unwrap().clone();
        let error = log
            .store_notifications(
                &[notification(Some(BACnetTimeStamp::Time(time(0))), None)],
                2_000,
            )
            .unwrap_err();
        assert!(matches!(error, Error::Protocol { class, code }
            if class == bacnet_types::enums::ErrorClass::DEVICE.to_raw() as u32
                && code == bacnet_types::enums::ErrorCode::OPERATIONAL_PROBLEM.to_raw() as u32));
        assert_eq!(log.total_record_count(), 0);
        assert_eq!(*persistence.snapshot.lock().unwrap(), durable);
    }
}

#[test]
fn disabled_sink_denies_before_clock_or_persistence_change() {
    let (mut log, persistence) = log(10);
    log.write_property(
        bacnet_types::enums::PropertyIdentifier::LOG_ENABLE,
        None,
        bacnet_types::primitives::PropertyValue::Boolean(false),
        None,
    )
    .unwrap();
    let generation = log.generation();
    let error = log
        .store_notifications(&[notification(None, None)], 2_000)
        .unwrap_err();
    assert!(matches!(error, Error::Protocol { class, code }
        if class == bacnet_types::enums::ErrorClass::SERVICES.to_raw() as u32
            && code == bacnet_types::enums::ErrorCode::SERVICE_REQUEST_DENIED.to_raw() as u32));
    assert_eq!(log.generation(), generation);
    assert_eq!(
        persistence
            .snapshot
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .generation,
        generation
    );
}