//! Streams subsystem. Owns the stream handle and all stream-
//! specific types (configs, retention, consumer types).
//!
//! Entry points live on `Client`:
//! - `client.create_stream(stream::StreamConfig { ... })`
//! - `client.get_stream("name")`
//! - `client.get_or_create_stream(cfg)`
//!
//! Returned `Stream` carries the per-stream operations:
//! - `stream.publish(subject, payload)`
//! - `stream.create_consumer(stream::ConsumerConfig { ... })`

pub mod consumer;

pub use consumer::{Consumer, ConsumerConfig, DeliverPolicy, Message, Messages};

use crate::client::Client;
use crate::error::{Error, Result};
use crate::streams_server::{
    self as pb, CreateConsumerRequest, DeleteConsumerRequest, GetConsumerInfoRequest,
    GetStreamInfoRequest, PublishRequest,
};
use std::time::Duration;
use tonic::Request;

/// One row of per-(sourcing, source) status returned by
/// `Stream::sources_status` and `Client::get_stream_sources`.
#[derive(Debug, Clone)]
pub struct SourceStatus {
    pub sourcing_stream: String,
    pub source_stream: String,
    pub last_sourced_seq: u64,
    pub pulled_total: u64,
    pub last_error: String,
    pub last_error_ts_ms: i64,
}

/// A reference to a stream on the server. Cheap to clone.
#[derive(Clone)]
pub struct Stream {
    pub(crate) client: Client,
    pub(crate) name: String,
}

impl Stream {
    pub(crate) fn new(client: Client, name: String) -> Self {
        Self { client, name }
    }

    /// The stream's name.
    pub fn name(&self) -> &str {
        &self.name
    }

    /// Publish a message into this stream.
    pub async fn publish(
        &self,
        subject: impl Into<String>,
        payload: impl Into<Vec<u8>>,
    ) -> Result<PublishAck> {
        self.publish_with_headers(subject, std::iter::empty::<(String, String)>(), payload)
            .await
    }

    /// Publish with explicit headers.
    pub async fn publish_with_headers(
        &self,
        subject: impl Into<String>,
        headers: impl IntoIterator<Item = (String, String)>,
        payload: impl Into<Vec<u8>>,
    ) -> Result<PublishAck> {
        let mut c = self.client.streams_client();
        let req = PublishRequest {
            stream: self.name.clone(),
            subject: subject.into(),
            payload: payload.into(),
            headers: headers
                .into_iter()
                .map(|(k, v)| pb::MessageHeader { key: k, value: v })
                .collect(),
            ts_ms: 0,
            expected_last_seq: None,
        };
        let r = c.publish(Request::new(req)).await?.into_inner();
        if !r.success {
            return Err(Error::server(r.result_code, r.message));
        }
        Ok(PublishAck { sequence: r.seq })
    }

    /// Slice 3: per-source tail status for this stream. Returns
    /// one entry per source feeding this stream. Empty when the
    /// stream has no sources OR the tail tasks are running on a
    /// different node (this RPC returns the local primary's view).
    pub async fn sources_status(&self) -> Result<Vec<SourceStatus>> {
        let mut c = self.client.streams_client();
        let r = c
            .get_stream_info(Request::new(GetStreamInfoRequest {
                name: self.name.clone(),
            }))
            .await?
            .into_inner();
        if !r.success {
            return Err(Error::server(r.result_code, r.message));
        }
        Ok(r.sources_status
            .into_iter()
            .map(|s| SourceStatus {
                sourcing_stream: self.name.clone(),
                source_stream: s.source_stream,
                last_sourced_seq: s.last_sourced_seq,
                pulled_total: s.pulled_total,
                last_error: s.last_error,
                last_error_ts_ms: s.last_error_ts_ms,
            })
            .collect())
    }

    /// Create a new consumer on this stream. Errors if a consumer
    /// with the same `durable_name` already exists.
    pub async fn create_consumer(&self, config: ConsumerConfig) -> Result<Consumer> {
        let name = config
            .durable_name
            .clone()
            .ok_or_else(|| Error::Invalid("ConsumerConfig.durable_name is required".into()))?;
        let cfg_pb = config.into_pb(&name);
        let mut c = self.client.streams_client();
        let r = c
            .create_consumer(Request::new(CreateConsumerRequest {
                stream: self.name.clone(),
                config: Some(cfg_pb),
            }))
            .await?
            .into_inner();
        if !r.success {
            return Err(Error::server(r.result_code, r.message));
        }
        Ok(Consumer::new(self.client.clone(), self.name.clone(), name))
    }

