intercom-rs 1.1.1

//! JetStream consumer types and builders.

use std::{
    marker::PhantomData,
    pin::Pin,
    task::{Context, Poll},
};

use futures::Stream;
use pin_project_lite::pin_project;
use serde::de::DeserializeOwned;

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

/// A typed message from a JetStream consumer.
#[derive(Debug)]
pub struct JetStreamMessage<T> {
    /// The decoded message payload.
    pub payload: T,
    /// The subject the message was published to.
    pub subject: String,
    /// The stream sequence number.
    pub stream_sequence: u64,
    /// The consumer sequence number.
    pub consumer_sequence: u64,
    /// The reply subject for request-reply patterns.
    pub reply: Option<String>,
    /// The raw message for acknowledgment.
    raw: async_nats::jetstream::Message,
}

impl<T> JetStreamMessage<T> {
    /// Acknowledge the message.
    pub async fn ack(&self) -> Result<()> {
        self.raw
            .ack()
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Acknowledge the message with double ack.
    pub async fn double_ack(&self) -> Result<()> {
        self.raw
            .double_ack()
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Negatively acknowledge the message.
    pub async fn nak(&self) -> Result<()> {
        self.raw
            .ack_with(async_nats::jetstream::AckKind::Nak(None))
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Negatively acknowledge with a delay before redelivery.
    pub async fn nak_with_delay(&self, delay: std::time::Duration) -> Result<()> {
        self.raw
            .ack_with(async_nats::jetstream::AckKind::Nak(Some(delay)))
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Mark the message as in progress.
    pub async fn in_progress(&self) -> Result<()> {
        self.raw
            .ack_with(async_nats::jetstream::AckKind::Progress)
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Terminate the message (will not be redelivered).
    pub async fn term(&self) -> Result<()> {
        self.raw
            .ack_with(async_nats::jetstream::AckKind::Term)
            .await
            .map_err(|e| crate::error::Error::JetStream(e.to_string()))
    }

    /// Get the raw underlying JetStream message.
    pub fn raw(&self) -> &async_nats::jetstream::Message {
        &self.raw
    }
}

// ============================================================================
// Pull Consumer
// ============================================================================

/// A typed pull consumer with configurable codec.
///
/// # Type Parameters
///
/// * `T` - The message type
/// * `C` - The codec type used for deserialization
pub struct PullConsumer<T, C: CodecType> {
    inner: async_nats::jetstream::consumer::Consumer<async_nats::jetstream::consumer::pull::Config>,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> PullConsumer<T, C> {
    /// Create a new pull consumer wrapper.
    pub(crate) fn new(
        inner: async_nats::jetstream::consumer::Consumer<
            async_nats::jetstream::consumer::pull::Config,
        >,
    ) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }

    /// Get the underlying async-nats consumer.
    pub fn inner(
        &self,
    ) -> &async_nats::jetstream::consumer::Consumer<async_nats::jetstream::consumer::pull::Config>
    {
        &self.inner
    }

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

impl<T: DeserializeOwned, C: CodecType> PullConsumer<T, C> {
    /// Fetch a batch of messages.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    /// use futures::StreamExt;
    ///
    /// #[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.get_pull_consumer::<Event>("my-consumer").await?;
    ///
    /// let mut messages = consumer.fetch(10).await?;
    /// while let Some(result) = messages.next().await {
    ///     let msg = result?;
    ///     println!("Got: {:?}", msg.payload);
    ///     msg.ack().await?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn fetch(&self, batch_size: usize) -> Result<PullBatch<T, C>> {
        let inner = self
            .inner
            .fetch()
            .max_messages(batch_size)
            .messages()
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PullBatch::new(inner))
    }

    /// Fetch messages with a builder for advanced options.
    pub fn fetch_builder(&self) -> FetchBuilder<T, C> {
        FetchBuilder::new(self.inner.clone())
    }

    /// Create a continuous message stream.
    ///
    /// Returns a [`Stream`] that continuously delivers messages.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    /// use futures::StreamExt;
    ///
    /// #[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.get_pull_consumer::<Event>("my-consumer").await?;
    ///
    /// let mut messages = consumer.messages().await?;
    /// while let Some(result) = messages.next().await {
    ///     let msg = result?;
    ///     println!("Got: {:?}", msg.payload);
    ///     msg.ack().await?;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn messages(&self) -> Result<PullMessages<T, C>> {
        let inner = self
            .inner
            .messages()
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PullMessages::new(inner))
    }
}

pin_project! {
    /// A batch of messages from a fetch operation.
    pub struct PullBatch<T, C: CodecType> {
        #[pin]
        inner: async_nats::jetstream::consumer::pull::Batch,
        _marker: PhantomData<(T, C)>,
    }
}

impl<T, C: CodecType> PullBatch<T, C> {
    fn new(inner: async_nats::jetstream::consumer::pull::Batch) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }
}

impl<T: DeserializeOwned, C: CodecType> Stream for PullBatch<T, C> {
    type Item = Result<JetStreamMessage<T>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();

