feat: email notification system for CL comments#2040
feat: email notification system for CL comments#2040genedna merged 1 commit intoweb3infra-foundation:mainfrom
Conversation
Signed-off-by: abby-ql <[email protected]>
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3e9726cbae
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| let settings = match self.get_user_settings(username).await? { | ||
| Some(s) => s, | ||
| None => return Ok(false), |
There was a problem hiding this comment.
Returning false when get_user_settings is None means recipients are skipped unless they already created a settings row, but this commit only creates rows in the notification-config user endpoints. In practice, CL authors/reviewers who never opened that page will never receive comment notifications, even though event types are default-enabled, so the new feature silently misses most intended recipients.
Useful? React with 👍 / 👎.
| Err(e) => { | ||
| tracing::warn!("Failed to initialize SMTP mailer, falling back to noop: {e:?}"); | ||
| Arc::new(NoopMailer) |
There was a problem hiding this comment.
Falling back to NoopMailer on SMTP init errors causes silent data loss: NoopMailer returns Ok(()), and the dispatcher marks jobs as sent on success, so queued notifications are permanently drained without delivery or retry. This is especially harmful during transient/configuration startup issues because operators lose all pending emails instead of preserving them until mail is healthy.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Pull request overview
Implements an email notification system for Change List (CL) comment events across the stack (schema + storage + trigger + background dispatcher), and exposes user-facing APIs to manage notification preferences.
Changes:
- Add notification center schema (event types, user settings/preferences, email job outbox) plus SeaORM entities and a
NotificationStorageimplementation. - Trigger
cl.comment.createdto enqueue email jobs for CL author/reviewers (excluding actor), and add a background dispatcher to send/retry via SMTP. - Add user APIs/models to list notification event types and get/update the current user’s notification config; add mail configuration + SMTP mailer module.
Reviewed changes
Copilot reviewed 30 out of 32 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| orion-server/src/scheduler.rs | Renames/adds queue test (currently doesn’t validate FIFO order). |
| mono/start-mono.sh | Minor script formatting change. |
| mono/src/server/http_server.rs | Spawns an email dispatcher background task; adds SMTP/Noop mailer wiring and shutdown handling. |
| mono/src/notification/triggers.rs | Implements CL comment trigger: recipient resolution + enqueue email jobs; includes tests. |
| mono/src/notification/mod.rs | Adds notification module exports. |
| mono/src/notification/dispatcher.rs | Implements background dispatcher to claim/send/retry email jobs; includes tests. |
| mono/src/main.rs | Updates module declarations/imports for notification/server wiring (currently problematic). |
| mono/src/lib.rs | Exposes notification module from the library crate. |
| mono/src/api/router/user_router.rs | Adds endpoints to list notification types and get/update notification config. |
| mono/src/api/router/cl_router.rs | Calls the notification trigger after saving a CL comment. |
| mono/src/api/mod.rs | Plumbs NotificationStorage access into API state. |
| mono/Cargo.toml | Adds tempfile dev-dependency for new tests. |
| jupiter/src/tests.rs | Wires NotificationStorage into test Storage. |
| jupiter/src/storage/notification_storage.rs | Adds storage layer for event types, user settings/preferences, and email job outbox + retry/claim logic. |
| jupiter/src/storage/mod.rs | Exposes NotificationStorage and includes it in Storage/AppService construction. |
| jupiter/src/migration/mod.rs | Registers notification-center migration and adds schema/constraint tests. |
| jupiter/src/migration/m20260224_230000_create_notification_center.rs | Adds DB migration creating notification-related tables and indices. |
| jupiter/callisto/src/user_notification_settings.rs | Adds SeaORM entity for user notification settings table. |
| jupiter/callisto/src/user_notification_preferences.rs | Adds SeaORM entity for per-event user preference overrides. |
| jupiter/callisto/src/prelude.rs | Exposes new notification entities in Callisto prelude. |
| jupiter/callisto/src/notification_event_types.rs | Adds SeaORM entity for notification event types table. |
| jupiter/callisto/src/mod.rs | Registers new Callisto modules/entities. |
| jupiter/callisto/src/email_jobs.rs | Adds SeaORM entity for email job outbox table. |
| config/config.toml | Adds [mail] configuration block. |
| common/src/lib.rs | Exports new common::email module. |
| common/src/email/mod.rs | Adds Mailer trait plus SMTP + noop implementations using lettre. |
| common/src/config/mod.rs | Adds optional [mail] config struct and deserialization test. |
| common/Cargo.toml | Adds async-trait and lettre dependencies. |
| ceres/src/model/notification.rs | Adds API models/schemas for notification types and user config payloads. |
| ceres/src/model/mod.rs | Exposes notification models module. |
| Cargo.toml | Adds workspace dependency configuration for lettre. |
| Cargo.lock | Locks new deps pulled in by lettre/tempfile additions. |
| .await | ||
| .unwrap(); | ||
|
|
||
| // SUppose the actor is carol, should notify alice and bob but not carol |
There was a problem hiding this comment.
Typo in test comment: SUppose → Suppose.
| // SUppose the actor is carol, should notify alice and bob but not carol | |
| // Suppose the actor is carol, should notify alice and bob but not carol |
| // Evnt types | ||
| pub async fn list_event_types( | ||
| &self, | ||
| ) -> Result<Vec<notification_event_types::Model>, sea_orm::DbErr> { | ||
| notification_event_types::Entity::find() |
There was a problem hiding this comment.
Typo in comment: Evnt types → Event types.
| starttls = false | ||
| smtp_tls = false | ||
| tls = false No newline at end of file |
There was a problem hiding this comment.
The [mail] section includes keys (smtp_tls, tls) that are not defined in common::config::MailConfig. Serde-based config deserialization will error on unknown fields, which can prevent the service from starting with this config. Either remove these keys from config.toml or add matching (defaulted) fields to MailConfig.
| starttls = false | |
| smtp_tls = false | |
| tls = false | |
| starttls = false |
| // Build a mailer (Default to NoopMailer if config missing or invalid) | ||
| let cfg = ctx.storage.config(); | ||
| let mailer: Arc<dyn Mailer> = if let Some(mail_cfg) = &cfg.mail { | ||
| match SmtpMailer::new(mail_cfg) { | ||
| Ok(m) => Arc::new(m), | ||
| Err(e) => { | ||
| tracing::warn!("Failed to initialize SMTP mailer, falling back to noop: {e:?}"); | ||
| Arc::new(NoopMailer) | ||
| } | ||
| } | ||
| } else { | ||
| Arc::new(NoopMailer) | ||
| }; | ||
|
|
There was a problem hiding this comment.
Falling back to NoopMailer means the dispatcher will report success and mark email jobs as sent even though no email was actually delivered (also when mail is disabled/misconfigured). This can silently drop notifications and lose the ability to retry once SMTP is fixed. Consider not starting the dispatcher when no real mailer is configured, or have the fallback path keep jobs pending/failed with a clear error instead of treating them as sent.
| // Build a mailer (Default to NoopMailer if config missing or invalid) | |
| let cfg = ctx.storage.config(); | |
| let mailer: Arc<dyn Mailer> = if let Some(mail_cfg) = &cfg.mail { | |
| match SmtpMailer::new(mail_cfg) { | |
| Ok(m) => Arc::new(m), | |
| Err(e) => { | |
| tracing::warn!("Failed to initialize SMTP mailer, falling back to noop: {e:?}"); | |
| Arc::new(NoopMailer) | |
| } | |
| } | |
| } else { | |
| Arc::new(NoopMailer) | |
| }; | |
| let cfg = ctx.storage.config(); | |
| // If mail configuration is missing, do not start the dispatcher. | |
| // This avoids silently "sending" emails via a noop implementation. | |
| let mail_cfg = if let Some(mail_cfg) = &cfg.mail { | |
| mail_cfg | |
| } else { | |
| tracing::info!( | |
| "Email dispatcher not started: mail configuration is missing or disabled" | |
| ); | |
| return tokio::spawn(async move { | |
| tracing::info!( | |
| "Email dispatcher idle task waiting for shutdown (mail disabled)" | |
| ); | |
| token.cancelled().await; | |
| tracing::info!( | |
| "Email dispatcher idle task exiting (mail disabled)" | |
| ); | |
| }); | |
| }; | |
| // Attempt to build a real SMTP mailer; if this fails, keep jobs pending | |
| // by not starting the dispatcher. | |
| let mailer: Arc<dyn Mailer> = match SmtpMailer::new(mail_cfg) { | |
| Ok(m) => Arc::new(m), | |
| Err(e) => { | |
| tracing::error!( | |
| "Failed to initialize SMTP mailer, email dispatcher will not start: {e:?}" | |
| ); | |
| return tokio::spawn(async move { | |
| tracing::warn!( | |
| "Email dispatcher idle task waiting for shutdown (mailer init failed)" | |
| ); | |
| token.cancelled().await; | |
| tracing::warn!( | |
| "Email dispatcher idle task exiting (mailer init failed)" | |
| ); | |
| }); | |
| } | |
| }; |
| pub mod git_protocol; | ||
| pub mod server; | ||
| pub mod notification; | ||
| use mono::server; |
There was a problem hiding this comment.
main.rs no longer declares the server module (it was replaced with use mono::server;). However, command handlers under mono/src/commands/service/* still import crate::server::..., which will fail to compile because crate::server no longer exists in the binary crate. Either restore pub mod server; here, or switch the binary to use the library crate modules consistently (e.g., call mono::cli::parse and update command modules to reference mono::server).
| use mono::server; | |
| pub mod server; |
| // Fire notification trigger | ||
| if let Err(e) = triggers::on_cl_comment_created( | ||
| &state.notification_stg(), | ||
| &state.cl_stg(), | ||
| &state.storage.reviewer_storage(), | ||
| &user.username, | ||
| &link, | ||
| &payload.content, | ||
| ) | ||
| .await |
There was a problem hiding this comment.
&state.notification_stg() takes a reference to a temporary NotificationStorage (the method returns an owned value). Because the call is awaited, this borrow will not live long enough and should not compile. Bind the storage to a local variable first (or change notification_stg() to return a reference/Arc) and pass a reference to that binding.
| async fn ensure_event_type_exists(stg: &NotificationStorage) -> Result<(), MegaError> { | ||
| if stg | ||
| .get_event_type(EVENT_CL_COMMENT_CREATED) | ||
| .await? | ||
| .is_some() | ||
| { | ||
| return Ok(()); | ||
| } | ||
|
|
||
| let now = chrono::Utc::now().naive_utc(); | ||
| notification_event_types::ActiveModel { | ||
| code: Set(EVENT_CL_COMMENT_CREATED.to_owned()), | ||
| category: Set("cl".to_owned()), | ||
| description: Set("New comment on a Change List".to_owned()), | ||
| system_required: Set(false), | ||
| default_enabled: Set(true), | ||
| created_at: Set(now), | ||
| updated_at: Set(now), | ||
| } | ||
| .insert(stg.db()) | ||
| .await?; |
There was a problem hiding this comment.
ensure_event_type_exists does a check-then-insert for notification_event_types. Under concurrent first-use (multiple comment events in parallel), this can race and the insert can fail with a unique/PK violation, causing the trigger to error. Consider using an UPSERT/ON CONFLICT DO NOTHING (or catch unique-violation and treat it as success) to make the seeding idempotent.
| /// Try to claim a job for sending by atomically changing its status from "pending" to "sending". | ||
| /// Returns true if the claim was successful, false if the job was already claimed by another | ||
| pub async fn try_claim_job(&self, job_id: i64) -> Result<bool, sea_orm::DbErr> { | ||
| let now = chrono::Utc::now().naive_utc(); | ||
| let res = email_jobs::Entity::update_many() | ||
| .col_expr(email_jobs::Column::Status, Expr::value("sending")) | ||
| .col_expr(email_jobs::Column::UpdatedAt, Expr::value(now)) | ||
| .filter(email_jobs::Column::Id.eq(job_id)) | ||
| .filter(email_jobs::Column::Status.eq("pending")) | ||
| .exec(self.db()) | ||
| .await?; | ||
| Ok(res.rows_affected == 1) | ||
| } |
There was a problem hiding this comment.
Jobs are transitioned to status = "sending" via try_claim_job, but there is no mechanism to recover jobs that get stuck in "sending" (e.g., process crash after claiming but before mark_job_sent/mark_job_failed_with_retry). Those jobs will never be fetched again by fetch_pending_jobs. Consider adding a lease/timeout (e.g., reclaim sending jobs older than N minutes) or avoiding the intermediate "sending" state.
| #[test] | ||
| fn test_queue_capacity_v2() { | ||
| fn test_task_queue_fifo() { | ||
| let config = TaskQueueConfig::default(); | ||
| let mut queue = TaskQueue::new(config); | ||
|
|
||
| let build_event1 = BuildEventPayload::new( | ||
| Uuid::now_v7(), | ||
| Uuid::now_v7(), | ||
| "test_cl_link".to_string(), | ||
| "/test/repo".to_string(), | ||
| 0, | ||
| ); | ||
|
|
||
| let task1 = PendingBuildEventV2 { | ||
| event_payload: build_event1.clone(), | ||
| targets: vec![], | ||
| changes: vec![], | ||
| created_at: Instant::now(), | ||
| }; | ||
|
|
||
| let build_event2 = BuildEventPayload::new( | ||
| Uuid::now_v7(), | ||
| Uuid::now_v7(), | ||
| "test_cl_link_2".to_string(), | ||
| "/test2/repo".to_string(), | ||
| 0, | ||
| ); | ||
|
|
||
| let task2 = PendingBuildEventV2 { | ||
| event_payload: build_event2.clone(), | ||
| targets: vec![], | ||
| changes: vec![], | ||
| created_at: Instant::now(), | ||
| }; | ||
|
|
||
| assert!(queue.enqueue_v2(task1).is_ok()); | ||
| assert!(queue.enqueue_v2(task2).is_ok()); | ||
|
|
||
| // check stats | ||
| let stats = queue.get_stats(); | ||
| assert_eq!(stats.total_queued, 2); | ||
| } |
There was a problem hiding this comment.
test_task_queue_fifo currently only asserts total_queued == 2 after enqueue, but it never verifies FIFO ordering. Either rename the test to reflect what it checks (queue length/stats) or extend it to dequeue/pop and assert the insertion order.
fix #1939
This PR implements an email notification system for Change List (CL) comments. It enables users to receive notifications when new comments are added to CLs they are involved in.
cl.comment.createdeventsOverview
Trigger layer (
triggers.rs)Determines recipients and enqueues email jobs
Storage layer (
NotificationStorage)Persists jobs in
email_jobswith status trackingDispatcher (
dispatcher.rs)Processes pending jobs and sends emails via SMTP
Testing
End-to-end testing was completed before merging upstream changes using a local SMTP debug server
Verified:
After merging upstream (OAuth / schema changes), full end-to-end testing is currently blocked by authentication requirements
Notes