intercom-rs 1.1.1

//! JetStream stream configuration and management.

use std::marker::PhantomData;

use crate::{codec::CodecType, error::{Error, Result}};

use super::consumer::{PullConsumerBuilder, PushConsumerBuilder};

/// A JetStream stream with configurable codec.
///
/// # Type Parameters
///
/// * `C` - The codec type used for message serialization
#[derive(Clone)]
pub struct Stream<C: CodecType> {
    inner: async_nats::jetstream::stream::Stream,
    _codec: PhantomData<C>,
}

impl<C: CodecType> Stream<C> {
    /// Create a new stream wrapper.
    pub(crate) fn new(inner: async_nats::jetstream::stream::Stream) -> Self {
        Self {
            inner,
            _codec: PhantomData,
        }
    }

    /// Get the underlying async-nats stream.
    pub fn inner(&self) -> &async_nats::jetstream::stream::Stream {
        &self.inner
    }

    /// Get the stream name.
    pub fn name(&self) -> &str {
        &self.inner.cached_info().config.name
    }

    /// Get stream information.
    pub async fn info(&self) -> Result<StreamInfo> {
        let mut stream = self.inner.clone();
        let info = stream
            .info()
            .await
            .map_err(|e| Error::JetStreamStream(e.to_string()))?;
        Ok(StreamInfo {
            config: StreamConfig::from_native(&info.config),
            state: StreamState {
                messages: info.state.messages,
                bytes: info.state.bytes,
                first_sequence: info.state.first_sequence,
                last_sequence: info.state.last_sequence,
                consumer_count: info.state.consumer_count,
            },
        })
    }

    /// Create a pull consumer builder.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type for this consumer
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize, Debug)]
    /// struct Event { id: u64 }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// let jetstream = client.jetstream();
    /// let stream = jetstream.get_stream("events").await?;
    ///
    /// let consumer = stream
    ///     .pull_consumer_builder::<Event>("my-consumer")
    ///     .durable()
    ///     .filter_subject("events.user.>")
    ///     .create()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn pull_consumer_builder<T>(&self, name: &str) -> PullConsumerBuilder<T, C> {
        PullConsumerBuilder::new(self.inner.clone(), name.to_string())
    }

    /// Create a push consumer builder.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type for this consumer
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize, Debug)]
    /// struct Event { id: u64 }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// let jetstream = client.jetstream();
    /// let stream = jetstream.get_stream("events").await?;
    ///
    /// let consumer = stream
    ///     .push_consumer_builder::<Event>("my-push-consumer")
    ///     .deliver_subject("deliver.events")
    ///     .create()
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn push_consumer_builder<T>(&self, name: &str) -> PushConsumerBuilder<T, C> {
        PushConsumerBuilder::new(self.inner.clone(), name.to_string())
    }

    /// Get an existing pull consumer by name.
    pub async fn get_pull_consumer<T>(
        &self,
        name: &str,
    ) -> Result<super::consumer::PullConsumer<T, C>> {
        let inner: async_nats::jetstream::consumer::Consumer<
            async_nats::jetstream::consumer::pull::Config,
        > = self
            .inner
            .get_consumer(name)
            .await
            .map_err(|e| Error::JetStreamConsumer(e.to_string()))?;
        Ok(super::consumer::PullConsumer::new(inner))
    }

    /// Get an existing push consumer by name.
    pub async fn get_push_consumer<T>(
        &self,
        name: &str,
    ) -> Result<super::consumer::PushConsumer<T, C>> {
        let inner: async_nats::jetstream::consumer::Consumer<
            async_nats::jetstream::consumer::push::Config,
        > = self
            .inner
            .get_consumer(name)
            .await
            .map_err(|e| Error::JetStreamConsumer(e.to_string()))?;
        Ok(super::consumer::PushConsumer::new(inner))
    }

    /// Delete a consumer by name.
    pub async fn delete_consumer(&self, name: &str) -> Result<()> {
        self.inner
            .delete_consumer(name)
            .await
            .map_err(|e| Error::JetStreamConsumer(e.to_string()))?;
        Ok(())
    }

    /// Purge all messages from the stream.
    pub async fn purge(&self) -> Result<u64> {
        let response = self
            .inner
            .clone()
            .purge()
            .await
            .map_err(|e| Error::JetStreamStream(e.to_string()))?;
        Ok(response.purged)
    }

    /// Purge messages matching a filter subject.
    pub async fn purge_subject(&self, filter: &str) -> Result<u64> {
        let response = self
            .inner
            .clone()
            .purge()
            .filter(filter)
            .await
            .map_err(|e| Error::JetStreamStream(e.to_string()))?;
        Ok(response.purged)
    }
}

