nstreams-core 0.3.1

use std::sync::Arc;

use async_trait::async_trait;
use tracing::{debug, error, info, warn};

use crate::handler::NStreamsHandler;
use crate::namespace::{Namespace, StreamNamespace};
use crate::store::EventStore;
use crate::stream::{ReadStreamBackend, WriteQueueBackend};

/// Consumes the write queue, persists events, and pipes to ready read streams.
pub struct StreamWorker<S, Q, R> {
    store: Arc<S>,
    read_stream: Arc<R>,
    handler: Arc<NStreamsHandler<S, R>>,
    _queue: std::marker::PhantomData<Q>,
}

impl<S, Q, R> StreamWorker<S, Q, R>
where
    S: EventStore + 'static,
    Q: WriteQueueBackend + 'static,
    R: ReadStreamBackend + 'static,
{
    pub fn new(
        store: S,
        _queue: Q,
        read_stream: R,
        handler: NStreamsHandler<S, R>,
    ) -> Self {
        Self {
            store: Arc::new(store),
            read_stream: Arc::new(read_stream),
            handler: Arc::new(handler),
            _queue: std::marker::PhantomData,
        }
    }

    pub async fn run<QC: QueueConsumer>(&self, mut consumer: QC) -> crate::Result<()> {
        info!("stream worker started");
        while let Some(delivery) = consumer.next().await {
            match delivery {
                Ok(message) => {
                    if let Err(error) = self
                        .process_message(
                            &message.namespace,
                            &message.payload,
                            message.filter_value.as_deref(),
                        )
                        .await
                    {
                        error!(namespace = %message.namespace, %error, "failed to process queue message");
                    }
                }
                Err(error) => {
                    warn!(%error, "queue consumer error");
                }
            }
        }
        Ok(())
    }

    async fn process_message(
        &self,
        namespace: &str,
        payload: &[u8],
        filter_value: Option<&str>,
    ) -> crate::Result<()> {
        let namespace = crate::namespace::StreamNamespace::from(namespace);
        StreamNamespace::validate(namespace.as_str())?;
        self.store.ensure_namespace(&namespace).await?;

        let event = self
            .store
            .persist_next(&namespace, payload, filter_value)
            .await?;
        debug!(
            namespace = namespace.as_str(),
            version = event.version,
            "persisted event"
        );

        if self.handler.is_read_stream_ready(&namespace).await {
            self.read_stream.publish_to_stream(&event).await?;
        }

        Ok(())
    }
}

/// Raw queue message consumed by the worker.
#[derive(Debug, Clone)]
pub struct QueueMessage {
    pub namespace: String,
    pub payload: Vec<u8>,
    pub filter_value: Option<String>,
}

/// Abstraction over the RabbitMQ queue consumer loop.
#[async_trait]
pub trait QueueConsumer: Send {
    async fn next(&mut self) -> Option<std::result::Result<QueueMessage, crate::Error>>;
}