    /// Idempotent variant — returns an existing consumer if one
    /// with the same `durable_name` already exists, otherwise
    /// creates it.
    pub async fn get_or_create_consumer(&self, config: ConsumerConfig) -> Result<Consumer> {
        let name = config
            .durable_name
            .clone()
            .ok_or_else(|| Error::Invalid("ConsumerConfig.durable_name is required".into()))?;
        match self.get_consumer(&name).await {
            Ok(c) => Ok(c),
            Err(Error::Server { code, .. }) if code == "no_such_consumer" => {
                self.create_consumer(config).await
            }
            Err(e) => Err(e),
        }
    }

    /// Return a handle to an existing consumer.
    pub async fn get_consumer(&self, name: impl Into<String>) -> Result<Consumer> {
        let name: String = name.into();
        let mut c = self.client.streams_client();
        let r = c
            .get_consumer_info(Request::new(GetConsumerInfoRequest {
                stream: self.name.clone(),
                consumer: name.clone(),
            }))
            .await?
            .into_inner();
        if !r.success {
            return Err(Error::server(r.result_code, r.message));
        }
        Ok(Consumer::new(self.client.clone(), self.name.clone(), name))
    }

    /// Delete a consumer by name.
    pub async fn delete_consumer(&self, name: impl Into<String>) -> Result<()> {
        let mut c = self.client.streams_client();
        let r = c
            .delete_consumer(Request::new(DeleteConsumerRequest {
                stream: self.name.clone(),
                consumer: name.into(),
            }))
            .await?
            .into_inner();
        if !r.success {
            return Err(Error::server(r.result_code, r.message));
        }
        Ok(())
    }
}

/// Returned by `Stream::publish` — just the assigned stream
/// sequence; new fields can be added without breaking callers.
#[derive(Debug, Clone, Copy)]
pub struct PublishAck {
    pub sequence: u64,
}

/// Retention policy.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum RetentionPolicy {
    #[default]
    Limits,
    WorkQueue,
    Interest,
}

/// Stream configuration. `Default` produces an unbounded Limits
/// stream with no subject filter, suitable for
/// `StreamConfig { name: ..., ..Default::default() }`.
#[derive(Debug, Clone, Default)]
pub struct StreamConfig {
    pub name: String,
    pub subjects: Vec<String>,
    pub retention: RetentionPolicy,
    /// Maximum age before messages are pruned. `None` = unbounded.
    pub max_age: Option<Duration>,
    /// Maximum number of messages. `None` = unbounded.
    pub max_messages: Option<u64>,
    /// Maximum total stored bytes. `None` = unbounded.
    pub max_bytes: Option<u64>,
    /// Max bytes per individual message. `None` = no cap.
    pub max_message_size: Option<u64>,
    /// Block size override. 0 = server default (100_000).
    pub block_size: u64,
    /// Reject publishes that would push over `max_bytes` /
    /// `max_messages` instead of silently dropping the oldest.
    pub strict_limits: bool,
    /// Memory-only stream. Survives node failover via replication
    /// but a full-cluster restart loses it.
    pub ephemeral: bool,
    /// Per-subject revision cap. `0` (default) = unbounded.
    /// When N > 0, after each publish, older messages at that
    /// subject beyond the N most recent are dropped. Mirrors
    /// NATS JetStream's `MaxMsgsPerSubject` (and backs KV's
    /// `max_revisions` knob).
    pub max_msgs_per_subject: u64,
    /// Cross-stream sources. Slice 1: at most one entry, name-only.
    /// See `crates/streams/specs/SOURCES_DESIGN.md`.
    pub sources: Vec<StreamSource>,
}