/// Builder for creating JetStream streams.
pub struct StreamBuilder<C: CodecType> {
    context: async_nats::jetstream::Context,
    config: async_nats::jetstream::stream::Config,
    _codec: PhantomData<C>,
}

impl<C: CodecType> StreamBuilder<C> {
    /// Create a new stream builder.
    pub(crate) fn new(context: async_nats::jetstream::Context, name: String) -> Self {
        Self {
            context,
            config: async_nats::jetstream::stream::Config {
                name,
                ..Default::default()
            },
            _codec: PhantomData,
        }
    }

    /// Set the subjects for this stream.
    pub fn subjects(mut self, subjects: Vec<String>) -> Self {
        self.config.subjects = subjects;
        self
    }

    /// Add a single subject to this stream.
    pub fn subject(mut self, subject: impl Into<String>) -> Self {
        self.config.subjects.push(subject.into());
        self
    }

    /// Set the description for this stream.
    pub fn description(mut self, description: impl Into<String>) -> Self {
        self.config.description = Some(description.into());
        self
    }

    /// Set the retention policy.
    pub fn retention(mut self, retention: RetentionPolicy) -> Self {
        self.config.retention = retention.into();
        self
    }

    /// Set the maximum number of messages.
    pub fn max_messages(mut self, max: i64) -> Self {
        self.config.max_messages = max;
        self
    }

    /// Set the maximum number of messages per subject.
    pub fn max_messages_per_subject(mut self, max: i64) -> Self {
        self.config.max_messages_per_subject = max;
        self
    }

    /// Set the maximum bytes.
    pub fn max_bytes(mut self, max: i64) -> Self {
        self.config.max_bytes = max;
        self
    }

    /// Set the maximum message size.
    pub fn max_message_size(mut self, max: i32) -> Self {
        self.config.max_message_size = max;
        self
    }

    /// Set the maximum age for messages.
    pub fn max_age(mut self, age: std::time::Duration) -> Self {
        self.config.max_age = age;
        self
    }

    /// Set the maximum number of consumers.
    pub fn max_consumers(mut self, max: i32) -> Self {
        self.config.max_consumers = max;
        self
    }

    /// Set the number of replicas.
    pub fn replicas(mut self, replicas: usize) -> Self {
        self.config.num_replicas = replicas;
        self
    }

    /// Set the storage type.
    pub fn storage(mut self, storage: StorageType) -> Self {
        self.config.storage = storage.into();
        self
    }

    /// Set the discard policy.
    pub fn discard_policy(mut self, policy: DiscardPolicy) -> Self {
        self.config.discard = policy.into();
        self
    }

    /// Enable or disable duplicate detection window.
    pub fn duplicate_window(mut self, window: std::time::Duration) -> Self {
        self.config.duplicate_window = window;
        self
    }

    /// Allow direct gets.
    pub fn allow_direct(mut self, allow: bool) -> Self {
        self.config.allow_direct = allow;
        self
    }

    /// Enable mirroring from another stream.
    pub fn mirror(mut self, source: StreamSource) -> Self {
        self.config.mirror = Some(source.into());
        self
    }

