aster_forge_config/notification/
message.rs

1use serde::{Deserialize, Serialize};
2use std::future::Future;
3
4use crate::Result;
5
6/// Source that emitted a configuration reload notification.
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
8#[serde(rename_all = "snake_case")]
9pub enum ConfigNotificationSource {
10    /// Notification was emitted by an API mutation.
11    Api,
12    /// Notification was emitted by a CLI operation.
13    Cli,
14    /// Notification was emitted by a startup/bootstrap path.
15    Startup,
16    /// Notification was emitted by an unspecified or product-specific source.
17    Other(String),
18}
19
20/// Notification payload published after configuration changes.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct ConfigReloadMessage {
23    /// Product or service namespace, for example `aster_yggdrasil`.
24    pub namespace: String,
25    /// Runtime instance ID that emitted the message. Receivers can use this to ignore
26    /// their own message after already applying the local change.
27    pub origin_runtime_id: String,
28    /// Changed keys. Empty means receivers should reload all runtime config.
29    pub keys: Vec<String>,
30    /// Source of the change.
31    pub source: ConfigNotificationSource,
32}
33
34impl ConfigReloadMessage {
35    /// Creates a reload message and sorts/deduplicates keys.
36    pub fn new(
37        namespace: impl Into<String>,
38        origin_runtime_id: impl Into<String>,
39        keys: impl IntoIterator<Item = impl Into<String>>,
40        source: ConfigNotificationSource,
41    ) -> Self {
42        let mut keys = keys.into_iter().map(Into::into).collect::<Vec<_>>();
43        keys.sort();
44        keys.dedup();
45        Self {
46            namespace: namespace.into(),
47            origin_runtime_id: origin_runtime_id.into(),
48            keys,
49            source,
50        }
51    }
52
53    /// Serializes the message for transport.
54    ///
55    /// # Errors
56    ///
57    /// Returns [`ConfigError`] when the reload message cannot be serialized.
58    pub fn encode(&self) -> Result<String> {
59        serde_json::to_string(self).map_err(Into::into)
60    }
61
62    /// Decodes a transport payload.
63    ///
64    /// # Errors
65    ///
66    /// Returns [`ConfigError`] when the transport payload is malformed or fails message validation.
67    pub fn decode(payload: &str) -> Result<Self> {
68        serde_json::from_str(payload).map_err(Into::into)
69    }
70}
71
72/// Local event delivered by a notifier.
73#[derive(Debug, Clone, PartialEq, Eq)]
74pub enum ConfigChangeEvent {
75    /// Receivers should reload from storage.
76    Reload(ConfigReloadMessage),
77}
78
79impl ConfigChangeEvent {
80    /// Returns the reload message carried by this event.
81    #[must_use]
82    pub const fn reload_message(&self) -> &ConfigReloadMessage {
83        match self {
84            Self::Reload(message) => message,
85        }
86    }
87}
88
89/// Result of handling one reload notification.
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91pub enum ConfigReloadDecision {
92    /// The notification matched this process and triggered a reload.
93    Reloaded,
94    /// The notification belongs to another product namespace.
95    IgnoredNamespace,
96    /// The notification came from this process and should not be replayed.
97    IgnoredOrigin,
98}
99
100impl ConfigReloadDecision {
101    /// Returns the stable metrics label for this decision.
102    #[must_use]
103    pub const fn as_label(self) -> &'static str {
104        match self {
105            Self::Reloaded => "reloaded",
106            Self::IgnoredNamespace => "ignored_namespace",
107            Self::IgnoredOrigin => "ignored_origin",
108        }
109    }
110}
111
112/// Runtime reload worker configuration.
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub struct ConfigReloadWorkerConfig {
115    /// Product or service namespace accepted by this worker.
116    pub namespace: String,
117    /// Runtime instance ID for the current process.
118    pub runtime_id: String,
119}
120
121impl ConfigReloadWorkerConfig {
122    /// Creates a worker config.
123    pub fn new(namespace: impl Into<String>, runtime_id: impl Into<String>) -> Self {
124        Self {
125            namespace: namespace.into(),
126            runtime_id: runtime_id.into(),
127        }
128    }
129
130    /// Returns whether a message belongs to this worker namespace.
131    #[must_use]
132    pub fn accepts_namespace(&self, message: &ConfigReloadMessage) -> bool {
133        message.namespace == self.namespace
134    }
135
136    /// Returns whether a message was emitted by this process.
137    #[must_use]
138    pub fn is_local_origin(&self, message: &ConfigReloadMessage) -> bool {
139        message.origin_runtime_id == self.runtime_id
140    }
141}
142
143/// Decodes one transport payload into a config reload event.
144///
145/// Transport adapters should use this helper before forwarding data into the common notifier path.
146/// Malformed payloads are returned as errors so listeners can log and continue instead of ending the
147/// subscription loop.
148///
149/// # Errors
150///
151/// Returns [`ConfigError`] when the transport payload is malformed or fails message validation.
152pub fn decode_config_reload_transport_payload(payload: &str) -> Result<ConfigChangeEvent> {
153    ConfigReloadMessage::decode(payload).map(ConfigChangeEvent::Reload)
154}
155
156/// Handles one reload notification by filtering namespace/origin and invoking `reload`.
157///
158/// # Errors
159///
160/// Returns [`ConfigError`] when notification decoding or the selected reload callback fails.
161pub async fn handle_config_reload_notification<F, Fut>(
162    config: &ConfigReloadWorkerConfig,
163    message: ConfigReloadMessage,
164    reload: F,
165) -> Result<ConfigReloadDecision>
166where
167    F: FnOnce(ConfigReloadMessage) -> Fut,
168    Fut: Future<Output = Result<()>>,
169{
170    if !config.accepts_namespace(&message) {
171        return Ok(ConfigReloadDecision::IgnoredNamespace);
172    }
173    if config.is_local_origin(&message) {
174        return Ok(ConfigReloadDecision::IgnoredOrigin);
175    }
176
177    reload(message).await?;
178    Ok(ConfigReloadDecision::Reloaded)
179}
aster_forge_config/notification/message.rs

aster_forge_config/notification/
message.rs
