1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
//! The facade a service actually calls.
//!
//! [`Notifier`] composes the three concerns a trigger site would otherwise re-implement:
//! resolving the audience, assembling the wire request, and choosing what a delivery
//! failure means. The last one is the reason there are three methods rather than one —
//! see [`Notifier::notify_detached`].
use std::sync::Arc;
use super::admin_topics::AdminTopics;
use super::audience::{AudienceError, AudienceResolver, RecipientRef};
use super::client::{NotificationClient, NotificationError};
use super::event::NotificationEvent;
use super::trigger::{Recipient, Severity, TopicRecipient, TriggerRequest, TriggerTarget};
/// A notification that did not go out.
#[derive(Debug, thiserror::Error)]
pub enum NotifyError {
#[error(transparent)]
Audience(#[from] AudienceError),
#[error(transparent)]
Trigger(#[from] NotificationError),
}
/// Sends [`NotificationEvent`]s. Build one per process and clone it — cloning shares the
/// HTTP connection pool and the resolver.
#[derive(Clone)]
pub struct Notifier {
client: NotificationClient,
audience: Arc<dyn AudienceResolver>,
admin_topics: AdminTopics,
}
impl Notifier {
/// Build from the trigger URL (the value of
/// [`TRIGGER_URL_ENV`](super::client::TRIGGER_URL_ENV)).
pub fn new(trigger_url: impl Into<String>, audience: Arc<dyn AudienceResolver>) -> Self {
Self::with_client(NotificationClient::new(trigger_url), audience)
}
/// Build from an existing client, to share a connection pool the service already has.
/// Internal channels use [`AdminTopics::from_env`]; override with
/// [`Self::with_admin_topics`].
pub fn with_client(client: NotificationClient, audience: Arc<dyn AudienceResolver>) -> Self {
Self {
client,
audience,
admin_topics: AdminTopics::from_env(),
}
}
/// Route internal channels with an explicit mapping instead of the environment's.
pub fn with_admin_topics(mut self, admin_topics: AdminTopics) -> Self {
self.admin_topics = admin_topics;
self
}
/// The whole stack from the environment, or `None` when this service is not set up to
/// notify: `NOTIFICATION_API_URL` for where to fire, plus `ACCOUNT_INTERNAL_GRPC_URL`
/// and `APIKEY_INTERNAL_SECRET` for resolving who to notify.
///
/// This is the one line a service needs in its container. It exists here rather than
/// in each service because there is nothing service-specific about it, and a copy per
/// service is exactly the duplication this module was created to remove.
///
/// Absence is never an error — notifications report on operations rather than perform
/// them, so a service must start and work without a notification center. The two
/// cases are logged differently on purpose: no trigger URL means notifications are
/// deliberately off in this environment and stays quiet, whereas a trigger URL with
/// no way to resolve recipients is a misconfiguration and says so.
#[cfg(feature = "notification-grpc")]
pub fn from_env() -> Option<Self> {
use super::client::TRIGGER_URL_ENV;
use super::grpc::{AccountAudienceResolver, ACCOUNT_GRPC_URL_ENV, INTERNAL_SECRET_ENV};
fn non_empty(key: &str) -> Option<String> {
std::env::var(key)
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
// Off by choice — nothing to report.
let trigger_url = non_empty(TRIGGER_URL_ENV)?;
let (Some(account_url), Some(secret)) = (
non_empty(ACCOUNT_GRPC_URL_ENV),
non_empty(INTERNAL_SECRET_ENV),
) else {
tracing::warn!(
"{TRIGGER_URL_ENV} is set but {ACCOUNT_GRPC_URL_ENV}/{INTERNAL_SECRET_ENV} \
are not; notification recipients cannot be resolved and nothing will be sent"
);
return None;
};
match AccountAudienceResolver::new(&account_url, &secret) {
Ok(resolver) => Some(Self::new(trigger_url, Arc::new(resolver))),
Err(error) => {
tracing::error!(
"the notification audience resolver could not be built; nothing will be sent: {error}"
);
None
}
}
}
/// Resolve the audience and fire the workflow, surfacing any failure.
///
/// Use this where the notification *is* the operation — an OTP the user is waiting
/// on, where "sent" and "not sent" are different outcomes for the caller.
pub async fn notify(
&self,
to: &RecipientRef,
event: &dyn NotificationEvent,
) -> Result<(), NotifyError> {
// An internal channel is a configuration lookup, not a directory lookup — there
// is no member behind it, so cex-account is never consulted and an outage there
// cannot stop an operations alert.
let target: TriggerTarget = match to {
RecipientRef::Admin(channel) => {
TopicRecipient::new(self.admin_topics.key(*channel)).into()
}
person => {
let audience = self.audience.resolve(person).await?;
Recipient::new(audience.subscriber_id, audience.locale).into()
}
};
let mut request = TriggerRequest::new(event.workflow(), target, event.payload());
// Only ride the wire when it differs from the center's own default, so a plain
// event serializes as plainly as a hand-built request.
let severity = event.severity();
if severity != Severity::default() {
request = request.with_severity(severity);
}
if let Some(transaction_id) = event.transaction_id() {
request = request.with_transaction_id(transaction_id);
}
self.client.trigger(&request).await?;
Ok(())
}
/// Fire the workflow, logging rather than returning a failure.
///
/// Use this where the notification *reports* an operation that already succeeded. A
/// notification-center outage must not turn a completed refund into an error the
/// caller has to handle, and there is nothing useful the caller could do with the
/// error anyway.
pub async fn notify_best_effort(&self, to: &RecipientRef, event: &dyn NotificationEvent) {
if let Err(error) = self.notify(to, event).await {
tracing::warn!(
workflow = event.workflow(),
recipient = %to.log_label(),
"notification not delivered: {error}"
);
}
}
/// Fire the workflow on a background task, returning immediately.
///
/// This is the right choice on a request path that has already committed: the
/// audience lookup plus the trigger can take seconds, and making an admin's
/// decision request wait on the notification center — after the money has already
/// moved — trades user-visible latency for nothing. Failures are logged exactly as
/// in [`Self::notify_best_effort`].
///
/// Requires a Tokio runtime, which every service in the platform has.
pub fn notify_detached(&self, to: RecipientRef, event: impl NotificationEvent + 'static) {
let notifier = self.clone();
tokio::spawn(async move { notifier.notify_best_effort(&to, &event).await });
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::notification::audience::{Audience, StaticAudienceResolver};
struct Refunded;
impl NotificationEvent for Refunded {
fn workflow(&self) -> &str {
"fiat-withdrawal-failed-refunded"
}
fn payload(&self) -> serde_json::Value {
serde_json::json!({ "amount": "10.00" })
}
fn transaction_id(&self) -> Option<String> {
Some("req-1".into())
}
}
/// The resolved audience — not anything the caller passed — is what reaches the wire.
#[tokio::test]
async fn a_trigger_carries_the_resolved_subscriber_and_locale() {
let resolver = StaticAudienceResolver::new(Audience::new("sub-7", "en"));
let audience = resolver
.resolve(&RecipientRef::AccountId(42))
.await
.unwrap();
let request = TriggerRequest::new(
Refunded.workflow(),
Recipient::new(audience.subscriber_id, audience.locale),
Refunded.payload(),
)
.with_transaction_id(Refunded.transaction_id().unwrap());
let value = serde_json::to_value(&request).unwrap();
assert_eq!(value["name"], "fiat-withdrawal-failed-refunded");
assert_eq!(value["to"]["subscriberId"], "sub-7");
assert_eq!(value["to"]["locale"], "en");
assert_eq!(value["transactionId"], "req-1");
// Normal severity is the center's default, so it stays off the wire.
assert!(value.get("overrides").is_none());
}
}