eventuary-core 0.1.0

//! Multiplexer handler: fans out one delivered event to multiple
//! filter+handler subscribers under a single ack/nack lifecycle.
//!
//! Each route has a stable `SubscriberId` used as the storage key for
//! the optional `MultiplexerStore`. When configured, the store skips
//! subscribers that already completed for a given event on retry, so
//! redelivery from the backend does not re-run already-acked work.
//!
//! Without a store (the default `NoMultiplexerStore`), every matching
//! subscriber runs on every delivery — at-least-once at the
//! subscriber level matches the backend's at-least-once semantics.
//!
//! Subscribers run concurrently up to `concurrency` via a
//! `FuturesUnordered` drain loop. The handle aggregates every
//! subscriber failure into a single `Error::Handler`; no fail-fast
//! shortcut — slow successors still run so the handler reports the
//! full failure set.
//!
//! Failure modes affecting delivery semantics:
//! - Handler succeeds but `mark_completed` fails: at-least-once
//!   double-run on the next redelivery. Callers should accept idempotent
//!   handlers or pair this with the dedupe wrapper.
//! - Empty match set: governed by `NoMatchPolicy` (default `Ack`).

use std::collections::HashSet;
use std::fmt;
use std::future::Future;
use std::num::NonZeroUsize;
use std::sync::Arc;

use futures::StreamExt;
use futures::stream::FuturesUnordered;

use crate::error::{Error, Result};
use crate::event::{Event, EventId};
use crate::io::{ArcFilter, ArcHandler, Filter, Handler, HandlerExt};
use crate::payload::Payload;

#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct SubscriberId(Arc<str>);

impl SubscriberId {
    pub fn new(value: impl AsRef<str>) -> Result<Self> {
        let value = value.as_ref();
        if value.is_empty() || value.len() > 128 {
            return Err(Error::Config(format!(
                "invalid subscriber id length: {value:?}"
            )));
        }
        if !value
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-' || c == '.' || c == ':')
        {
            return Err(Error::Config(format!("invalid subscriber id: {value:?}")));
        }
        Ok(Self(Arc::from(value)))
    }

    /// Build a subscriber id by joining `prefix` and `id` with a `:`
    /// separator. Useful for namespacing subscriber ids when multiple
    /// multiplexers share one `MultiplexerStore` — a prefix per
    /// multiplexer keeps `(event_id, subscriber_id)` collisions
    /// impossible across multiplexers.
    pub fn with_prefix(prefix: impl AsRef<str>, id: impl AsRef<str>) -> Result<Self> {
        Self::new(format!("{}:{}", prefix.as_ref(), id.as_ref()))
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl fmt::Display for SubscriberId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct MultiplexerKey {
    pub event_id: EventId,
    pub subscriber_id: SubscriberId,
}

impl MultiplexerKey {
    pub fn new(event_id: EventId, subscriber_id: SubscriberId) -> Self {
        Self {
            event_id,
            subscriber_id,
        }
    }
}

pub trait MultiplexerStore: Clone + Send + Sync + 'static {
    fn is_completed<'a>(
        &'a self,
        key: &'a MultiplexerKey,
    ) -> impl Future<Output = Result<bool>> + Send + 'a;

    fn mark_completed<'a>(
        &'a self,
        key: &'a MultiplexerKey,
    ) -> impl Future<Output = Result<()>> + Send + 'a;
}

#[derive(Debug, Clone, Copy, Default)]
pub struct NoMultiplexerStore;

impl MultiplexerStore for NoMultiplexerStore {
    async fn is_completed(&self, _: &MultiplexerKey) -> Result<bool> {
        Ok(false)
    }

    async fn mark_completed(&self, _: &MultiplexerKey) -> Result<()> {
        Ok(())
    }
}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Default)]
pub enum NoMatchPolicy {
    #[default]
    Ack,
    Error,
}

struct MultiplexerRoute<P = Payload> {
    subscriber_id: SubscriberId,
    filter: ArcFilter<P>,
    handler: ArcHandler<P>,
}

impl<P> MultiplexerRoute<P>
where
    P: Send + Sync + 'static,
{
    fn new<F, H>(subscriber_id: impl AsRef<str>, filter: F, handler: H) -> Result<Self>
    where
        F: Filter<P> + 'static,
        H: Handler<P> + 'static,
    {
        Ok(Self {
            subscriber_id: SubscriberId::new(subscriber_id)?,
            filter: Arc::new(filter),
            handler: handler.into_arced(),
        })
    }
}

#[derive(Debug)]
struct MultiplexerFailure {
    subscriber_id: SubscriberId,
    error: Error,
}

