nstreams 0.3.1

use std::pin::Pin;

use nstreams_core::event::DeliveredEvent;
use nstreams_core::handler::NStreamsHandler;
use nstreams_core::publisher::EventPublisher;
use nstreams_core::worker::{QueueConsumer, StreamWorker};
use nstreams_core::namespace::StreamNamespace;
use nstreams_core::{DEFAULT_MAX_EVENT_BYTES, Result};
use nstreams_postgres::PostgresEventStore;
use nstreams_rabbit::{LapinWriteQueue, QueueConsumerConfig, RabbitReadStream};
use tracing::info;

/// Wired nstreams components ready to embed in an existing service.
#[derive(Clone)]
pub struct NStreams {
    handler: NStreamsHandler<PostgresEventStore, RabbitReadStream>,
    publisher: EventPublisher<LapinWriteQueue>,
    write_queue: LapinWriteQueue,
    store: PostgresEventStore,
    read_stream: RabbitReadStream,
    worker_config: WorkerConfig,
}

#[derive(Clone)]
struct WorkerConfig {
    amqp_url: String,
    prefetch: u16,
}

/// Builder for constructing an embeddable [`NStreams`] instance.
///
/// `database_url` and `amqp_url` are required and must be provided by the
/// host service — nstreams does not read them from the environment.
#[derive(Debug, Clone)]
pub struct NStreamsBuilder {
    database_url: Option<String>,
    amqp_url: Option<String>,
    max_event_bytes: u64,
    worker_prefetch: u16,
}

impl Default for NStreamsBuilder {
    fn default() -> Self {
        Self {
            database_url: None,
            amqp_url: None,
            max_event_bytes: DEFAULT_MAX_EVENT_BYTES,
            worker_prefetch: 100,
        }
    }
}

impl NStreamsBuilder {
    pub fn new() -> Self {
        Self::default()
    }

    /// PostgreSQL connection string. Required; supplied by the host service.
    pub fn database_url(mut self, database_url: impl Into<String>) -> Self {
        self.database_url = Some(database_url.into());
        self
    }

    /// RabbitMQ AMQP URL. Required; supplied by the host service.
    pub fn amqp_url(mut self, amqp_url: impl Into<String>) -> Self {
        self.amqp_url = Some(amqp_url.into());
        self
    }

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

    pub fn worker_prefetch(mut self, prefetch: u16) -> Self {
        self.worker_prefetch = prefetch;
        self
    }

    pub async fn build(self) -> Result<NStreams> {
        let database_url = self
            .database_url
            .ok_or_else(|| nstreams_core::Error::Other("database_url is required".into()))?;
        let amqp_url = self
            .amqp_url
            .ok_or_else(|| nstreams_core::Error::Other("amqp_url is required".into()))?;

        let store = PostgresEventStore::connect(&database_url)
            .await
            .map_err(|error| nstreams_core::Error::Store(error.to_string()))?;
        let write_queue = LapinWriteQueue::connect(&amqp_url)
            .await
            .map_err(|error| nstreams_core::Error::WriteQueue(error.to_string()))?;
        let read_stream = RabbitReadStream::connect(&amqp_url)
            .await
            .map_err(|error| nstreams_core::Error::ReadStream(error.to_string()))?;

        let handler = NStreamsHandler::with_max_event_bytes(
            store.clone(),
            read_stream.clone(),
            self.max_event_bytes,
        );

        Ok(NStreams {
            handler,
            publisher: EventPublisher::new(write_queue.clone()),
            write_queue,
            store,
            read_stream,
            worker_config: WorkerConfig {
                amqp_url,
                prefetch: self.worker_prefetch,
            },
        })
    }
}

impl NStreams {
    pub fn builder() -> NStreamsBuilder {
        NStreamsBuilder::new()
    }

    pub fn handler(&self) -> &NStreamsHandler<PostgresEventStore, RabbitReadStream> {
        &self.handler
    }

    pub fn publisher(&self) -> &EventPublisher<LapinWriteQueue> {
        &self.publisher
    }

    pub fn store(&self) -> &PostgresEventStore {
        &self.store
    }

    pub fn read_stream(&self) -> &RabbitReadStream {
        &self.read_stream
    }

    /// Start the background worker that persists queue messages and pipes to read streams.
    pub async fn spawn_worker(&self) -> Result<()> {
        let worker = StreamWorker::new(
            self.store.clone(),
            self.write_queue.clone(),
            self.read_stream.clone(),
            self.handler.clone(),
        );
        let consumer = self
            .write_queue
            .create_consumer(QueueConsumerConfig {
                amqp_url: self.worker_config.amqp_url.clone(),
                prefetch: self.worker_config.prefetch,
            })
            .await
            .map_err(|error| nstreams_core::Error::WriteQueue(error.to_string()))?;

        tokio::spawn(async move {
            if let Err(error) = worker.run(consumer).await {
                tracing::error!(%error, "nstreams worker exited");
            }
        });

        info!("nstreams worker spawned");
        Ok(())
    }

    /// Run the worker inline on the current task (useful in tests or dedicated worker processes).
    pub async fn run_worker<QC: QueueConsumer>(&self, consumer: QC) -> Result<()> {
        let worker = StreamWorker::new(
            self.store.clone(),
            self.write_queue.clone(),
            self.read_stream.clone(),
            self.handler.clone(),
        );
        worker.run(consumer).await
    }

    /// Subscribe to a namespace from service code without HTTP.
    pub async fn subscribe(
        &self,
        namespace: &StreamNamespace,
        history: u64,
    ) -> Result<nstreams_core::Subscription> {
        self.handler.subscribe(namespace, history).await
    }

    /// Subscribe using a JSON-compatible request, including optional stream filters.
    pub async fn subscribe_request(
        &self,
        request: nstreams_core::SubscribeRequest,
    ) -> Result<nstreams_core::Subscription> {
        self.handler.subscribe_with_request(request).await
    }

    /// Publish an event from service code without HTTP.
    pub async fn publish(
        &self,
        namespace: &StreamNamespace,
        payload: &[u8],
        filter_value: Option<&str>,
    ) -> Result<()> {
        self.publisher.publish(namespace, payload, filter_value).await
    }

    #[cfg(feature = "axum")]
    pub fn axum_routes() -> axum::Router<Self> {
        crate::axum::routes()
    }

    #[cfg(feature = "axum")]
    pub fn axum_router(self) -> axum::Router<Self> {
        crate::axum::router(self)
    }
}

/// Type alias for a boxed event stream usable from any HTTP framework.
pub type EventStream =
    Pin<Box<dyn futures::Stream<Item = Result<DeliveredEvent>> + Send>>;