nstreams-core 0.2.2

use serde::{Deserialize, Serialize};

/// Identifies an event namespace.
///
/// The namespace determines queue/stream names, backend storage, and payload shape.
pub trait Namespace: Clone + Send + Sync + 'static {
    fn as_str(&self) -> &str;
}

/// String-backed namespace identifier.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct StreamNamespace(String);

impl StreamNamespace {
    pub fn new(namespace: impl Into<String>) -> Self {
        let namespace = namespace.into();
        debug_assert!(!namespace.is_empty(), "namespace must not be empty");
        Self(namespace)
    }

    pub fn validate(namespace: &str) -> crate::Result<()> {
        if namespace.is_empty() {
            return Err(crate::Error::InvalidNamespace(
                namespace.to_string(),
                "must not be empty".to_string(),
            ));
        }
        if namespace.chars().any(|c| c.is_whitespace()) {
            return Err(crate::Error::InvalidNamespace(
                namespace.to_string(),
                "must not contain whitespace".to_string(),
            ));
        }
        Ok(())
    }
}

impl Namespace for StreamNamespace {
    fn as_str(&self) -> &str {
        &self.0
    }
}

impl From<String> for StreamNamespace {
    fn from(value: String) -> Self {
        Self(value)
    }
}

impl From<&str> for StreamNamespace {
    fn from(value: &str) -> Self {
        Self(value.to_string())
    }
}

/// RabbitMQ queue name for the write path of a namespace.
pub fn write_queue_name(namespace: &str) -> String {
    format!("nstreams_{}_write", sanitize_identifier(namespace))
}

/// RabbitMQ stream name for the read path of a namespace.
pub fn read_stream_name(namespace: &str) -> String {
    format!("nstreams_{}_read", sanitize_identifier(namespace))
}

/// PostgreSQL table name for a namespace's persisted events.
pub fn postgres_table_name(namespace: &str) -> String {
    format!("nstreams_{}", sanitize_identifier(namespace))
}

fn sanitize_identifier(namespace: &str) -> String {
    namespace
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '_' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// Resolve resource names for a typed namespace.
pub struct NamespaceResources {
    pub namespace: String,
    pub write_queue: String,
    pub read_stream: String,
    pub postgres_table: String,
}

impl NamespaceResources {
    pub fn new<N: Namespace>(namespace: &N) -> Self {
        let namespace = namespace.as_str().to_string();
        Self {
            write_queue: write_queue_name(&namespace),
            read_stream: read_stream_name(&namespace),
            postgres_table: postgres_table_name(&namespace),
            namespace,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::namespace::StreamNamespace;

    #[test]
    fn resource_names_are_derived_from_namespace() {
        let ns = StreamNamespace::new("candles.btc.5m");
        let resources = NamespaceResources::new(&ns);
        assert_eq!(resources.write_queue, "nstreams_candles_btc_5m_write");
        assert_eq!(resources.read_stream, "nstreams_candles_btc_5m_read");
        assert_eq!(resources.postgres_table, "nstreams_candles_btc_5m");
    }

    #[test]
    fn rejects_whitespace_in_namespace() {
        let error = StreamNamespace::validate("bad namespace").unwrap_err();
        assert!(matches!(error, crate::Error::InvalidNamespace(_, _)));
    }
}