const DEFAULT_CONCURRENCY: NonZeroUsize = NonZeroUsize::MIN;

pub struct MultiplexerBuilder<S = NoMultiplexerStore, P = Payload> {
    id: String,
    routes: Vec<MultiplexerRoute<P>>,
    subscriber_ids: HashSet<SubscriberId>,
    store: S,
    concurrency: NonZeroUsize,
    no_match: NoMatchPolicy,
}

impl<P> Default for MultiplexerBuilder<NoMultiplexerStore, P> {
    fn default() -> Self {
        Self {
            id: "multiplexer".to_owned(),
            routes: Vec::new(),
            subscriber_ids: HashSet::new(),
            store: NoMultiplexerStore,
            concurrency: DEFAULT_CONCURRENCY,
            no_match: NoMatchPolicy::Ack,
        }
    }
}

impl<S, P> MultiplexerBuilder<S, P>
where
    P: Send + Sync + 'static,
{
    pub fn id(mut self, id: impl Into<String>) -> Self {
        self.id = id.into();
        self
    }

    pub fn route<F, H>(
        mut self,
        subscriber_id: impl AsRef<str>,
        filter: F,
        handler: H,
    ) -> Result<Self>
    where
        F: Filter<P> + 'static,
        H: Handler<P> + 'static,
    {
        let route = MultiplexerRoute::new(subscriber_id, filter, handler)?;
        if !self.subscriber_ids.insert(route.subscriber_id.clone()) {
            return Err(Error::Config(format!(
                "multiplexer {}: duplicate subscriber id {}",
                self.id, route.subscriber_id
            )));
        }
        self.routes.push(route);
        Ok(self)
    }

    pub fn concurrency(mut self, concurrency: NonZeroUsize) -> Self {
        self.concurrency = concurrency;
        self
    }

    pub fn on_no_match(mut self, policy: NoMatchPolicy) -> Self {
        self.no_match = policy;
        self
    }

    pub fn store<S2: MultiplexerStore>(self, store: S2) -> MultiplexerBuilder<S2, P> {
        MultiplexerBuilder {
            id: self.id,
            routes: self.routes,
            subscriber_ids: self.subscriber_ids,
            store,
            concurrency: self.concurrency,
            no_match: self.no_match,
        }
    }
}

impl<S, P> MultiplexerBuilder<S, P>
where
    S: MultiplexerStore,
    P: Send + Sync + 'static,
{
    pub fn build(self) -> Result<Multiplexer<S, P>> {
        if self.id.trim().is_empty() {
            return Err(Error::Config("multiplexer id must not be empty".to_owned()));
        }
        Ok(Multiplexer {
            id: self.id,
            routes: self.routes,
            store: self.store,
            concurrency: self.concurrency,
            no_match: self.no_match,
        })
    }
}

pub struct Multiplexer<S = NoMultiplexerStore, P = Payload> {
    id: String,
    routes: Vec<MultiplexerRoute<P>>,
    store: S,
    concurrency: NonZeroUsize,
    no_match: NoMatchPolicy,
}

impl<P> Multiplexer<NoMultiplexerStore, P> {
    pub fn builder() -> MultiplexerBuilder<NoMultiplexerStore, P> {
        MultiplexerBuilder::default()
    }
}

impl<S, P> Multiplexer<S, P>
where
    S: MultiplexerStore,
    P: Send + Sync + 'static,
{
    async fn run_subscriber(
        &self,
        subscriber_id: SubscriberId,
        handler: ArcHandler<P>,
        event: &Event<P>,
    ) -> std::result::Result<(), MultiplexerFailure> {
        let key = MultiplexerKey::new(event.id(), subscriber_id.clone());
        match self.store.is_completed(&key).await {
            Ok(true) => return Ok(()),
            Ok(false) => {}
            Err(error) => {
                return Err(MultiplexerFailure {
                    subscriber_id,
                    error,
                });
            }
        }

        if let Err(error) = handler.handle(event).await {
            return Err(MultiplexerFailure {
                subscriber_id,
                error,
            });
        }

        if let Err(error) = self.store.mark_completed(&key).await {
            return Err(MultiplexerFailure {
                subscriber_id,
                error,
            });
        }

        Ok(())
    }
}