        match this.inner.poll_next(cx) {
            Poll::Ready(Some(Ok(msg))) => {
                let info = msg.info().ok();
                let stream_sequence = info.as_ref().map(|i| i.stream_sequence).unwrap_or(0);
                let consumer_sequence = info.as_ref().map(|i| i.consumer_sequence).unwrap_or(0);
                let result = C::decode(&msg.payload).map(|payload| JetStreamMessage {
                    payload,
                    subject: msg.subject.to_string(),
                    stream_sequence,
                    consumer_sequence,
                    reply: msg.reply.clone().map(|s| s.to_string()),
                    raw: msg,
                });
                Poll::Ready(Some(result))
            }
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(
                crate::error::Error::JetStreamConsumer(e.to_string()),
            ))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

pin_project! {
    /// A stream of messages from a pull consumer.
    pub struct PullMessages<T, C: CodecType> {
        #[pin]
        inner: async_nats::jetstream::consumer::pull::Stream,
        _marker: PhantomData<(T, C)>,
    }
}

impl<T, C: CodecType> PullMessages<T, C> {
    fn new(inner: async_nats::jetstream::consumer::pull::Stream) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }
}

impl<T: DeserializeOwned, C: CodecType> Stream for PullMessages<T, C> {
    type Item = Result<JetStreamMessage<T>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();

        match this.inner.poll_next(cx) {
            Poll::Ready(Some(Ok(msg))) => {
                let info = msg.info().ok();
                let stream_sequence = info.as_ref().map(|i| i.stream_sequence).unwrap_or(0);
                let consumer_sequence = info.as_ref().map(|i| i.consumer_sequence).unwrap_or(0);
                let result = C::decode(&msg.payload).map(|payload| JetStreamMessage {
                    payload,
                    subject: msg.subject.to_string(),
                    stream_sequence,
                    consumer_sequence,
                    reply: msg.reply.clone().map(|s| s.to_string()),
                    raw: msg,
                });
                Poll::Ready(Some(result))
            }
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(
                crate::error::Error::JetStreamConsumer(e.to_string()),
            ))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

/// Builder for fetching messages from a pull consumer.
pub struct FetchBuilder<T, C: CodecType> {
    inner: async_nats::jetstream::consumer::Consumer<async_nats::jetstream::consumer::pull::Config>,
    max_messages: usize,
    max_bytes: Option<usize>,
    expires: Option<std::time::Duration>,
    idle_heartbeat: Option<std::time::Duration>,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> FetchBuilder<T, C> {
    fn new(
        inner: async_nats::jetstream::consumer::Consumer<
            async_nats::jetstream::consumer::pull::Config,
        >,
    ) -> Self {
        Self {
            inner,
            max_messages: 10,
            max_bytes: None,
            expires: None,
            idle_heartbeat: None,
            _marker: PhantomData,
        }
    }

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

    /// Set the maximum bytes to fetch.
    pub fn max_bytes(mut self, max: usize) -> Self {
        self.max_bytes = Some(max);
        self
    }

    /// Set the expiration time.
    pub fn expires(mut self, duration: std::time::Duration) -> Self {
        self.expires = Some(duration);
        self
    }

    /// Set the idle heartbeat interval.
    pub fn idle_heartbeat(mut self, duration: std::time::Duration) -> Self {
        self.idle_heartbeat = Some(duration);
        self
    }
}

impl<T: DeserializeOwned, C: CodecType> FetchBuilder<T, C> {
    /// Execute the fetch.
    pub async fn fetch(self) -> Result<PullBatch<T, C>> {
        let mut fetch = self.inner.fetch().max_messages(self.max_messages);

        if let Some(bytes) = self.max_bytes {
            fetch = fetch.max_bytes(bytes);
        }
        if let Some(expires) = self.expires {
            fetch = fetch.expires(expires);
        }
        if let Some(heartbeat) = self.idle_heartbeat {
            fetch = fetch.heartbeat(heartbeat);
        }

        let inner = fetch
            .messages()
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PullBatch::new(inner))
    }
}

// ============================================================================
// Push Consumer
// ============================================================================

/// A typed push consumer with configurable codec.
///
/// # Type Parameters
///
/// * `T` - The message type
/// * `C` - The codec type used for deserialization
pub struct PushConsumer<T, C: CodecType> {
    inner: async_nats::jetstream::consumer::Consumer<async_nats::jetstream::consumer::push::Config>,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> PushConsumer<T, C> {
    /// Create a new push consumer wrapper.
    pub(crate) fn new(
        inner: async_nats::jetstream::consumer::Consumer<
            async_nats::jetstream::consumer::push::Config,
        >,
    ) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }

