#![doc = include_str!("../README.md")]
#![forbid(unsafe_code)]
mod handlers;
mod mail;
mod store;
pub use mail::{ConfirmMailData, WelcomeMailData, default_templates, themed_templates};
use cratefield_core::{
AnyError, BoxFuture, Config, ConfigError, DataKind, Disposition, IdGen, Migrations, Module,
ModuleConfig, ModuleContext, PersonalDataSet, Port, SqlMigration, UlidIdGen, normalize_email,
validation_error,
};
use std::sync::{Arc, OnceLock};
use handlers::Settings;
const MIGRATION_INIT: SqlMigration = SqlMigration::new(
"0001",
"init",
include_str!("../migrations/sqlite/0001_init.sql"),
);
const MIGRATION_SUBSCRIPTION_GENERATION: SqlMigration = SqlMigration::new(
"0002",
"subscription_generation",
include_str!("../migrations/sqlite/0002_subscription_generation.sql"),
);
const MIGRATION_UNSUBSCRIBE_TOKEN: SqlMigration = SqlMigration::new(
"0003",
"unsubscribe_token",
include_str!("../migrations/sqlite/0003_unsubscribe_token.sql"),
);
#[derive(Clone)]
pub struct EmailSignup {
settings: Settings,
ctx_cell: Arc<OnceLock<Arc<ModuleContext>>>,
}
impl Default for EmailSignup {
fn default() -> Self {
Self::new()
}
}
impl EmailSignup {
pub fn new() -> Self {
Self {
settings: Settings {
double_opt_in: true,
confirm_ttl_days: 7,
retention_days_pending: 30,
confirmed_redirect: None,
unsubscribed_redirect: None,
expired_redirect: None,
welcome_on_confirm: false,
subscribe_on_waitlist_confirm: false,
},
ctx_cell: Arc::new(OnceLock::new()),
}
}
#[must_use]
pub fn double_opt_in(mut self, enabled: bool) -> Self {
self.settings.double_opt_in = enabled;
self
}
#[must_use]
pub fn confirm_ttl_days(mut self, days: u32) -> Self {
self.settings.confirm_ttl_days = days;
self
}
#[must_use]
pub fn confirmed_redirect(mut self, url: impl Into<String>) -> Self {
self.settings.confirmed_redirect = Some(url.into());
self
}
#[must_use]
pub fn unsubscribed_redirect(mut self, url: impl Into<String>) -> Self {
self.settings.unsubscribed_redirect = Some(url.into());
self
}
#[must_use]
pub fn expired_redirect(mut self, url: impl Into<String>) -> Self {
self.settings.expired_redirect = Some(url.into());
self
}
#[must_use]
pub fn retention_days_pending(mut self, days: u32) -> Self {
self.settings.retention_days_pending = days;
self
}
#[must_use]
pub fn welcome_on_confirm(mut self, enabled: bool) -> Self {
self.settings.welcome_on_confirm = enabled;
self
}
#[must_use]
pub fn subscribe_on_waitlist_confirm(mut self, enabled: bool) -> Self {
self.settings.subscribe_on_waitlist_confirm = enabled;
self
}
}
impl Module for EmailSignup {
fn name(&self) -> &'static str {
"email-signup"
}
fn version(&self) -> &'static str {
env!("CARGO_PKG_VERSION")
}
fn requires(&self) -> &'static [Port] {
&[Port::Db, Port::Mailer, Port::Signer]
}
fn optional(&self) -> &'static [Port] {
&[Port::Captcha, Port::RateLimiter]
}
fn tables(&self) -> &'static [&'static str] {
&["subscribers"]
}
fn personal_data(&self) -> &'static [PersonalDataSet] {
const SETS: &[PersonalDataSet] = &[PersonalDataSet {
table: "subscribers",
subject: "id",
kind: DataKind::Contact,
disposition: Disposition::Erase,
description: "The email address you gave us and the one we matched it against, \
where you signed up from, the language you asked to be written to \
in, and when you signed up, confirmed and unsubscribed.",
redacted: &["unsubscribe_token"],
subject_via: None,
}];
SETS
}
fn emits(&self) -> &'static [&'static str] {
&[handlers::EVENT_CONFIRMED, handlers::EVENT_UNSUBSCRIBED]
}
fn public_writes(&self) -> bool {
true
}
fn migrations(&self) -> Migrations {
const MIGRATIONS: [SqlMigration; 3] = [
MIGRATION_INIT,
MIGRATION_SUBSCRIPTION_GENERATION,
MIGRATION_UNSUBSCRIBE_TOKEN,
];
const _: () = cratefield_core::assert_migration_set(&MIGRATIONS);
Migrations {
sqlite: &MIGRATIONS,
postgres: &[],
}
}
fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError> {
let module = ModuleConfig::new("email-signup", cfg);
let mut errors = ConfigError::default();
for key in ["CONFIRM_TTL_DAYS", "RETENTION_DAYS_PENDING"] {
match cfg.get(&module.key(key)).map(|raw| raw.parse::<u32>()) {
Some(Ok(value)) if value > 0 => {}
Some(_) => errors.push(format!(
"email-signup: {} must be a positive integer",
module.key(key)
)),
None => {}
}
}
for key in [
"CONFIRMED_REDIRECT",
"UNSUBSCRIBED_REDIRECT",
"EXPIRED_REDIRECT",
"API_BASE",
] {
if let Some(raw) = cfg.get(&module.key(key))
&& !(raw.starts_with("https://") || raw.starts_with("http://"))
{
errors.push(format!(
"email-signup: {} must be an absolute http(s) URL, got {raw:?}",
module.key(key)
));
}
}
for key in [
"DOUBLE_OPT_IN",
"WELCOME_ON_CONFIRM",
"SUBSCRIBE_ON_WAITLIST_CONFIRM",
] {
if let Some(raw) = cfg.get(&module.key(key))
&& !matches!(
raw.to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on" | "0" | "false" | "no" | "off"
)
{
errors.push(format!(
"email-signup: {} must be a boolean, got {raw:?}",
module.key(key)
));
}
}
if let Some(raw) = cfg.get(&module.key("FROM"))
&& !raw.contains('@')
{
errors.push(format!(
"email-signup: {} must be an address, got {raw:?}",
module.key("FROM")
));
}
errors.into_result()
}
fn router(&self, ctx: ModuleContext) -> axum::Router {
let shared = Arc::new(ctx);
let _ = self.ctx_cell.set(Arc::clone(&shared));
handlers::router(shared, self.settings.clone())
}
fn surface(&self) -> cratefield_core::Surface {
handlers::surface(&self.settings)
}
fn events(&self) -> Vec<(cratefield_core::EventName, cratefield_core::EventHandler)> {
if !self.settings.subscribe_on_waitlist_confirm {
return Vec::new();
}
let cell = Arc::clone(&self.ctx_cell);
let handler: cratefield_core::EventHandler = Arc::new(move |_scope, payload| {
let cell = Arc::clone(&cell);
Box::pin(async move { subscribe_waitlist_confirmed(&cell, payload).await })
});
vec![(handlers::WAITLIST_CONFIRMED.to_owned(), handler)]
}
fn scheduled<'a>(
&'a self,
ctx: &'a ModuleContext,
cron: &'a str,
) -> BoxFuture<'a, Result<(), AnyError>> {
Box::pin(async move {
let Some(db) = ctx.ports.db.clone() else {
return Ok(());
};
let cfg = ModuleConfig::new("email-signup", &*ctx.config);
let days = i64::from(cfg.get_u32(
"RETENTION_DAYS_PENDING",
self.settings.retention_days_pending,
));
let cutoff = handlers::iso_ago(days.saturating_mul(86_400));
let deleted = store::purge_pending_older_than(&*db, &cutoff)
.await
.map_err(|err| Box::new(err) as AnyError)?;
if deleted > 0 {
tracing::info!(deleted, cron, "purged stale pending subscribers");
}
Ok(())
})
}
}
async fn subscribe_waitlist_confirmed(
cell: &Arc<OnceLock<Arc<ModuleContext>>>,
payload: serde_json::Value,
) -> Result<(), AnyError> {
use serde_json::Value;
let Some(email) = payload.get("email").and_then(Value::as_str) else {
tracing::warn!("waitlist.confirmed without an email; ignored");
return Ok(());
};
let product = payload
.get("product")
.and_then(Value::as_str)
.unwrap_or("unknown");
let normalized = normalize_email(email);
if validation_error(&normalized).is_some() {
tracing::warn!(product, "waitlist.confirmed with an invalid email; ignored");
return Ok(());
}
let Some(ctx) = cell.get() else {
return Err("email-signup router was never built".into());
};
let Some(db) = ctx.ports.db.clone() else {
return Err("email-signup has no Database port".into());
};
if store::find_by_normalized_email(&*db, &normalized)
.await
.map_err(|err| Box::new(err) as AnyError)?
.is_some()
{
return Ok(());
}
let now = handlers::now_iso();
store::insert_row(
&*db,
&store::SubscriberRow {
id: UlidIdGen.ulid(),
generation: 1,
email: normalized.clone(),
email_normalized: normalized,
status: store::STATUS_CONFIRMED.to_owned(),
source: Some(format!("waitlist:{product}")),
locale: None,
confirmed_at: Some(now.clone()),
unsubscribed_at: None,
unsubscribe_token: None,
created_at: now.clone(),
updated_at: now,
},
)
.await
.map_err(|err| Box::new(err) as AnyError)?;
tracing::info!(product, "subscribed a confirmed waitlist address");
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn defaults_match_the_issue() {
let module = EmailSignup::new();
assert!(module.settings.double_opt_in);
assert_eq!(module.settings.confirm_ttl_days, 7);
assert_eq!(module.settings.retention_days_pending, 30);
assert!(!module.settings.welcome_on_confirm);
assert!(!module.settings.subscribe_on_waitlist_confirm);
assert_eq!(module.name(), "email-signup");
assert_eq!(module.tables(), ["subscribers"]);
assert!(module.public_writes());
assert_eq!(module.requires(), [Port::Db, Port::Mailer, Port::Signer]);
assert_eq!(module.optional(), [Port::Captcha, Port::RateLimiter]);
}
#[test]
fn instances_do_not_share_a_context_cell() {
let a = EmailSignup::new();
let b = EmailSignup::new();
assert!(!Arc::ptr_eq(&a.ctx_cell, &b.ctx_cell));
}
}