impl<S, P> Handler<P> for Multiplexer<S, P>
where
    S: MultiplexerStore,
    P: Send + Sync + 'static,
{
    fn id(&self) -> &str {
        &self.id
    }

    async fn handle(&self, event: &Event<P>) -> Result<()> {
        let matching: Vec<(SubscriberId, ArcHandler<P>)> = self
            .routes
            .iter()
            .filter(|route| route.filter.matches(event))
            .map(|route| (route.subscriber_id.clone(), Arc::clone(&route.handler)))
            .collect();

        if matching.is_empty() {
            return match self.no_match {
                NoMatchPolicy::Ack => Ok(()),
                NoMatchPolicy::Error => Err(Error::Handler(format!(
                    "multiplexer {} found no matching subscriber for event {}",
                    self.id,
                    event.id()
                ))),
            };
        }

        let concurrency = self.concurrency.get();
        let mut in_flight: FuturesUnordered<_> = FuturesUnordered::new();
        let mut iter = matching.into_iter();
        let mut failures: Vec<MultiplexerFailure> = Vec::new();

        for _ in 0..concurrency {
            if let Some((subscriber_id, handler)) = iter.next() {
                in_flight.push(self.run_subscriber(subscriber_id, handler, event));
            } else {
                break;
            }
        }

        while let Some(result) = in_flight.next().await {
            if let Err(failure) = result {
                failures.push(failure);
            }
            if let Some((subscriber_id, handler)) = iter.next() {
                in_flight.push(self.run_subscriber(subscriber_id, handler, event));
            }
        }

        if failures.is_empty() {
            return Ok(());
        }

        let summary = failures
            .iter()
            .map(|f| format!("{}: {}", f.subscriber_id, f.error))
            .collect::<Vec<_>>()
            .join(", ");
        Err(Error::Handler(format!(
            "multiplexer {} failed subscribers: {summary}",
            self.id
        )))
    }
}

#[cfg(test)]
mod tests {
    use std::num::NonZeroUsize;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;
    use crate::event::{Event, EventId};
    use crate::io::filter::EventFilter;
    use crate::io::{Filter, Handler};
    use crate::payload::Payload;

    #[test]
    fn subscriber_id_accepts_stable_names() {
        let id = SubscriberId::new("orders-projection.v1:blue").unwrap();
        assert_eq!(id.as_str(), "orders-projection.v1:blue");
    }

    #[test]
    fn subscriber_id_rejects_invalid_values() {
        assert!(SubscriberId::new("").is_err());
        assert!(SubscriberId::new("bad space").is_err());
        assert!(SubscriberId::new("bad/slash").is_err());
        assert!(SubscriberId::new("a".repeat(129)).is_err());
    }

    #[test]
    fn multiplexer_key_exposes_components() {
        let event_id = EventId::new();
        let subscriber_id = SubscriberId::new("audit").unwrap();
        let key = MultiplexerKey::new(event_id, subscriber_id.clone());
        assert_eq!(key.event_id, event_id);
        assert_eq!(key.subscriber_id, subscriber_id);
    }

    #[tokio::test]
    async fn no_multiplexer_store_reports_nothing_completed() {
        let store = NoMultiplexerStore;
        let key = MultiplexerKey::new(EventId::new(), SubscriberId::new("audit").unwrap());
        assert!(!store.is_completed(&key).await.unwrap());
        store.mark_completed(&key).await.unwrap();
        assert!(!store.is_completed(&key).await.unwrap());
    }

    #[derive(Clone, Default)]
    struct TestMultiplexerStore {
        completed: Arc<std::sync::Mutex<HashSet<MultiplexerKey>>>,
    }

    impl MultiplexerStore for TestMultiplexerStore {
        async fn is_completed(&self, key: &MultiplexerKey) -> Result<bool> {
            Ok(self.completed.lock().unwrap().contains(key))
        }

        async fn mark_completed(&self, key: &MultiplexerKey) -> Result<()> {
            self.completed.lock().unwrap().insert(key.clone());
            Ok(())
        }
    }

    struct CountingHandler {
        id: String,
        count: Arc<AtomicUsize>,
    }

    impl Handler for CountingHandler {
        fn id(&self) -> &str {
            &self.id
        }

        async fn handle(&self, _: &Event) -> Result<()> {
            self.count.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }

    struct FailingHandler;

    impl Handler for FailingHandler {
        fn id(&self) -> &str {
            "failing"
        }

        async fn handle(&self, _: &Event) -> Result<()> {
            Err(Error::Store("handler failed".to_owned()))
        }
    }

    struct AllowNothing;

    impl Filter for AllowNothing {
        fn matches(&self, _: &Event) -> bool {
            false
        }
    }

    fn event() -> Event {
        Event::builder(
            "acme",
            "/orders",
            "order.created",
            "order-1",
            Payload::from_string("p"),
        )
        .unwrap()
        .build()
        .unwrap()
    }