    /// Add a source stream.
    pub fn add_source(mut self, source: StreamSource) -> Self {
        self.config.sources.get_or_insert_with(Vec::new).push(source.into());
        self
    }

    /// Set whether this is a sealed stream.
    pub fn sealed(mut self, sealed: bool) -> Self {
        self.config.sealed = sealed;
        self
    }

    /// Set whether to deny delete.
    pub fn deny_delete(mut self, deny: bool) -> Self {
        self.config.deny_delete = deny;
        self
    }

    /// Set whether to deny purge.
    pub fn deny_purge(mut self, deny: bool) -> Self {
        self.config.deny_purge = deny;
        self
    }

    /// Set whether to allow rollup headers.
    pub fn allow_rollup(mut self, allow: bool) -> Self {
        self.config.allow_rollup = allow;
        self
    }

    /// Set the compression type.
    pub fn compression(mut self, compression: Compression) -> Self {
        self.config.compression = Some(compression.into());
        self
    }

    /// Set the first sequence number.
    pub fn first_sequence(mut self, seq: u64) -> Self {
        self.config.first_sequence = Some(seq);
        self
    }

    /// Set a subject transform.
    pub fn subject_transform(mut self, source: &str, destination: &str) -> Self {
        self.config.subject_transform = Some(async_nats::jetstream::stream::SubjectTransform {
            source: source.to_string(),
            destination: destination.to_string(),
        });
        self
    }

    /// Create the stream.
    pub async fn create(self) -> Result<Stream<C>> {
        let inner = self
            .context
            .create_stream(self.config)
            .await?;
        Ok(Stream::new(inner))
    }

    /// Create or update the stream.
    pub async fn create_or_update(self) -> Result<Stream<C>> {
        let inner = self
            .context
            .get_or_create_stream(self.config)
            .await?;
        Ok(Stream::new(inner))
    }
}

/// Retention policy for a stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum RetentionPolicy {
    /// Messages are retained until the limits are reached.
    #[default]
    Limits,
    /// Messages are retained until acknowledged by all consumers (interest-based).
    Interest,
    /// Messages are removed after being acknowledged (work queue).
    WorkQueue,
}

impl From<RetentionPolicy> for async_nats::jetstream::stream::RetentionPolicy {
    fn from(policy: RetentionPolicy) -> Self {
        match policy {
            RetentionPolicy::Limits => async_nats::jetstream::stream::RetentionPolicy::Limits,
            RetentionPolicy::Interest => async_nats::jetstream::stream::RetentionPolicy::Interest,
            RetentionPolicy::WorkQueue => async_nats::jetstream::stream::RetentionPolicy::WorkQueue,
        }
    }
}

/// Storage type for a stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum StorageType {
    /// File-based storage.
    #[default]
    File,
    /// Memory-based storage.
    Memory,
}

impl From<StorageType> for async_nats::jetstream::stream::StorageType {
    fn from(storage: StorageType) -> Self {
        match storage {
            StorageType::File => async_nats::jetstream::stream::StorageType::File,
            StorageType::Memory => async_nats::jetstream::stream::StorageType::Memory,
        }
    }
}

/// Discard policy for a stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum DiscardPolicy {
    /// Discard old messages when limits are reached.
    #[default]
    Old,
    /// Discard new messages when limits are reached.
    New,
}

impl From<DiscardPolicy> for async_nats::jetstream::stream::DiscardPolicy {
    fn from(policy: DiscardPolicy) -> Self {
        match policy {
            DiscardPolicy::Old => async_nats::jetstream::stream::DiscardPolicy::Old,
            DiscardPolicy::New => async_nats::jetstream::stream::DiscardPolicy::New,
        }
    }
}

/// Compression type for a stream.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Compression {
    /// No compression.
    #[default]
    None,
    /// S2 compression.
    S2,
}

impl From<Compression> for async_nats::jetstream::stream::Compression {
    fn from(compression: Compression) -> Self {
        match compression {
            Compression::None => async_nats::jetstream::stream::Compression::None,
            Compression::S2 => async_nats::jetstream::stream::Compression::S2,
        }
    }
}

