aster_forge_config/notification/
config.rs

1use serde::{Deserialize, Serialize};
2#[cfg(feature = "redis-pubsub")]
3use std::sync::Arc;
4
5use crate::{ConfigCoreError, Result};
6
7#[cfg(feature = "redis-pubsub")]
8use super::notifier::RedisConfigChangeNotifier;
9#[cfg(feature = "redis-pubsub")]
10use super::notifier::SharedConfigChangeNotifier;
11use super::runtime::ConfigSyncRuntime;
12
13/// Disabled config-sync backend name.
14pub const CONFIG_SYNC_BACKEND_DISABLED: &str = "disabled";
15/// Redis pub/sub config-sync backend name.
16pub const CONFIG_SYNC_BACKEND_REDIS: &str = "redis";
17
18/// Static configuration for cross-process config reload synchronization.
19///
20/// The field names describe a generic broker contract instead of a Redis-only
21/// shape. Current services can map `backend = "redis"` to Redis pub/sub, while
22/// future RabbitMQ, NATS, or other transports can reuse the same product config
23/// surface and add backend-specific interpretation behind the notifier factory.
24#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
25pub struct ConfigSyncConfig {
26    /// Transport backend name, for example `disabled` or `redis`.
27    #[serde(default = "ConfigSyncConfig::default_backend")]
28    pub backend: String,
29    /// Broker endpoint URL. Redis uses a Redis URL.
30    #[serde(default)]
31    pub endpoint: String,
32    /// Logical reload topic. Transports may map this to a channel, exchange,
33    /// subject, or routing key.
34    #[serde(default = "ConfigSyncConfig::default_topic")]
35    pub topic: String,
36}
37
38impl Default for ConfigSyncConfig {
39    fn default() -> Self {
40        Self {
41            backend: Self::default_backend(),
42            endpoint: String::new(),
43            topic: Self::default_topic(),
44        }
45    }
46}
47
48impl ConfigSyncConfig {
49    /// Returns the default disabled backend name.
50    pub fn default_backend() -> String {
51        CONFIG_SYNC_BACKEND_DISABLED.to_string()
52    }
53
54    /// Returns the default logical reload topic.
55    pub fn default_topic() -> String {
56        "aster.config_reload".to_string()
57    }
58
59    /// Returns whether cross-process sync is enabled.
60    pub fn enabled(&self) -> bool {
61        !matches!(
62            self.backend.trim().to_ascii_lowercase().as_str(),
63            "" | "disabled" | "none"
64        )
65    }
66}
67
68/// Returns the conventional config-sync topic for a product namespace.
69pub fn default_config_sync_topic(namespace: &str) -> String {
70    format!("{}.config_reload", namespace.trim())
71}
72
73/// Builds a namespaced config-sync runtime from static config.
74///
75/// This common backend factory owns backend dispatch, runtime ID generation, and
76/// transport-specific topic mapping. Product crates only pass their namespace
77/// and provide their reload callback to [`ConfigSyncRuntime::run_reload_subscription`].
78pub fn build_config_sync_runtime(
79    config: &ConfigSyncConfig,
80    namespace: &str,
81) -> Result<ConfigSyncRuntime> {
82    build_config_sync_runtime_with_runtime_id(
83        config,
84        namespace,
85        aster_forge_utils::id::new_runtime_id(),
86    )
87}
88
89/// Builds a namespaced config-sync runtime with an explicit runtime ID.
90///
91/// Products normally use [`build_config_sync_runtime`]. This variant is useful when the product
92/// already has a stable process identity or when tests need deterministic self-origin filtering.
93pub fn build_config_sync_runtime_with_runtime_id(
94    config: &ConfigSyncConfig,
95    namespace: &str,
96    runtime_id: impl Into<String>,
97) -> Result<ConfigSyncRuntime> {
98    let namespace = namespace.trim();
99    let runtime_id = runtime_id.into();
100    let topic = config_sync_topic(config, namespace);
101    match config.backend.trim().to_ascii_lowercase().as_str() {
102        "" | "disabled" | "none" => Ok(ConfigSyncRuntime::disabled_with_runtime_id(
103            namespace, runtime_id,
104        )),
105        CONFIG_SYNC_BACKEND_REDIS => {
106            build_redis_config_sync_runtime(config, namespace, runtime_id, &topic)
107        }
108        backend => Err(ConfigCoreError::invalid_value(format!(
109            "unsupported config_sync.backend '{backend}'"
110        ))),
111    }
112}
113fn config_sync_topic(config: &ConfigSyncConfig, namespace: &str) -> String {
114    let topic = config.topic.trim();
115    if topic.is_empty() || topic == ConfigSyncConfig::default_topic() {
116        default_config_sync_topic(namespace)
117    } else {
118        topic.to_string()
119    }
120}
121
122#[cfg(feature = "redis-pubsub")]
123fn build_redis_config_sync_runtime(
124    config: &ConfigSyncConfig,
125    namespace: &str,
126    runtime_id: String,
127    topic: &str,
128) -> Result<ConfigSyncRuntime> {
129    if config.endpoint.trim().is_empty() {
130        return Err(ConfigCoreError::invalid_value(
131            "config_sync.endpoint is required when config_sync.backend is redis",
132        ));
133    }
134    let notifier = RedisConfigChangeNotifier::from_url(
135        config.endpoint.trim(),
136        redis_channel_from_topic(topic),
137    )?;
138    Ok(ConfigSyncRuntime::new(
139        namespace,
140        runtime_id,
141        Arc::new(notifier) as SharedConfigChangeNotifier,
142    ))
143}
144
145#[cfg(not(feature = "redis-pubsub"))]
146fn build_redis_config_sync_runtime(
147    _config: &ConfigSyncConfig,
148    _namespace: &str,
149    _runtime_id: String,
150    _topic: &str,
151) -> Result<ConfigSyncRuntime> {
152    Err(ConfigCoreError::invalid_value(
153        "config_sync.backend 'redis' requires the redis-pubsub feature",
154    ))
155}
156
157#[cfg(any(feature = "redis-pubsub", test))]
158pub(super) fn redis_channel_from_topic(topic: &str) -> String {
159    topic.trim().replace('.', ":")
160}
aster_forge_config/notification/config.rs

aster_forge_config/notification/
config.rs