    /// Get the underlying async-nats consumer.
    pub fn inner(
        &self,
    ) -> &async_nats::jetstream::consumer::Consumer<async_nats::jetstream::consumer::push::Config>
    {
        &self.inner
    }

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

impl<T: DeserializeOwned, C: CodecType> PushConsumer<T, C> {
    /// Get a message stream from this push consumer.
    pub async fn messages(&self) -> Result<PushMessages<T, C>> {
        let inner = self
            .inner
            .messages()
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PushMessages::new(inner))
    }
}

pin_project! {
    /// A stream of messages from a push consumer.
    pub struct PushMessages<T, C: CodecType> {
        #[pin]
        inner: async_nats::jetstream::consumer::push::Messages,
        _marker: PhantomData<(T, C)>,
    }
}

impl<T, C: CodecType> PushMessages<T, C> {
    fn new(inner: async_nats::jetstream::consumer::push::Messages) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }
}

impl<T: DeserializeOwned, C: CodecType> Stream for PushMessages<T, C> {
    type Item = Result<JetStreamMessage<T>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();

        match this.inner.poll_next(cx) {
            Poll::Ready(Some(Ok(msg))) => {
                let info = msg.info().ok();
                let stream_sequence = info.as_ref().map(|i| i.stream_sequence).unwrap_or(0);
                let consumer_sequence = info.as_ref().map(|i| i.consumer_sequence).unwrap_or(0);
                let result = C::decode(&msg.payload).map(|payload| JetStreamMessage {
                    payload,
                    subject: msg.subject.to_string(),
                    stream_sequence,
                    consumer_sequence,
                    reply: msg.reply.clone().map(|s| s.to_string()),
                    raw: msg,
                });
                Poll::Ready(Some(result))
            }
            Poll::Ready(Some(Err(e))) => Poll::Ready(Some(Err(
                crate::error::Error::JetStreamConsumer(e.to_string()),
            ))),
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}

// ============================================================================
// Pull Consumer Builder
// ============================================================================

/// Builder for creating pull consumers.
pub struct PullConsumerBuilder<T, C: CodecType> {
    stream: async_nats::jetstream::stream::Stream,
    config: async_nats::jetstream::consumer::pull::Config,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> PullConsumerBuilder<T, C> {
    /// Create a new pull consumer builder.
    pub(crate) fn new(stream: async_nats::jetstream::stream::Stream, name: String) -> Self {
        Self {
            stream,
            config: async_nats::jetstream::consumer::pull::Config {
                name: Some(name),
                ..Default::default()
            },
            _marker: PhantomData,
        }
    }

    /// Make this a durable consumer.
    pub fn durable(mut self) -> Self {
        self.config.durable_name = self.config.name.clone();
        self
    }

    /// Set the durable name (different from consumer name).
    pub fn durable_name(mut self, name: impl Into<String>) -> Self {
        self.config.durable_name = Some(name.into());
        self
    }

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

    /// Set multiple filter subjects.
    pub fn filter_subjects(mut self, subjects: Vec<String>) -> Self {
        self.config.filter_subjects = subjects;
        self
    }

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

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

    /// Set the ack wait duration.
    pub fn ack_wait(mut self, duration: std::time::Duration) -> Self {
        self.config.ack_wait = duration;
        self
    }

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

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

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

    /// Start at a specific sequence.
    pub fn start_sequence(mut self, seq: u64) -> Self {
        self.config.deliver_policy =
            async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence {
                start_sequence: seq,
            };
        self
    }

    /// Start at a specific time.
    pub fn start_time(mut self, time: time::OffsetDateTime) -> Self {
        self.config.deliver_policy =
            async_nats::jetstream::consumer::DeliverPolicy::ByStartTime { start_time: time };
        self
    }

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

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

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

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

    /// Set the maximum expiry.
    pub fn max_expires(mut self, duration: std::time::Duration) -> Self {
        self.config.max_expires = duration;
        self
    }

    /// Set the inactive threshold.
    pub fn inactive_threshold(mut self, duration: std::time::Duration) -> Self {
        self.config.inactive_threshold = duration;
        self
    }

    /// Set headers only.
    pub fn headers_only(mut self, headers_only: bool) -> Self {
        self.config.headers_only = headers_only;
        self
    }

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

    /// Set memory storage.
    pub fn memory_storage(mut self, memory: bool) -> Self {
        self.config.memory_storage = memory;
        self
    }

    /// Set the backoff durations for redelivery.
    pub fn backoff(mut self, backoff: Vec<std::time::Duration>) -> Self {
        self.config.backoff = backoff;
        self
    }

    /// Set metadata.
    pub fn metadata(mut self, metadata: std::collections::HashMap<String, String>) -> Self {
        self.config.metadata = metadata;
        self
    }

    /// Create the consumer.
    pub async fn create(self) -> Result<PullConsumer<T, C>> {
        let inner = self
            .stream
            .create_consumer(self.config)
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PullConsumer::new(inner))
    }

