feat(notifications): email REQUESTED recipients immediately (per-job opt-in)

A recipient with emailStatus REQUESTED is emailed at once, without a
standing category preference or minimum severity, and only once even when
an IMMEDIATE preference also applies. No address: SKIPPED. ALERT
notifications are still never emailed here. This is the delivery half of
the durable-jobs "Email me when done" opt-in (QRun-IO/qqq durable jobs spec
2.11); marking recipients REQUESTED from core's NotificationSignal
attribute emailRequested lands with the RC.6 core notify() adoption.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to this project are documented here. The format follows Keep
 - Webhook signing delegates to core `WebhookSigner`/`WebhookSecret` (qbit-alerts#15, F11). Headers, `v1,` signatures and `whsec_` encoding are unchanged. A thin adapter preserves legacy keys shorter than core's 16-byte minimum using equivalent HMAC zero-padding.
 
 ### Added
+- Per-notification email opt-in via recipient email status `REQUESTED`: immediate delivery without a standing category preference, skipped when no address is known, and never emailed for ALERT notifications.
 - Optional external channels:
   - Slack (an Incoming Webhook, Block Kit)
   - Microsoft Teams (a Workflows webhook, an Adaptive Card)

src/main/java/com/kingsrook/qbits/alerts/model/AlertsPossibleValues.java

@@ -288,7 +288,7 @@ public final class AlertsPossibleValues
     *******************************************************************************/
    public enum EmailStatus
    {
-      NONE, PENDING, SENT, DIGESTED, SKIPPED, FAILED
+      NONE, PENDING, REQUESTED, SENT, DIGESTED, SKIPPED, FAILED
    }
 
 

src/main/java/com/kingsrook/qbits/alerts/notifications/NotificationEmailer.java

@@ -28,6 +28,7 @@ import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import com.kingsrook.qbits.alerts.AlertsQBitConfig;
 import com.kingsrook.qbits.alerts.AlertsRuntime;
 import com.kingsrook.qbits.alerts.engine.AlertsDb;
@@ -61,6 +62,8 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
  ** - IMMEDIATE: one email per notification at or above the minimum severity;
  ** - DAILY_DIGEST: one email at the user's digest hour (in their time zone,
  **   from My Alert Settings) listing the day's unread notifications.
+ ** An explicit REQUESTED recipient is emailed immediately without a standing
+ ** category preference or minimum severity.
  ** Inert without a messaging provider and a From address. ALERT
  ** notifications are never emailed here (alert subscriptions already email
  ** them). SECURITY emails carry the title and link only. Runs as the system
@@ -124,18 +127,25 @@ public class NotificationEmailer
          {
             preferences.computeIfAbsent(preference.getValueString("userId"), k -> new HashMap<>()).put(preference.getValueString("category"), preference);
          }
-         if(preferences.isEmpty())
+         QQueryFilter eligible = new QQueryFilter().withBooleanOperator(QQueryFilter.BooleanOperator.OR)
+            .withCriteria("emailStatus", QCriteriaOperator.EQUALS, EmailStatus.REQUESTED.name());
+         if(!preferences.isEmpty())
          {
-            return (new Result(0, 0, 0, 0));
+            eligible.withSubFilter(new QQueryFilter()
+               .withCriteria("userId", QCriteriaOperator.IN, new ArrayList<>(preferences.keySet()))
+               .withCriteria("emailStatus", QCriteriaOperator.IN, List.of(EmailStatus.NONE.name(), EmailStatus.PENDING.name())));
          }
 
          List<QRecord> recipients = AlertsDb.query(AlertsNotificationRecipient.TABLE_NAME, new QQueryFilter()
-            .withCriteria("userId", QCriteriaOperator.IN, new ArrayList<>(preferences.keySet()))
-            .withCriteria("emailStatus", QCriteriaOperator.IN, List.of(EmailStatus.NONE.name(), EmailStatus.PENDING.name()))
+            .withSubFilter(eligible)
             .withCriteria("deliveredAt", QCriteriaOperator.GREATER_THAN_OR_EQUALS, now.minus(1, ChronoUnit.DAYS))
             .withLimit(SCAN));