    #[tokio::test]
    async fn multiplexer_invokes_every_matching_handler() {
        let first = Arc::new(AtomicUsize::new(0));
        let second = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "first",
                EventFilter::default(),
                CountingHandler {
                    id: "first-handler".to_owned(),
                    count: Arc::clone(&first),
                },
            )
            .unwrap()
            .route(
                "second",
                EventFilter::default(),
                CountingHandler {
                    id: "second-handler".to_owned(),
                    count: Arc::clone(&second),
                },
            )
            .unwrap()
            .build()
            .unwrap();

        multiplexer.handle(&event()).await.unwrap();

        assert_eq!(first.load(Ordering::SeqCst), 1);
        assert_eq!(second.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn multiplexer_skips_non_matching_routes() {
        let count = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "skipped",
                AllowNothing,
                CountingHandler {
                    id: "skipped-handler".to_owned(),
                    count: Arc::clone(&count),
                },
            )
            .unwrap()
            .build()
            .unwrap();

        multiplexer.handle(&event()).await.unwrap();

        assert_eq!(count.load(Ordering::SeqCst), 0);
    }

    #[tokio::test]
    async fn no_match_error_policy_returns_handler_error() {
        let multiplexer = Multiplexer::builder()
            .route("skipped", AllowNothing, FailingHandler)
            .unwrap()
            .on_no_match(NoMatchPolicy::Error)
            .build()
            .unwrap();

        let err = multiplexer.handle(&event()).await.unwrap_err();
        assert!(matches!(err, Error::Handler(_)));
    }

    #[tokio::test]
    async fn matching_handler_failure_returns_handler_error() {
        let multiplexer = Multiplexer::builder()
            .route("failing", EventFilter::default(), FailingHandler)
            .unwrap()
            .build()
            .unwrap();

        let err = multiplexer.handle(&event()).await.unwrap_err();
        assert!(matches!(err, Error::Handler(_)));
    }

    #[tokio::test]
    async fn duplicate_subscriber_id_is_rejected_at_route() {
        let result = Multiplexer::builder()
            .route("same", EventFilter::default(), FailingHandler)
            .unwrap()
            .route("same", EventFilter::default(), FailingHandler);

        assert!(matches!(result, Err(Error::Config(_))));
    }