    /// Create or update the consumer.
    pub async fn create_or_update(self) -> Result<PullConsumer<T, C>> {
        let name = self.config.name.clone().unwrap_or_default();
        let inner = self
            .stream
            .get_or_create_consumer(&name, self.config)
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PullConsumer::new(inner))
    }
}

// ============================================================================
// Push Consumer Builder
// ============================================================================

/// Builder for creating push consumers.
pub struct PushConsumerBuilder<T, C: CodecType> {
    stream: async_nats::jetstream::stream::Stream,
    config: async_nats::jetstream::consumer::push::Config,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> PushConsumerBuilder<T, C> {
    /// Create a new push consumer builder.
    pub(crate) fn new(stream: async_nats::jetstream::stream::Stream, name: String) -> Self {
        Self {
            stream,
            config: async_nats::jetstream::consumer::push::Config {
                name: Some(name),
                ..Default::default()
            },
            _marker: PhantomData,
        }
    }

    /// Make this a durable consumer.
    pub fn durable(mut self) -> Self {
        self.config.durable_name = self.config.name.clone();
        self
    }

    /// Set the durable name.
    pub fn durable_name(mut self, name: impl Into<String>) -> Self {
        self.config.durable_name = Some(name.into());
        self
    }

    /// Set the deliver subject (required for push consumers).
    pub fn deliver_subject(mut self, subject: impl Into<String>) -> Self {
        self.config.deliver_subject = subject.into();
        self
    }

    /// Set the deliver group (for queue groups).
    pub fn deliver_group(mut self, group: impl Into<String>) -> Self {
        self.config.deliver_group = Some(group.into());
        self
    }

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

    /// Set multiple filter subjects.
    pub fn filter_subjects(mut self, subjects: Vec<String>) -> Self {
        self.config.filter_subjects = subjects;
        self
    }

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

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

    /// Set the ack wait duration.
    pub fn ack_wait(mut self, duration: std::time::Duration) -> Self {
        self.config.ack_wait = duration;
        self
    }

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

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

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

    /// Start at a specific sequence.
    pub fn start_sequence(mut self, seq: u64) -> Self {
        self.config.deliver_policy =
            async_nats::jetstream::consumer::DeliverPolicy::ByStartSequence {
                start_sequence: seq,
            };
        self
    }

    /// Start at a specific time.
    pub fn start_time(mut self, time: time::OffsetDateTime) -> Self {
        self.config.deliver_policy =
            async_nats::jetstream::consumer::DeliverPolicy::ByStartTime { start_time: time };
        self
    }

    /// Set the rate limit.
    pub fn rate_limit(mut self, limit: u64) -> Self {
        self.config.rate_limit = limit;
        self
    }

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

    /// Set the idle heartbeat.
    pub fn idle_heartbeat(mut self, duration: std::time::Duration) -> Self {
        self.config.idle_heartbeat = duration;
        self
    }

    /// Enable flow control.
    pub fn flow_control(mut self, enabled: bool) -> Self {
        self.config.flow_control = enabled;
        self
    }

    /// Set headers only.
    pub fn headers_only(mut self, headers_only: bool) -> Self {
        self.config.headers_only = headers_only;
        self
    }

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

    /// Set memory storage.
    pub fn memory_storage(mut self, memory: bool) -> Self {
        self.config.memory_storage = memory;
        self
    }

    /// Set the backoff durations for redelivery.
    pub fn backoff(mut self, backoff: Vec<std::time::Duration>) -> Self {
        self.config.backoff = backoff;
        self
    }

    /// Set metadata.
    pub fn metadata(mut self, metadata: std::collections::HashMap<String, String>) -> Self {
        self.config.metadata = metadata;
        self
    }

    /// Set the inactive threshold.
    pub fn inactive_threshold(mut self, duration: std::time::Duration) -> Self {
        self.config.inactive_threshold = duration;
        self
    }

    /// Create the consumer.
    pub async fn create(self) -> Result<PushConsumer<T, C>> {
        let inner = self
            .stream
            .create_consumer(self.config)
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PushConsumer::new(inner))
    }