+         if(recipients.isEmpty())
+         {
+            return (new Result(0, 0, 0, 0));
+         }
          Map<Integer, QRecord> notifications = notifications(recipients, now);
-         Map<String, QRecord>  settings      = userSettings(preferences.keySet());
+         Map<String, QRecord>  settings      = userSettings(recipients.stream().map(recipient -> recipient.getValueString("userId")).collect(Collectors.toSet()));
 
          int                         sent    = 0;
          int                         skipped = 0;
@@ -145,14 +155,16 @@ public class NotificationEmailer
          {
             QRecord notification = notifications.get(recipient.getValueInteger("notificationId"));
             String  userId       = recipient.getValueString("userId");
-            QRecord preference   = notification == null ? null : preferences.get(userId).get(notification.getValueString("category"));
-            if(notification == null || preference == null || NotificationCategory.ALERT.name().equals(notification.getValueString("category"))
-               || !atLeast(notification.getValueString("severity"), preference.getValueString("emailMinSeverity")))
+            QRecord preference   = notification == null ? null : preferences.getOrDefault(userId, Map.of()).get(notification.getValueString("category"));
+            boolean requested    = EmailStatus.REQUESTED.name().equals(recipient.getValueString("emailStatus"));
+            if(notification == null || NotificationCategory.ALERT.name().equals(notification.getValueString("category"))
+               || (!requested && (preference == null || !atLeast(notification.getValueString("severity"), preference.getValueString("emailMinSeverity"))))
+               || (requested && !StringUtils.hasContent(address(settings.get(userId)))))
             {
                mark(recipient, EmailStatus.SKIPPED, null);
                skipped++;
             }
-            else if(EmailMode.IMMEDIATE.name().equals(preference.getValueString("emailMode")))
+            else if(requested || EmailMode.IMMEDIATE.name().equals(preference.getValueString("emailMode")))
             {
                boolean ok = send(address(settings.get(userId)), notification.getValueString("title"), List.of(notification));
                mark(recipient, ok ? EmailStatus.SENT : EmailStatus.FAILED, ok ? now : null);

src/test/java/com/kingsrook/qbits/alerts/notifications/NotificationEmailAndNoticesTest.java

@@ -166,6 +166,162 @@ class NotificationEmailAndNoticesTest extends BaseAlertsTest
 
 
 
+   /*******************************************************************************
+    ** A requested email does not require a standing category preference.
+    *******************************************************************************/
+   @Test
+   void testRequestedWithoutPreference() throws Exception
+   {
+      addressOnly();
+      requested("JOB");
+      assertThat(new NotificationEmailer(runtime).run().sent()).isEqualTo(1);
+      assertThat(emailStatus("viewer-1")).isEqualTo("SENT");
+      assertThat(new NotificationEmailer(runtime).run().sent()).isZero();
+      assertThat(SENT).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** An explicit request and an immediate preference still send only once.
+    *******************************************************************************/
+   @Test
+   void testRequestedWithImmediatePreference() throws Exception
+   {
+      emailSettings("viewer-1", "JOB", "IMMEDIATE", "ERROR");
+      requested("JOB");
+      assertThat(new NotificationEmailer(runtime).run().sent()).isEqualTo(1);
+      assertThat(new NotificationEmailer(runtime).run().sent()).isZero();
+      assertThat(SENT).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** A request overrides digest timing and category severity preferences.
+    *******************************************************************************/
+   @Test
+   void testRequestedWithDigestPreference() throws Exception
+   {
+      emailSettings("viewer-1", "JOB", "DAILY_DIGEST", "ERROR");
+      requested("JOB");
+      NotificationEmailer.Result result = new NotificationEmailer(runtime).run();
+      assertThat(result.sent()).isEqualTo(1);
+      assertThat(result.digests()).isZero();
+      assertThat(SENT).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Knowing an address alone does not opt a notification into email.
+    *******************************************************************************/
+   @Test
+   void testNotRequestedWithoutPreference() throws Exception
+   {
+      addressOnly();
+      runtime.getNotifications().notify(new NotificationSignal("Done").withCategory(NotificationCategory.JOB)
+         .withAudience(NotificationAudience.user("viewer-1")));
+      assertThat(new NotificationEmailer(runtime).run().sent()).isZero();
+      assertThat(emailStatus("viewer-1")).isEqualTo("NONE");
+      assertThat(SENT).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Missing addresses are skipped rather than counted as delivery failures.
+    *******************************************************************************/
+   @Test
+   void testRequestedWithoutAddress() throws Exception
+   {
+      requested("JOB");
+      NotificationEmailer.Result result = new NotificationEmailer(runtime).run();
+      assertThat(result.skipped()).isEqualTo(1);
+      assertThat(result.failed()).isZero();
+      assertThat(emailStatus("viewer-1")).isEqualTo("SKIPPED");
+      assertThat(SENT).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Alert subscriptions own alert emails, even with an explicit request.
+    *******************************************************************************/
+   @Test
+   void testRequestedAlertNeverEmailed() throws Exception
+   {
+      emailSettings("viewer-1", "ALERT", "IMMEDIATE", "INFO");
+      requested("ALERT");
+      assertThat(new NotificationEmailer(runtime).run().skipped()).isEqualTo(1);
+      assertThat(emailStatus("viewer-1")).isEqualTo("SKIPPED");
+      assertThat(SENT).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Requests older than the delivery window are not emailed.
+    *******************************************************************************/
+   @Test
+   void testRequestedOutsideDeliveryWindow() throws Exception
+   {
+      addressOnly();
+      requested("JOB");
+      clock.set(clock.instant().plusSeconds(86401));
+      assertThat(new NotificationEmailer(runtime).run().sent()).isZero();
+      assertThat(emailStatus("viewer-1")).isEqualTo("REQUESTED");
+      assertThat(SENT).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Provider failures remain terminal and do not cause repeated sends.
+    *******************************************************************************/
+   @Test
+   void testRequestedDeliveryFailure() throws Exception
+   {
+      addressOnly();
+      requested("JOB");
+      runtime.getConfig().withMessagingProviderName("missing-provider");
+      NotificationEmailer.Result result = new NotificationEmailer(runtime).run();
+      assertThat(result.failed()).isEqualTo(1);
+      assertThat(emailStatus("viewer-1")).isEqualTo("FAILED");
+      assertThat(new NotificationEmailer(runtime).run().failed()).isZero();
+      assertThat(SENT).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** Store an address without opting into any category.
+    *******************************************************************************/
+   private void addressOnly() throws QException
+   {
+      runtime.asSystem(() -> AlertsDb.insert(AlertsUserPreference.TABLE_NAME,
+         new QRecord().withValue("userId", "viewer-1").withValue("emailAddress", "viewer-1@example.com")));
+   }
+
+
+
+   /*******************************************************************************
+    ** Set up a recipient explicitly requesting email for worker tests.
+    *******************************************************************************/
+   private void requested(String category) throws QException
+   {
+      Integer id = runtime.getNotifications().notify(new NotificationSignal("Done").withCategory(NotificationCategory.valueOf(category))
+         .withAudience(NotificationAudience.user("viewer-1")));
+      runtime.asSystem(() ->
+      {
+         QRecord recipient = AlertsDb.first(AlertsNotificationRecipient.TABLE_NAME, AlertsDb.eq("notificationId", id));
+         return (AlertsDb.update(AlertsNotificationRecipient.TABLE_NAME,
+            new QRecord().withValue("id", recipient.getValue("id")).withValue("emailStatus", "REQUESTED")));
+      });
+   }
+
+
+
    /*******************************************************************************
     ** The daily digest waits for the user's hour, then lists the unread ones.
     *******************************************************************************/