    #[tokio::test]
    async fn nested_multiplexer_composes() {
        let inner_count = Arc::new(AtomicUsize::new(0));
        let inner = Multiplexer::builder()
            .id("inner")
            .route(
                "inner-sub",
                EventFilter::default(),
                CountingHandler {
                    id: "inner-handler".to_owned(),
                    count: Arc::clone(&inner_count),
                },
            )
            .unwrap()
            .build()
            .unwrap();

        let outer = Multiplexer::builder()
            .id("outer")
            .route("outer-sub", EventFilter::default(), inner)
            .unwrap()
            .build()
            .unwrap();

        outer.handle(&event()).await.unwrap();
        assert_eq!(inner_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn store_skips_completed_subscribers_on_redelivery() {
        let first = Arc::new(AtomicUsize::new(0));
        let second = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "first",
                EventFilter::default(),
                CountingHandler {
                    id: "first-handler".to_owned(),
                    count: Arc::clone(&first),
                },
            )
            .unwrap()
            .route(
                "second",
                EventFilter::default(),
                CountingHandler {
                    id: "second-handler".to_owned(),
                    count: Arc::clone(&second),
                },
            )
            .unwrap()
            .store(TestMultiplexerStore::default())
            .build()
            .unwrap();
        let event = event();

        multiplexer.handle(&event).await.unwrap();
        multiplexer.handle(&event).await.unwrap();

        assert_eq!(first.load(Ordering::SeqCst), 1);
        assert_eq!(second.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn redelivery_runs_only_subscriber_that_did_not_complete() {
        let completed = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "completed",
                EventFilter::default(),
                CountingHandler {
                    id: "completed-handler".to_owned(),
                    count: Arc::clone(&completed),
                },
            )
            .unwrap()
            .route("failing", EventFilter::default(), FailingHandler)
            .unwrap()
            .store(TestMultiplexerStore::default())
            .build()
            .unwrap();
        let event = event();

        assert!(multiplexer.handle(&event).await.is_err());
        assert!(multiplexer.handle(&event).await.is_err());

        assert_eq!(completed.load(Ordering::SeqCst), 1);
    }

    #[derive(Clone)]
    struct MarkFailsStore;

    impl MultiplexerStore for MarkFailsStore {
        async fn is_completed(&self, _: &MultiplexerKey) -> Result<bool> {
            Ok(false)
        }

        async fn mark_completed(&self, _: &MultiplexerKey) -> Result<()> {
            Err(Error::Store("mark failed".to_owned()))
        }
    }

    #[tokio::test]
    async fn mark_completed_failure_surfaces_as_handler_error() {
        let count = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "sub",
                EventFilter::default(),
                CountingHandler {
                    id: "h".to_owned(),
                    count: Arc::clone(&count),
                },
            )
            .unwrap()
            .store(MarkFailsStore)
            .build()
            .unwrap();

        let err = multiplexer.handle(&event()).await.unwrap_err();
        assert!(matches!(err, Error::Handler(_)));
        assert_eq!(count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn concurrency_greater_than_one_runs_handlers() {
        let first = Arc::new(AtomicUsize::new(0));
        let second = Arc::new(AtomicUsize::new(0));
        let multiplexer = Multiplexer::builder()
            .route(
                "first",
                EventFilter::default(),
                CountingHandler {
                    id: "first-handler".to_owned(),
                    count: Arc::clone(&first),
                },
            )
            .unwrap()
            .route(
                "second",
                EventFilter::default(),
                CountingHandler {
                    id: "second-handler".to_owned(),
                    count: Arc::clone(&second),
                },
            )
            .unwrap()
            .concurrency(NonZeroUsize::new(4).unwrap())
            .build()
            .unwrap();

        multiplexer.handle(&event()).await.unwrap();

        assert_eq!(first.load(Ordering::SeqCst), 1);
        assert_eq!(second.load(Ordering::SeqCst), 1);
    }

    #[test]
    fn subscriber_id_with_prefix_joins_with_colon() {
        let id = SubscriberId::with_prefix("orders", "projection").unwrap();
        assert_eq!(id.as_str(), "orders:projection");
    }

    #[test]
    fn subscriber_id_with_prefix_rejects_invalid_segments() {
        // Whitespace in either segment is invalid for SubscriberId.
        assert!(SubscriberId::with_prefix("orders", "bad id").is_err());
        assert!(SubscriberId::with_prefix("bad prefix", "ok").is_err());
        // Length cap (128) is enforced after joining.
        let long = "a".repeat(130);
        assert!(SubscriberId::with_prefix(&long, "id").is_err());
    }

    #[derive(Debug, Clone, PartialEq, Eq)]
    struct OrderPlaced {
        order_id: String,
    }

    struct TypedRecordingHandler {
        id: String,
        seen: Arc<std::sync::Mutex<Vec<String>>>,
    }

    impl Handler<OrderPlaced> for TypedRecordingHandler {
        fn id(&self) -> &str {
            &self.id
        }

        async fn handle(&self, event: &Event<OrderPlaced>) -> Result<()> {
            self.seen
                .lock()
                .unwrap()
                .push(event.payload().order_id.clone());
            Ok(())
        }
    }

    fn typed_event(order_id: &str) -> Event<OrderPlaced> {
        Event::create(
            "acme",
            "/orders",
            "order.placed",
            "order-1",
            OrderPlaced {
                order_id: order_id.to_owned(),
            },
        )
        .unwrap()
    }

    #[tokio::test]
    async fn multiplexer_fanouts_typed_events() {
        let projection_seen: Arc<std::sync::Mutex<Vec<String>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));
        let audit_seen: Arc<std::sync::Mutex<Vec<String>>> =
            Arc::new(std::sync::Mutex::new(Vec::new()));

        let multiplexer: Multiplexer<NoMultiplexerStore, OrderPlaced> = Multiplexer::builder()
            .route(
                SubscriberId::with_prefix("orders", "projection")
                    .unwrap()
                    .as_str(),
                crate::io::filter::AllFilter,
                TypedRecordingHandler {
                    id: "projection".to_owned(),
                    seen: Arc::clone(&projection_seen),
                },
            )
            .unwrap()
            .route(
                SubscriberId::with_prefix("orders", "audit")
                    .unwrap()
                    .as_str(),
                crate::io::filter::AllFilter,
                TypedRecordingHandler {
                    id: "audit".to_owned(),
                    seen: Arc::clone(&audit_seen),
                },
            )
            .unwrap()
            .build()
            .unwrap();

        multiplexer.handle(&typed_event("o-1")).await.unwrap();
        multiplexer.handle(&typed_event("o-2")).await.unwrap();

        assert_eq!(*projection_seen.lock().unwrap(), vec!["o-1", "o-2"]);
        assert_eq!(*audit_seen.lock().unwrap(), vec!["o-1", "o-2"]);
    }
}