    /// Create or update the consumer.
    pub async fn create_or_update(self) -> Result<PushConsumer<T, C>> {
        let name = self.config.name.clone().unwrap_or_default();
        let inner = self
            .stream
            .get_or_create_consumer(&name, self.config)
            .await
            .map_err(|e| crate::error::Error::JetStreamConsumer(e.to_string()))?;
        Ok(PushConsumer::new(inner))
    }
}

// ============================================================================
// Consumer Policies
// ============================================================================

/// Acknowledgment policy for consumers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum AckPolicy {
    /// Acknowledge each message individually.
    #[default]
    Explicit,
    /// Messages are considered acknowledged on delivery.
    None,
    /// Acknowledge all messages up to and including the acknowledged one.
    All,
}

impl From<AckPolicy> for async_nats::jetstream::consumer::AckPolicy {
    fn from(policy: AckPolicy) -> Self {
        match policy {
            AckPolicy::Explicit => async_nats::jetstream::consumer::AckPolicy::Explicit,
            AckPolicy::None => async_nats::jetstream::consumer::AckPolicy::None,
            AckPolicy::All => async_nats::jetstream::consumer::AckPolicy::All,
        }
    }
}

/// Replay policy for consumers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ReplayPolicy {
    /// Replay messages as fast as possible.
    #[default]
    Instant,
    /// Replay messages at the original rate.
    Original,
}

impl From<ReplayPolicy> for async_nats::jetstream::consumer::ReplayPolicy {
    fn from(policy: ReplayPolicy) -> Self {
        match policy {
            ReplayPolicy::Instant => async_nats::jetstream::consumer::ReplayPolicy::Instant,
            ReplayPolicy::Original => async_nats::jetstream::consumer::ReplayPolicy::Original,
        }
    }
}

/// Deliver policy for consumers.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum DeliverPolicy {
    /// Deliver all messages.
    #[default]
    All,
    /// Deliver only the last message.
    Last,
    /// Deliver only new messages.
    New,
    /// Deliver the last message per subject.
    LastPerSubject,
}

impl From<DeliverPolicy> for async_nats::jetstream::consumer::DeliverPolicy {
    fn from(policy: DeliverPolicy) -> Self {
        match policy {
            DeliverPolicy::All => async_nats::jetstream::consumer::DeliverPolicy::All,
            DeliverPolicy::Last => async_nats::jetstream::consumer::DeliverPolicy::Last,
            DeliverPolicy::New => async_nats::jetstream::consumer::DeliverPolicy::New,
            DeliverPolicy::LastPerSubject => {
                async_nats::jetstream::consumer::DeliverPolicy::LastPerSubject
            }
        }
    }
}