#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
mod clock;
mod handlers;
mod locale;
mod message;
mod notify;
mod store;
mod webhook;
pub use handlers::{
ChannelPatch, NO_APPLICATION_SERVER_KEY, PreferencesBody, REHOME_LIMIT, RecipientBody,
RegisterBody, UNKNOWN_CATEGORY, WEBHOOK_UNVERIFIED,
};
pub use locale::EVENT_MISSING_TRANSLATION;
pub use message::{Localizable, Message, RenderMode};
pub use notify::{
DrainReport, EVENT_REQUESTED, EVENT_SUBSCRIPTION_PRUNED, EVENT_SUBSCRIPTION_REHOMED, Enqueued,
Notifier, NotifyError, Skipped, TOPIC_SEND,
};
pub use store::{DeadLetterReason, Transport};
use std::sync::{Arc, OnceLock};
use cratefield_core::{
AnyError, BoxFuture, Config, ConfigError, DataKind, Disposition, Migrations, Module,
ModuleConfig, ModuleContext, NoopDefer, Notification, PersonalDataSet, Port, RoutePolicy,
Scope, SqlMigration, VentureEnv,
};
use cratefield_i18n::{BODY, Catalog, LanguageIdentifier, TITLE};
pub const MODULE_NAME: &str = "notifications";
pub type TransportProbe = Arc<dyn Fn(&dyn Config) -> bool + Send + Sync>;
pub type VapidKeyProbe = Arc<dyn Fn(&dyn Config) -> Option<String> + Send + Sync>;
const MIGRATION_INIT: SqlMigration = SqlMigration::new(
"0001",
"init",
include_str!("../migrations/sqlite/0001_init.sql"),
);
const MIGRATION_REHOME_AND_DUE_INDEX: SqlMigration = SqlMigration::new(
"0002",
"rehome_and_due_index",
include_str!("../migrations/sqlite/0002_rehome_and_due_index.sql"),
);
const MIGRATION_INBOX: SqlMigration = SqlMigration::new(
"0003",
"inbox",
include_str!("../migrations/sqlite/0003_inbox.sql"),
);
const MIGRATION_EMAIL_TARGETS: SqlMigration = SqlMigration::new(
"0004",
"email_targets",
include_str!("../migrations/sqlite/0004_email_targets.sql"),
);
const MIGRATION_EMAIL_BOUNCE_INDEX: SqlMigration = SqlMigration::new(
"0005",
"email_bounce_index",
include_str!("../migrations/sqlite/0005_email_bounce_index.sql"),
);
const MIGRATION_LOCALES: SqlMigration = SqlMigration::new(
"0006",
"locales",
include_str!("../migrations/sqlite/0006_locales.sql"),
);
const MIGRATION_DEAD_LETTER_ACCOUNT: SqlMigration = SqlMigration::new(
"0007",
"dead_letter_account",
include_str!("../migrations/sqlite/0007_dead_letter_account.sql"),
);
const MIGRATION_EMAIL_SUPPRESSED: SqlMigration = SqlMigration::new(
"0008",
"email_suppressed",
include_str!("../migrations/sqlite/0008_email_suppressed.sql"),
);
const MIGRATION_OUTBOX_SUBJECT: SqlMigration = SqlMigration::new(
"0009",
"outbox_subject",
include_str!("../migrations/sqlite/0009_outbox_subject.sql"),
);
const SHIPPED_MIGRATIONS: [SqlMigration; 9] = [
MIGRATION_INIT,
MIGRATION_REHOME_AND_DUE_INDEX,
MIGRATION_INBOX,
MIGRATION_EMAIL_TARGETS,
MIGRATION_EMAIL_BOUNCE_INDEX,
MIGRATION_LOCALES,
MIGRATION_DEAD_LETTER_ACCOUNT,
MIGRATION_EMAIL_SUPPRESSED,
MIGRATION_OUTBOX_SUBJECT,
];
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Category {
pub(crate) name: String,
pub(crate) badge: bool,
pub(crate) defaults: crate::store::Channels,
pub(crate) subject_template: Option<String>,
pub(crate) render: RenderMode,
}
impl Category {
#[must_use]
pub fn new(name: impl Into<String>) -> Self {
Self {
name: name.into(),
badge: false,
defaults: crate::store::Channels {
push: true,
in_app: true,
email: false,
},
subject_template: None,
render: RenderMode::Server,
}
}
#[must_use]
pub fn default_enabled(mut self, enabled: bool) -> Self {
self.defaults.push = enabled;
self
}
#[must_use]
pub fn badge(mut self, allowed: bool) -> Self {
self.badge = allowed;
self
}
#[must_use]
pub fn in_app(mut self, keeps_a_row: bool) -> Self {
self.defaults.in_app = keeps_a_row;
self
}
#[must_use]
pub fn email(mut self, sends_mail: bool) -> Self {
self.defaults.email = sends_mail;
self
}
#[must_use]
pub fn subject_template(mut self, subject: impl Into<String>) -> Self {
self.subject_template = Some(subject.into());
self
}
#[must_use]
pub fn render(mut self, mode: RenderMode) -> Self {
self.render = mode;
self
}
#[must_use]
pub fn name(&self) -> &str {
&self.name
}
}
#[derive(Clone)]
pub(crate) struct Settings {
pub categories: Arc<Vec<Category>>,
pub max_attempts: u32,
pub drain_batch: u64,
pub drain_concurrency: u32,
pub rehome_max_per_hour: u32,
pub inbox_retention_days: u32,
pub email_window_secs: u32,
pub email_max_per_window: u32,
pub transport_probe: Option<TransportProbe>,
pub mailer_probe: Option<TransportProbe>,
pub vapid_key_probe: Option<VapidKeyProbe>,
pub catalog: Option<Arc<dyn Catalog>>,
pub default_locale: Option<LanguageIdentifier>,
pub messages: Arc<Vec<String>>,
}
impl std::fmt::Debug for Settings {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Settings")
.field("categories", &self.categories)
.field("max_attempts", &self.max_attempts)
.field("drain_batch", &self.drain_batch)
.field("drain_concurrency", &self.drain_concurrency)
.field("rehome_max_per_hour", &self.rehome_max_per_hour)
.field("inbox_retention_days", &self.inbox_retention_days)
.field("email_window_secs", &self.email_window_secs)
.field("email_max_per_window", &self.email_max_per_window)
.field("transport_probe", &self.transport_probe.is_some())
.field("mailer_probe", &self.mailer_probe.is_some())
.field("vapid_key_probe", &self.vapid_key_probe.is_some())
.field(
"locales",
&self.catalog.as_ref().map(|catalog| {
catalog
.locales()
.iter()
.map(ToString::to_string)
.collect::<Vec<_>>()
}),
)
.field("messages", &self.messages.len())
.finish_non_exhaustive()
}
}
impl Settings {
pub(crate) fn venture_locale(&self) -> LanguageIdentifier {
if let Some(locale) = &self.default_locale {
return locale.clone();
}
if let Some(catalog) = &self.catalog {
return catalog.default_locale().clone();
}
crate::locale::FALLBACK_LOCALE
.parse()
.unwrap_or_else(|_| LanguageIdentifier::default())
}
pub(crate) fn configured_locale(&self, ctx: &ModuleContext) -> LanguageIdentifier {
ModuleConfig::new(MODULE_NAME, &*ctx.config)
.get_opt("DEFAULT_LOCALE")
.as_deref()
.and_then(cratefield_i18n::parse_locale)
.unwrap_or_else(|| self.venture_locale())
}
pub(crate) fn category(&self, name: &str) -> Option<&Category> {
self.categories
.iter()
.find(|category| category.name == name)
}
pub(crate) fn declares(&self, name: &str) -> bool {
self.category(name).is_some()
}
}
pub struct Notifications {
settings: Settings,
ctx_cell: Arc<OnceLock<Arc<ModuleContext>>>,
settings_cell: Arc<OnceLock<Settings>>,
auth_cell: Arc<OnceLock<Option<Arc<cratefield_auth_client::AuthClient>>>>,
vapid_key_cell: Arc<OnceLock<Option<String>>>,
}
impl Default for Notifications {
fn default() -> Self {
Self::new()
}
}
impl Notifications {
#[must_use]
pub fn new() -> Self {
Self {
settings: Settings {
categories: Arc::new(Vec::new()),
max_attempts: 5,
drain_batch: 50,
drain_concurrency: 8,
rehome_max_per_hour: 3,
inbox_retention_days: 90,
email_window_secs: 3_600,
email_max_per_window: 5,
transport_probe: None,
mailer_probe: None,
vapid_key_probe: None,
catalog: None,
default_locale: None,
messages: Arc::new(Vec::new()),
},
ctx_cell: Arc::new(OnceLock::new()),
settings_cell: Arc::new(OnceLock::new()),
auth_cell: Arc::new(OnceLock::new()),
vapid_key_cell: Arc::new(OnceLock::new()),
}
}
#[must_use]
pub fn category(mut self, category: Category) -> Self {
Arc::make_mut(&mut self.settings.categories).push(category);
self
}
#[must_use]
pub fn categories(mut self, names: impl IntoIterator<Item = impl Into<String>>) -> Self {
let categories = Arc::make_mut(&mut self.settings.categories);
for name in names {
categories.push(Category::new(name));
}
self
}
#[must_use]
pub fn max_attempts(mut self, attempts: u32) -> Self {
self.settings.max_attempts = attempts;
self
}
#[must_use]
pub fn drain_batch(mut self, rows: u64) -> Self {
self.settings.drain_batch = rows;
self
}
#[must_use]
pub fn drain_concurrency(mut self, rows: u32) -> Self {
self.settings.drain_concurrency = rows;
self
}
#[must_use]
pub fn rehome_max_per_hour(mut self, devices: u32) -> Self {
self.settings.rehome_max_per_hour = devices;
self
}
#[must_use]
pub fn inbox_retention_days(mut self, days: u32) -> Self {
self.settings.inbox_retention_days = days;
self
}
#[must_use]
pub fn email_max_per_window(mut self, mails: u32) -> Self {
self.settings.email_max_per_window = mails;
self
}
#[must_use]
pub fn email_window_secs(mut self, seconds: u32) -> Self {
self.settings.email_window_secs = seconds;
self
}
#[must_use]
pub fn catalog(mut self, catalog: impl Catalog + 'static) -> Self {
self.settings.catalog = Some(Arc::new(catalog));
self
}
#[must_use]
pub fn shared_catalog(mut self, catalog: Arc<dyn Catalog>) -> Self {
self.settings.catalog = Some(catalog);
self
}
#[must_use]
pub fn default_locale(mut self, tag: &str) -> Self {
self.settings.default_locale = Some(
cratefield_i18n::parse_locale(tag)
.unwrap_or_else(|| panic!("notifications: {tag:?} is not a BCP 47 language tag")),
);
self
}
#[must_use]
pub fn messages(mut self, ids: impl IntoIterator<Item = impl Into<String>>) -> Self {
let messages = Arc::make_mut(&mut self.settings.messages);
for id in ids {
messages.push(id.into());
}
self
}
#[must_use]
pub fn missing_translations(&self) -> Vec<String> {
let Some(catalog) = self.settings.catalog.as_ref() else {
return Vec::new();
};
let keys: Vec<&str> = self.settings.messages.iter().map(String::as_str).collect();
cratefield_i18n::missing_messages(&**catalog, &keys, &[TITLE, BODY])
.iter()
.map(ToString::to_string)
.collect()
}
fn probe_problems(&self, cfg: &dyn Config, errors: &mut ConfigError) {
if self
.settings
.transport_probe
.as_ref()
.is_some_and(|probe| !probe(cfg))
{
errors.push(
"notifications: production, but this venture wired no push transport: every \
send would dead-letter as `not_configured`. `fz doctor` names the variables"
.to_owned(),
);
}
let emails = self
.settings
.categories
.iter()
.filter(|category| category.defaults.email)
.map(|category| category.name.clone())
.collect::<Vec<_>>();
if !emails.is_empty()
&& self
.settings
.mailer_probe
.as_ref()
.is_some_and(|probe| !probe(cfg))
{
errors.push(format!(
"notifications: production, and {} opted into email, but this venture wired no \
Mailer port: every one of those would dead-letter as `not_configured`. \
Configure the mailer, or drop `.email(true)` from those categories",
emails.join(", "),
));
}
}
#[must_use]
pub fn mailer_probe(
mut self,
probe: impl Fn(&dyn Config) -> bool + Send + Sync + 'static,
) -> Self {
self.settings.mailer_probe = Some(Arc::new(probe));
self
}
#[must_use]
pub fn transport_probe(
mut self,
probe: impl Fn(&dyn Config) -> bool + Send + Sync + 'static,
) -> Self {
self.settings.transport_probe = Some(Arc::new(probe));
self
}
#[must_use]
pub fn vapid_public_key(
mut self,
probe: impl Fn(&dyn Config) -> Option<String> + Send + Sync + 'static,
) -> Self {
self.settings.vapid_key_probe = Some(Arc::new(probe));
self
}
#[must_use]
pub fn notifier(&self) -> Notifier {
Notifier {
cell: Arc::clone(&self.ctx_cell),
settings_cell: Arc::clone(&self.settings_cell),
settings_at_handout: self.settings.clone(),
}
}
#[must_use]
pub fn declared_categories(&self) -> &[Category] {
&self.settings.categories
}
}
fn is_category_name(name: &str) -> bool {
!name.is_empty()
&& name
.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
}
impl Module for Notifications {
fn name(&self) -> &'static str {
MODULE_NAME
}
fn version(&self) -> &'static str {
env!("CARGO_PKG_VERSION")
}
fn requires(&self) -> &'static [Port] {
&[Port::Db, Port::Push, Port::Clock, Port::IdGen]
}
fn optional(&self) -> &'static [Port] {
&[
Port::Defer,
Port::HttpClient,
Port::Realtime,
Port::Mailer,
Port::Signer,
]
}
fn tables(&self) -> &'static [&'static str] {
&[
store::SUBSCRIPTIONS,
store::PREFERENCES,
store::OUTBOX,
store::DEAD_LETTERS,
store::INBOX,
store::EMAIL_TARGETS,
store::EMAIL_SENDS,
store::LOCALES,
store::EMAIL_SUPPRESSED,
]
}
fn personal_data(&self) -> &'static [PersonalDataSet] {
const SETS: &[PersonalDataSet] = &[
PersonalDataSet {
table: store::SUBSCRIPTIONS,
subject: "account_id",
kind: DataKind::Identifier,
disposition: Disposition::Erase,
description: "Each phone or browser you turned notifications on for, with the \
address the notification is delivered to, the language that device \
is set to, and when it last checked in.",
redacted: &["recipient_json"],
subject_via: None,
},
PersonalDataSet {
table: store::PREFERENCES,
subject: "account_id",
kind: DataKind::Usage,
disposition: Disposition::Erase,
description: "Which kinds of notification you have switched on or off, for each \
of the three ways we can reach you.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::LOCALES,
subject: "account_id",
kind: DataKind::Usage,
disposition: Disposition::Erase,
description: "The language you asked to be written to in.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::INBOX,
subject: "account_id",
kind: DataKind::Content,
disposition: Disposition::Erase,
description: "The notifications you can read back in the app: what each one said, \
when it arrived, and when you opened or archived it.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::EMAIL_TARGETS,
subject: "account_id",
kind: DataKind::Contact,
disposition: Disposition::Erase,
description: "The email address notifications are sent to, whether it has been \
confirmed, and whether you have unsubscribed from them.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::EMAIL_SENDS,
subject: "account_id",
kind: DataKind::Usage,
disposition: Disposition::Erase,
description: "When we last emailed you about each kind of notification, which is \
what stops us emailing you about it again too soon.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::DEAD_LETTERS,
subject: "account_id",
kind: DataKind::Content,
disposition: Disposition::Erase,
description: "A notification we gave up trying to deliver, kept with what it said \
so somebody can find out why it failed.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::EMAIL_SUPPRESSED,
subject: "account_id",
kind: DataKind::Usage,
disposition: Disposition::Erase,
description: "Which notifications we held back from your inbox because too many \
arrived at once, kept only until they are gathered into one \
summary email.",
redacted: &[],
subject_via: None,
},
PersonalDataSet {
table: store::OUTBOX,
subject: "subject",
kind: DataKind::Content,
disposition: Disposition::Erase,
description: "A notification waiting to be sent, holding the message it was \
queued with.",
redacted: &[],
subject_via: None,
},
];
SETS
}
fn emits(&self) -> &'static [&'static str] {
&[
notify::EVENT_SUBSCRIPTION_PRUNED,
notify::EVENT_SUBSCRIPTION_REHOMED,
locale::EVENT_MISSING_TRANSLATION,
]
}
fn self_check(&self) -> Vec<String> {
self.missing_translations()
.into_iter()
.map(|missing| {
format!(
"notifications: no translation for {missing} — every message listed in \
`.messages(..)` needs a `.title` and a `.body` in every configured locale"
)
})
.collect()
}
fn public_writes(&self) -> bool {
true
}
fn public_write_policy(&self) -> RoutePolicy {
RoutePolicy::SignedLink
}
fn migrations(&self) -> Migrations {
const _: () = cratefield_core::assert_migration_set(&SHIPPED_MIGRATIONS);
Migrations {
sqlite: &SHIPPED_MIGRATIONS,
postgres: &[],
}
}
fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
let module = ModuleConfig::new(MODULE_NAME, cfg);
let mut errors = ConfigError::default();
if self.settings.categories.is_empty() {
errors.push(
"notifications: no categories are declared, so every notify() is an unknown \
category — declare at least one with .category(..)"
.to_owned(),
);
}
let mut seen: Vec<&str> = Vec::new();
for category in self.settings.categories.iter() {
if !is_category_name(&category.name) {
errors.push(format!(
"notifications: category {:?} must be lower-case letters, digits and \
underscores",
category.name
));
}
if seen.contains(&category.name.as_str()) {
errors.push(format!(
"notifications: category {:?} is declared twice",
category.name
));
}
seen.push(&category.name);
}
for (key, minimum) in [
("MAX_ATTEMPTS", 1u32),
("DRAIN_BATCH", 1),
("DRAIN_CONCURRENCY", 1),
("REHOME_MAX_PER_HOUR", 0),
] {
if let Some(raw) = module.get_opt(key) {
let valid = raw.parse::<u32>().is_ok_and(|value| value >= minimum);
if !valid {
errors.push(format!(
"notifications: {} must be an integer between {minimum} and {}, got \
{raw:?}",
module.key(key),
u32::MAX
));
}
}
}
let issuer = module.get_opt("AUTH_ISSUER");
let client_id = module.get_opt("AUTH_CLIENT_ID");
if let Some(issuer) = &issuer
&& !(issuer.starts_with("https://") || issuer.starts_with("http://"))
{
errors.push(format!(
"notifications: {} must be an absolute http(s) URL, got {issuer:?}",
module.key("AUTH_ISSUER")
));
}
if issuer.is_some() != client_id.is_some() {
errors.push(format!(
"notifications: set {} and {} together — one without the other cannot verify a \
token, so every route would answer 401",
module.key("AUTH_ISSUER"),
module.key("AUTH_CLIENT_ID"),
));
}
if let Some(raw) = module.get_opt("DEFAULT_LOCALE")
&& cratefield_i18n::parse_locale(&raw).is_none()
{
errors.push(format!(
"notifications: {} is not a BCP 47 language tag, so every recipient with no \
language of their own would silently get the compiled-in default instead",
module.key("DEFAULT_LOCALE")
));
}
for problem in self.self_check() {
errors.push(problem);
}
let env = cfg
.get("ENV")
.as_deref()
.and_then(VentureEnv::parse)
.unwrap_or_default();
if env == VentureEnv::Production {
if issuer.is_none() {
errors.push(format!(
"notifications: {} and {} are required in production: without them no \
account can register a device",
module.key("AUTH_ISSUER"),
module.key("AUTH_CLIENT_ID"),
));
}
self.probe_problems(cfg, &mut errors);
}
errors.into_result()
}
fn router(&self, ctx: ModuleContext) -> axum::Router {
let shared = Arc::new(ctx);
let _ = self.ctx_cell.set(Arc::clone(&shared));
let _ = self.settings_cell.set(self.settings.clone());
let auth = self
.auth_cell
.get_or_init(|| handlers::auth_client(&shared))
.clone();
let vapid_public_key = self
.vapid_key_cell
.get_or_init(|| {
self.settings
.vapid_key_probe
.as_ref()
.and_then(|probe| probe(&*shared.config))
})
.clone();
handlers::router(Arc::new(handlers::ModuleState {
ctx: shared,
settings: self.settings.clone(),
auth,
vapid_public_key,
}))
}
fn events(&self) -> Vec<(cratefield_core::EventName, cratefield_core::EventHandler)> {
let notifier = self.notifier();
let handler: cratefield_core::EventHandler = Arc::new(move |scope, payload| {
let notifier = notifier.clone();
let scope = scope.clone();
Box::pin(async move { on_requested(¬ifier, &scope, payload).await })
});
vec![(notify::EVENT_REQUESTED.to_owned(), handler)]
}
fn scheduled<'a>(
&'a self,
ctx: &'a ModuleContext,
cron: &'a str,
) -> BoxFuture<'a, Result<(), AnyError>> {
Box::pin(async move {
let scope = scheduled_scope(ctx);
let report = self
.notifier()
.drain_with(ctx, &scope)
.await
.map_err(|err| Box::new(err) as AnyError)?;
if report.claimed > 0 {
tracing::info!(?report, cron, "drained the notifications outbox");
}
let days = ModuleConfig::new(MODULE_NAME, &*ctx.config)
.get_u32("INBOX_RETENTION_DAYS", self.settings.inbox_retention_days);
if days > 0
&& let Some(db) = ctx.ports.db.as_ref()
{
let cutoff = clock::plus_secs(
&clock::now_iso(ctx.ports.clock.as_ref()),
-(i64::from(days) * 86_400),
);
if let Err(err) = store::prune_email_sends(&**db, &cutoff).await {
tracing::error!(error = %err, "pruning the email send window failed");
}
if let Err(err) = store::prune_email_suppressed(&**db, &cutoff).await {
tracing::error!(error = %err, "pruning suppressed email rows failed");
}
match store::prune_inbox(&**db, &cutoff).await {
Ok(pruned) if pruned > 0 => {
tracing::info!(pruned, cron, "pruned read notifications past retention");
}
Ok(_) => {}
Err(err) => {
tracing::error!(error = %err, "pruning the notifications inbox failed");
}
}
}
Ok(())
})
}
}
fn scheduled_scope(ctx: &ModuleContext) -> Scope {
Scope {
request_id: format!("scheduled-{MODULE_NAME}"),
defer: ctx
.ports
.defer
.clone()
.unwrap_or_else(|| Arc::new(NoopDefer)),
span: tracing::Span::none(),
}
}
async fn on_requested(
notifier: &Notifier,
scope: &Scope,
payload: serde_json::Value,
) -> Result<(), AnyError> {
use serde_json::Value;
let Some(account_id) = payload.get("account_id").and_then(Value::as_str) else {
return Err("notifications.requested without an account_id".into());
};
let Some(category) = payload.get("category").and_then(Value::as_str) else {
return Err("notifications.requested without a category".into());
};
let notification: Notification = match payload.get("notification") {
Some(value) => serde_json::from_value(value.clone())
.map_err(|err| format!("notifications.requested carries no Notification: {err}"))?,
None => return Err("notifications.requested without a notification".into()),
};
let ctx = notifier.context().ok_or("notifications is not mounted")?;
let db = ctx
.ports
.db
.clone()
.ok_or("notifications has no Database port")?;
notifier
.notify_now(&*db, scope, account_id, category, notification)
.await
.map_err(|err| Box::new(err) as AnyError)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn module() -> Notifications {
Notifications::new().categories(["booking", "room_starting"])
}
#[test]
fn the_declaration_matches_the_issue() {
let module = module();
assert_eq!(module.name(), "notifications");
assert_eq!(
module.requires(),
[Port::Db, Port::Push, Port::Clock, Port::IdGen]
);
assert_eq!(
module.optional(),
[
Port::Defer,
Port::HttpClient,
Port::Realtime,
Port::Mailer,
Port::Signer,
]
);
assert_eq!(
module.tables(),
[
"notifications_subscriptions",
"notifications_preferences",
"notifications_outbox",
"notifications_dead_letters",
"notifications_inbox",
"notifications_email_targets",
"notifications_email_sends",
"notifications_locales",
"notifications_email_suppressed",
]
);
assert_eq!(
module.emits(),
[
"notifications.subscription_pruned",
"notifications.subscription_rehomed",
"notifications.missing_translation",
]
);
assert!(
module.public_writes(),
"the one-click unsubscribe (#189) and the provider webhook (#233) are \
unauthenticated; left false, the production check never sees them"
);
assert_eq!(
module.public_write_policy(),
RoutePolicy::SignedLink,
"both are proved by an artifact this service issued, not by a CAPTCHA \
nobody is there to solve — and not by `Signature`, which asks for a \
Payments port this module has no use for"
);
assert!(
module.surface().is_empty(),
"ADR 0010 has no audience for a signed-in account; see the crate docs"
);
}
#[test]
fn a_category_is_on_with_no_badge_until_told_otherwise() {
let plain = Category::new("booking");
assert!(plain.defaults.push);
assert!(!plain.badge, "no badge counts by default");
assert!(!Category::new("x").default_enabled(false).defaults.push);
assert!(Category::new("x").badge(true).badge);
}
#[test]
fn config_validation_refuses_an_empty_or_duplicated_category_set() {
let empty = Notifications::new();
let err = empty
.validate_config(&cratefield_core::EmptyConfig)
.expect_err("no categories is a misconfiguration");
assert!(err.to_string().contains("no categories"), "{err}");
let dup = Notifications::new().categories(["booking", "booking"]);
let err = dup
.validate_config(&cratefield_core::EmptyConfig)
.expect_err("a duplicate category is a misconfiguration");
assert!(err.to_string().contains("declared twice"), "{err}");
let bad = Notifications::new().categories(["Room Starting"]);
let err = bad
.validate_config(&cratefield_core::EmptyConfig)
.expect_err("a category name is a slug");
assert!(err.to_string().contains("lower-case"), "{err}");
module()
.validate_config(&cratefield_core::EmptyConfig)
.expect("a declared, unique, well-named set is valid outside production");
}
#[test]
fn production_demands_an_issuer_and_a_transport() {
let cfg = cratefield_core::MapConfig::from_pairs([("ENV", "production")]);
let err = module()
.transport_probe(|_| false)
.validate_config(&cfg)
.expect_err("production without auth or a transport is a misconfiguration");
let text = err.to_string();
assert!(text.contains("NOTIFICATIONS_AUTH_ISSUER"), "{text}");
assert!(text.contains("wired no push transport"), "{text}");
let err = module()
.validate_config(&cfg)
.expect_err("the auth keys are still required");
assert!(
!err.to_string().contains("push transport"),
"a module with no probe must not claim a verdict it cannot reach: {err}"
);
let wired = cratefield_core::MapConfig::from_pairs([
("ENV", "production"),
("NOTIFICATIONS_AUTH_ISSUER", "https://auth.example.test"),
("NOTIFICATIONS_AUTH_CLIENT_ID", "client-x"),
]);
module()
.transport_probe(|_| true)
.validate_config(&wired)
.expect("a wired production deployment is valid");
}
#[test]
fn a_number_the_runtime_cannot_read_is_refused_by_validation() {
for key in [
"NOTIFICATIONS_MAX_ATTEMPTS",
"NOTIFICATIONS_DRAIN_BATCH",
"NOTIFICATIONS_DRAIN_CONCURRENCY",
] {
let cfg = cratefield_core::MapConfig::from_pairs([(key, "4294967296")]);
let err = module()
.validate_config(&cfg)
.expect_err("a value above u32::MAX is not the value the runtime would read");
assert!(err.to_string().contains(key), "{err}");
let zero = cratefield_core::MapConfig::from_pairs([(key, "0")]);
module()
.validate_config(&zero)
.expect_err("zero is not a usable batch, attempt count or concurrency");
}
module()
.validate_config(&cratefield_core::MapConfig::from_pairs([(
"NOTIFICATIONS_REHOME_MAX_PER_HOUR",
"0",
)]))
.expect("0 re-homes an hour is a policy, not a mistake");
}
#[test]
fn half_an_auth_configuration_is_refused() {
let cfg = cratefield_core::MapConfig::from_pairs([(
"NOTIFICATIONS_AUTH_ISSUER",
"https://auth.example.test",
)]);
let err = module()
.validate_config(&cfg)
.expect_err("an issuer with no client id can verify nothing");
assert!(err.to_string().contains("together"), "{err}");
let cfg = cratefield_core::MapConfig::from_pairs([
("NOTIFICATIONS_AUTH_ISSUER", "auth.example.test"),
("NOTIFICATIONS_AUTH_CLIENT_ID", "client-x"),
]);
let err = module()
.validate_config(&cfg)
.expect_err("an issuer without a scheme is not a URL");
assert!(err.to_string().contains("absolute http(s) URL"), "{err}");
}
#[test]
fn the_migration_ships_the_cores_outbox_ddl_verbatim() {
let core = cratefield_core::Outbox::new(store::OUTBOX).create_table_sql();
let pre_266 = core.replace(" subject TEXT,\n", "");
assert!(
MIGRATION_INIT.sql.contains(&pre_266),
"the migration no longer ships `Outbox::create_table_sql()` verbatim:\n{core}"
);
assert!(
MIGRATION_OUTBOX_SUBJECT
.sql
.contains("ALTER TABLE notifications_outbox ADD COLUMN subject TEXT;"),
"the subject column has to come from the shipped migration:\n{}",
MIGRATION_OUTBOX_SUBJECT.sql
);
}
#[test]
fn every_hot_read_has_an_index() {
for (index, table, column) in [
(
"notifications_outbox_due",
"notifications_outbox",
"next_attempt_at",
),
(
"notifications_subscriptions_by_account",
"notifications_subscriptions",
"account_id",
),
] {
let expected = format!("CREATE INDEX IF NOT EXISTS {index}\n ON {table} ({column})");
assert!(
SHIPPED_MIGRATIONS
.iter()
.any(|migration| migration.sql.contains(&expected)),
"some migration must index {table}.{column}:\n{expected}"
);
}
}
#[test]
fn the_migration_is_portable() {
for migration in SHIPPED_MIGRATIONS {
assert_eq!(
cratefield_core::lint_portable_sql(migration.sql),
Vec::new(),
"migration {} left the portable subset (ADR 0004)",
migration.id
);
assert_eq!(
cratefield_core::lint_card_data(migration.sql),
Vec::new(),
"migration {}",
migration.id
);
}
}
#[test]
fn sqlite_has_no_enum_so_a_closed_set_is_text_plus_check() {
assert!(
MIGRATION_INIT.sql.contains(
"transport TEXT NOT NULL CHECK (transport IN ('apns', 'fcm', 'webpush'))"
),
"the transport column must stay TEXT + CHECK"
);
let ddl: String = MIGRATION_INIT
.sql
.lines()
.filter(|line| !line.trim_start().starts_with("--"))
.collect::<Vec<_>>()
.join("\n")
.to_ascii_uppercase();
assert!(!ddl.contains("ENUM"), "SQLite has no ENUM:\n{ddl}");
assert!(
MIGRATION_INIT.sql.contains("reason TEXT NOT NULL CHECK ("),
"the dead-letter reason must stay TEXT + CHECK too"
);
}
#[test]
fn all_three_channel_switches_are_in_the_first_migration() {
for column in [
"push INTEGER NOT NULL",
"in_app INTEGER NOT NULL",
"email INTEGER NOT NULL",
] {
assert!(
MIGRATION_INIT.sql.contains(column),
"the first migration must create `{column}`"
);
}
}
}