/// Identifies a stream to tail from.
#[derive(Debug, Clone, Default)]
pub struct StreamSource {
    pub source_stream: String,
    /// Optional NATS pattern. Only source messages whose subject
    /// matches are pulled. Empty = pull all.
    pub filter_subject: String,
    /// First source-seq to pull. `0` (default) = pull from
    /// beginning of source (subject to `max_initial_backfill`).
    /// Mutually exclusive with `start_time_ms`.
    pub start_seq: u64,
    /// Slice 3: start position by wall-clock timestamp (ms since
    /// epoch). When set, the tail resolves to the first message
    /// with `ts_ms >= start_time_ms` and seeds there. Future-of-
    /// stream resolves to "seed at current end". Mutually exclusive
    /// with `start_seq`.
    pub start_time_ms: i64,
    /// Slice 2F: cap on the initial backfill window. `0` = no cap.
    /// When set, the tail seeds at `max(0, source.last_seq -
    /// max_initial_backfill)`. Ignored once durable state exists.
    pub max_initial_backfill: u64,
    /// Slice 3: NATS-style subject rewrite. When set, the source's
    /// subject is matched against `subject_transform.source_pattern`
    /// and the captured wildcards substituted into
    /// `subject_transform.destination` (1-indexed `{{wildcard(N)}}`
    /// placeholders). The sourcing stream sees the rewritten
    /// subject; the original is preserved on the
    /// `waymaker-source-subject` header.
    pub subject_transform: Option<SubjectTransform>,
    /// Slice 3: what to do when source retention drops messages
    /// past our last_sourced_seq. Default = Halt.
    pub on_drop: OnDropPolicy,
    /// Slice 3: dead-letter stream name. When non-empty, tail
    /// errors (transform mismatch, halt-on-drop firing, etc.) are
    /// published as JSON records to this stream so operators can
    /// triage without scraping logs.
    pub dlq_stream: String,
}

#[derive(Debug, Clone, Default)]
pub struct SubjectTransform {
    pub source_pattern: String,
    pub destination: String,
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum OnDropPolicy {
    /// Tail records a persistent error + stops advancing.
    #[default]
    Halt,
    /// Tail jumps watermark to source's first available seq.
    SkipToFirstAvailable,
}

impl StreamConfig {
    pub(crate) fn into_pb(self) -> pb::StreamConfigPb {
        let retention = match self.retention {
            RetentionPolicy::Limits => pb::Retention {
                policy: Some(pb::retention::Policy::Limits(pb::LimitsRetention {
                    max_age_ms: self.max_age.map(|d| d.as_millis() as u64),
                    max_msgs: self.max_messages,
                    max_bytes: self.max_bytes,
                    strict_limits: self.strict_limits,
                })),
            },
            RetentionPolicy::WorkQueue => pb::Retention {
                policy: Some(pb::retention::Policy::WorkQueue(pb::WorkQueueRetention {})),
            },
            RetentionPolicy::Interest => pb::Retention {
                policy: Some(pb::retention::Policy::Interest(pb::InterestRetention {})),
            },
        };
        pb::StreamConfigPb {
            name: self.name,
            subjects_filter: self.subjects,
            retention: Some(retention),
            block_size: self.block_size,
            max_msg_bytes: self.max_message_size.unwrap_or(0),
            ephemeral: self.ephemeral,
            max_msgs_per_subject: self.max_msgs_per_subject,
            sources: self
                .sources
                .into_iter()
                .map(|s| pb::StreamSourceConfigPb {
                    source_stream: s.source_stream,
                    filter_subject: s.filter_subject,
                    start_seq: s.start_seq,
                    start_time_ms: s.start_time_ms,
                    subject_transform: s.subject_transform.map(|t| {
                        pb::SubjectTransformPb {
                            source_pattern: t.source_pattern,
                            destination: t.destination,
                        }
                    }),
                    max_initial_backfill: s.max_initial_backfill,
                    on_drop: match s.on_drop {
                        OnDropPolicy::Halt => 0,
                        OnDropPolicy::SkipToFirstAvailable => 1,
                    },
                    dlq_stream: s.dlq_stream,
                })
                .collect(),
        }
    }
}

/// Partial-update payload for `Client::update_stream`. Any field
/// left at `None` is not touched server-side.
#[derive(Debug, Clone, Default)]
pub struct StreamUpdate {
    pub max_age_ms: Option<Duration>,
    pub max_msgs: Option<u64>,
    pub max_bytes: Option<u64>,
    pub max_msg_bytes: Option<u64>,
    pub strict_limits: Option<bool>,
}