soaprs-memory 0.2.0

//! In-process typed event delivery.

use std::{
    fmt,
    sync::{
        Arc, RwLock,
        atomic::{AtomicU64, Ordering},
    },
};

use soaprs_core::{BoxFuture, SoapError, SoapResult};
use soaprs_events::{Event, EventEnvelope, EventHandler, EventPublisher};

/// Opaque identity of an in-memory event subscription.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct SubscriptionId(u64);

impl SubscriptionId {
    /// Returns the numeric reference-adapter identity.
    pub const fn get(self) -> u64 {
        self.0
    }
}

struct Subscription<E>
where
    E: Event,
{
    id: SubscriptionId,
    handler: Arc<dyn EventHandler<E>>,
}

/// Subscription-order event bus for tests and single-process applications.
///
/// Publishing stops and returns the first handler error. It never silently
/// logs and discards a failed handler invocation.
pub struct MemoryEventBus<E>
where
    E: Event,
{
    next_id: AtomicU64,
    subscriptions: RwLock<Vec<Subscription<E>>>,
}

impl<E> MemoryEventBus<E>
where
    E: Event,
{
    /// Creates an empty event bus.
    pub const fn new() -> Self {
        Self {
            next_id: AtomicU64::new(1),
            subscriptions: RwLock::new(Vec::new()),
        }
    }

    /// Adds a handler.
    pub fn subscribe(&self, handler: Arc<dyn EventHandler<E>>) -> SoapResult<SubscriptionId> {
        let id = SubscriptionId(self.next_id.fetch_add(1, Ordering::Relaxed));
        self.subscriptions
            .write()
            .map_err(|_| SoapError::infrastructure("in-memory event bus write lock poisoned"))?
            .push(Subscription { id, handler });
        Ok(id)
    }

    /// Removes a handler and reports whether it existed.
    pub fn unsubscribe(&self, id: SubscriptionId) -> SoapResult<bool> {
        let mut subscriptions = self
            .subscriptions
            .write()
            .map_err(|_| SoapError::infrastructure("in-memory event bus write lock poisoned"))?;
        let Some(index) = subscriptions.iter().position(|entry| entry.id == id) else {
            return Ok(false);
        };
        subscriptions.remove(index);
        Ok(true)
    }

    /// Returns the number of current handlers.
    pub fn subscriber_count(&self) -> SoapResult<usize> {
        Ok(self
            .subscriptions
            .read()
            .map_err(|_| SoapError::infrastructure("in-memory event bus read lock poisoned"))?
            .len())
    }
}

impl<E> Default for MemoryEventBus<E>
where
    E: Event,
{
    fn default() -> Self {
        Self::new()
    }
}

impl<E> fmt::Debug for MemoryEventBus<E>
where
    E: Event,
{
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("MemoryEventBus")
            .field("event_family", &std::any::type_name::<E>())
            .finish_non_exhaustive()
    }
}

impl<E> EventPublisher<E> for MemoryEventBus<E>
where
    E: Event,
{
    fn publish(&self, event: EventEnvelope<E>) -> BoxFuture<'_, SoapResult<()>> {
        let handlers = match self.subscriptions.read() {
            Ok(subscriptions) => subscriptions
                .iter()
                .map(|entry| Arc::clone(&entry.handler))
                .collect::<Vec<_>>(),
            Err(_) => {
                return Box::pin(async {
                    Err(SoapError::infrastructure(
                        "in-memory event bus read lock poisoned",
                    ))
                });
            }
        };

        Box::pin(async move {
            for handler in handlers {
                handler.handle(&event).await?;
            }
            Ok(())
        })
    }
}

#[cfg(test)]
mod tests {
    use std::{
        sync::{Arc, Mutex},
        time::UNIX_EPOCH,
    };

    use soaprs_contract_tests::block_on;
    use soaprs_core::{BoxFuture, MessageMetadata, SoapError, SoapResult};
    use soaprs_events::{DomainEvent, Event, EventEnvelope, EventHandler, EventPublisher};

    use super::MemoryEventBus;

    #[derive(Debug)]
    struct TestEvent(u8);

    impl Event for TestEvent {
        fn event_type(&self) -> &'static str {
            "contract.bus-event"
        }
    }

    impl DomainEvent for TestEvent {}

    struct RecordingHandler {
        seen: Arc<Mutex<Vec<u8>>>,
        fails: bool,
    }

    impl EventHandler<TestEvent> for RecordingHandler {
        fn handle<'a>(
            &'a self,
            event: &'a EventEnvelope<TestEvent>,
        ) -> BoxFuture<'a, SoapResult<()>> {
            Box::pin(async move {
                self.seen
                    .lock()
                    .map_err(|_| SoapError::infrastructure("test event lock poisoned"))?
                    .push(event.message.0);
                if self.fails {
                    Err(SoapError::domain("event handler failed"))
                } else {
                    Ok(())
                }
            })
        }
    }

    #[test]
    fn bus_preserves_order_and_surfaces_handler_failure() {
        let bus = MemoryEventBus::new();
        let seen = Arc::new(Mutex::new(Vec::new()));
        for fails in [false, true, false] {
            let subscribed = bus.subscribe(Arc::new(RecordingHandler {
                seen: Arc::clone(&seen),
                fails,
            }));
            assert!(subscribed.is_ok(), "{subscribed:?}");
        }

        let result = block_on(bus.publish(EventEnvelope::new(
            TestEvent(7),
            MessageMetadata::new("event-1", UNIX_EPOCH),
        )));
        assert!(result.is_err());
        assert_eq!(
            seen.lock().ok().map(|values| values.clone()),
            Some(vec![7, 7])
        );
    }
}