fix: async newsletter emails, migration locking, webhook signature enforcement, DLQ metric

services/api/src/handlers.rs

@@ -320,7 +320,7 @@ pub async fn newsletter_subscribe(
     tracing::info!(request_id, email = %email, source = %source, ip = %ip, "newsletter subscription attempt");
 
     Ok((
-        StatusCode::OK,
+        StatusCode::ACCEPTED,
         Json(NewsletterResponse {
             success: true,
             message: "Please check your email to confirm your subscription.".to_string(),
@@ -975,6 +975,8 @@ pub async fn email_queue_stats(
         .await
         .map_err(into_api_error)?;
 
+    state.metrics.set_dlq_size(stats.dead_letter as i64);
+
     Ok((StatusCode::OK, Json(stats)))
 }
 

services/api/src/metrics.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use anyhow::Context;
-use prometheus::{Encoder, HistogramVec, IntCounterVec, Registry, TextEncoder};
+use prometheus::{Encoder, HistogramVec, IntCounterVec, IntGauge, Registry, TextEncoder};
 
 #[derive(Clone)]
 pub struct Metrics {
@@ -13,6 +13,7 @@ pub struct Metrics {
     rpc_errors: IntCounterVec,
     rpc_fallbacks: IntCounterVec,
     db_timeouts: IntCounterVec,
+    email_dlq_size: IntGauge,
 }
 
 impl Metrics {
@@ -70,13 +71,20 @@ impl Metrics {
         )
         .context("db_timeouts metric")?;
 
+        let email_dlq_size = IntGauge::new(
+            "email_dlq_size",
+            "Number of email jobs currently in the dead-letter queue",
+        )
+        .context("email_dlq_size metric")?;
+
         registry.register(Box::new(cache_hits.clone()))?;
         registry.register(Box::new(cache_misses.clone()))?;
         registry.register(Box::new(invalidations.clone()))?;
         registry.register(Box::new(request_latency.clone()))?;
         registry.register(Box::new(rpc_errors.clone()))?;
         registry.register(Box::new(rpc_fallbacks.clone()))?;
         registry.register(Box::new(db_timeouts.clone()))?;
+        registry.register(Box::new(email_dlq_size.clone()))?;
 
         Ok(Self {
             registry,
@@ -87,6 +95,7 @@ impl Metrics {
             rpc_errors,
             rpc_fallbacks,
             db_timeouts,
+            email_dlq_size,
         })
     }
 
@@ -126,6 +135,10 @@ impl Metrics {
         self.db_timeouts.with_label_values(&[operation]).inc();
     }
 
+    pub fn set_dlq_size(&self, n: i64) {
+        self.email_dlq_size.set(n);
+    }
+
     pub fn observe_tx_eviction(&self, count: u64) {
         if count > 0 {
             self.invalidations

services/api/src/migrations.rs

@@ -108,9 +108,49 @@ impl<'a> MigrationRunner<'a> {
     /// Ensure the tracking table exists, then apply every pending migration.
     /// Already-applied migrations are skipped. Returns the number of newly
     /// applied migrations.
+    ///
+    /// Uses a PostgreSQL session-level advisory lock to serialize concurrent
+    /// invocations (e.g. multiple instances starting simultaneously). If another
+    /// instance already holds the lock, this call aborts with an error so the
+    /// caller can surface it and halt startup cleanly.
     pub async fn run(&self) -> anyhow::Result<usize> {
         self.ensure_tracking_table().await?;
 
+        // Stable lock key — chosen to be unique to this codebase.
+        const MIGRATION_LOCK_KEY: i64 = 0x7072_6564_6963_7471_u64 as i64;
+
+        let mut lock_conn = self
+            .pool
+            .acquire()
+            .await
+            .context("acquire advisory lock connection")?;
+
+        let locked: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
+            .bind(MIGRATION_LOCK_KEY)
+            .fetch_one(&mut *lock_conn)
+            .await
+            .context("acquire migration advisory lock")?;
+
+        if !locked {
+            bail!(
+                "another instance holds the migration advisory lock — \
+                 aborting to prevent concurrent migration execution"
+            );
+        }
+
+        let result = self.run_inner().await;
+
+        // Always release the lock, even on failure, before the connection
+        // returns to the pool (session-level locks survive pool reuse).
+        let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
+            .bind(MIGRATION_LOCK_KEY)
+            .execute(&mut *lock_conn)
+            .await;
+
+        result
+    }
+
+    async fn run_inner(&self) -> anyhow::Result<usize> {
         let mut applied = 0usize;
 
         for migration in MIGRATIONS {

services/api/src/security.rs

@@ -468,7 +468,7 @@ pub async fn sendgrid_webhook_middleware(
 ) -> Result<Response, StatusCode> {
     let is_dev = std::env::var("ENVIRONMENT")
         .map(|e| e == "development")
-        .unwrap_or(true); // default to dev if not set
+        .unwrap_or(false); // default to non-dev so signature verification is enforced
 
     if config.secret.is_none() && !is_dev {
         return Err(StatusCode::UNAUTHORIZED);