/// Stream source configuration.
#[derive(Debug, Clone)]
pub struct StreamSource {
    /// The source stream name.
    pub name: String,
    /// Optional starting sequence.
    pub start_seq: Option<u64>,
    /// Optional starting time.
    pub start_time: Option<time::OffsetDateTime>,
    /// Optional filter subject.
    pub filter_subject: Option<String>,
}

impl StreamSource {
    /// Create a new stream source.
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            start_seq: None,
            start_time: None,
            filter_subject: None,
        }
    }

    /// Set the starting sequence.
    pub fn start_seq(mut self, seq: u64) -> Self {
        self.start_seq = Some(seq);
        self
    }

    /// Set the starting time.
    pub fn start_time(mut self, time: time::OffsetDateTime) -> Self {
        self.start_time = Some(time);
        self
    }

    /// Set the filter subject.
    pub fn filter_subject(mut self, subject: impl Into<String>) -> Self {
        self.filter_subject = Some(subject.into());
        self
    }
}

impl From<StreamSource> for async_nats::jetstream::stream::Source {
    fn from(source: StreamSource) -> Self {
        let mut s = async_nats::jetstream::stream::Source {
            name: source.name,
            ..Default::default()
        };
        if let Some(seq) = source.start_seq {
            s.start_sequence = Some(seq);
        }
        if let Some(time) = source.start_time {
            s.start_time = Some(time);
        }
        if let Some(subject) = source.filter_subject {
            s.filter_subject = Some(subject);
        }
        s
    }
}

/// Stream configuration.
#[derive(Debug, Clone)]
pub struct StreamConfig {
    /// Stream name.
    pub name: String,
    /// Stream description.
    pub description: Option<String>,
    /// Subjects this stream listens on.
    pub subjects: Vec<String>,
    /// Retention policy.
    pub retention: RetentionPolicy,
    /// Maximum messages.
    pub max_messages: i64,
    /// Maximum bytes.
    pub max_bytes: i64,
    /// Maximum age.
    pub max_age: std::time::Duration,
    /// Maximum message size.
    pub max_message_size: i32,
    /// Storage type.
    pub storage: StorageType,
    /// Number of replicas.
    pub replicas: usize,
}

impl StreamConfig {
    /// Convert from native async-nats config.
    pub(crate) fn from_native(config: &async_nats::jetstream::stream::Config) -> Self {
        Self {
            name: config.name.clone(),
            description: config.description.clone(),
            subjects: config.subjects.clone(),
            retention: match config.retention {
                async_nats::jetstream::stream::RetentionPolicy::Limits => RetentionPolicy::Limits,
                async_nats::jetstream::stream::RetentionPolicy::Interest => {
                    RetentionPolicy::Interest
                }
                async_nats::jetstream::stream::RetentionPolicy::WorkQueue => {
                    RetentionPolicy::WorkQueue
                }
            },
            max_messages: config.max_messages,
            max_bytes: config.max_bytes,
            max_age: config.max_age,
            max_message_size: config.max_message_size,
            storage: match config.storage {
                async_nats::jetstream::stream::StorageType::File => StorageType::File,
                async_nats::jetstream::stream::StorageType::Memory => StorageType::Memory,
            },
            replicas: config.num_replicas,
        }
    }
}

/// Stream state information.
#[derive(Debug, Clone)]
pub struct StreamState {
    /// Number of messages.
    pub messages: u64,
    /// Total bytes.
    pub bytes: u64,
    /// First sequence number.
    pub first_sequence: u64,
    /// Last sequence number.
    pub last_sequence: u64,
    /// Number of consumers.
    pub consumer_count: usize,
}

/// Stream information.
#[derive(Debug, Clone)]
pub struct StreamInfo {
    /// Stream configuration.
    pub config: StreamConfig,
    /// Stream state.
    pub state: StreamState,
}