nstreams_core/
namespace.rs

1use serde::{Deserialize, Serialize};
2
3/// Identifies an event namespace.
4///
5/// The namespace determines queue/stream names, backend storage, and payload shape.
6pub trait Namespace: Clone + Send + Sync + 'static {
7    fn as_str(&self) -> &str;
8}
9
10/// String-backed namespace identifier.
11#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
12#[serde(transparent)]
13pub struct StreamNamespace(String);
14
15impl StreamNamespace {
16    pub fn new(namespace: impl Into<String>) -> Self {
17        let namespace = namespace.into();
18        debug_assert!(!namespace.is_empty(), "namespace must not be empty");
19        Self(namespace)
20    }
21
22    pub fn validate(namespace: &str) -> crate::Result<()> {
23        if namespace.is_empty() {
24            return Err(crate::Error::InvalidNamespace(
25                namespace.to_string(),
26                "must not be empty".to_string(),
27            ));
28        }
29        if namespace.chars().any(|c| c.is_whitespace()) {
30            return Err(crate::Error::InvalidNamespace(
31                namespace.to_string(),
32                "must not contain whitespace".to_string(),
33            ));
34        }
35        Ok(())
36    }
37}
38
39impl Namespace for StreamNamespace {
40    fn as_str(&self) -> &str {
41        &self.0
42    }
43}
44
45impl From<String> for StreamNamespace {
46    fn from(value: String) -> Self {
47        Self(value)
48    }
49}
50
51impl From<&str> for StreamNamespace {
52    fn from(value: &str) -> Self {
53        Self(value.to_string())
54    }
55}
56
57/// RabbitMQ queue name for the write path of a namespace.
58pub fn write_queue_name(namespace: &str) -> String {
59    format!("nstreams_{}_write", sanitize_identifier(namespace))
60}
61
62/// RabbitMQ stream name for the read path of a namespace.
63pub fn read_stream_name(namespace: &str) -> String {
64    format!("nstreams_{}_read", sanitize_identifier(namespace))
65}
66
67/// PostgreSQL table name for a namespace's persisted events.
68pub fn postgres_table_name(namespace: &str) -> String {
69    format!("nstreams_{}", sanitize_identifier(namespace))
70}
71
72fn sanitize_identifier(namespace: &str) -> String {
73    namespace
74        .chars()
75        .map(|c| {
76            if c.is_ascii_alphanumeric() || c == '_' {
77                c
78            } else {
79                '_'
80            }
81        })
82        .collect()
83}
84
85/// Resolve resource names for a typed namespace.
86pub struct NamespaceResources {
87    pub namespace: String,
88    pub write_queue: String,
89    pub read_stream: String,
90    pub postgres_table: String,
91}
92
93impl NamespaceResources {
94    pub fn new<N: Namespace>(namespace: &N) -> Self {
95        let namespace = namespace.as_str().to_string();
96        Self {
97            write_queue: write_queue_name(&namespace),
98            read_stream: read_stream_name(&namespace),
99            postgres_table: postgres_table_name(&namespace),
100            namespace,
101        }
102    }
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108    use crate::namespace::StreamNamespace;
109
110    #[test]
111    fn resource_names_are_derived_from_namespace() {
112        let ns = StreamNamespace::new("candles.btc.5m");
113        let resources = NamespaceResources::new(&ns);
114        assert_eq!(resources.write_queue, "nstreams_candles_btc_5m_write");
115        assert_eq!(resources.read_stream, "nstreams_candles_btc_5m_read");
116        assert_eq!(resources.postgres_table, "nstreams_candles_btc_5m");
117    }
118
119    #[test]
120    fn rejects_whitespace_in_namespace() {
121        let error = StreamNamespace::validate("bad namespace").unwrap_err();
122        assert!(matches!(error, crate::Error::InvalidNamespace(_, _)));
123    }
124}
nstreams_core/namespace.rs

nstreams_core/
namespace.rs
