src/main/java/com/kingsrook/qbits/alerts/engine/AlertsProcessListener.java

@@ -18,29 +18,49 @@ package com.kingsrook.qbits.alerts.engine;
 
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import com.kingsrook.qbits.alerts.AlertsQBitConfig;
 import com.kingsrook.qbits.alerts.AlertsRuntime;
+import com.kingsrook.qbits.alerts.AlertsSecurity;
 import com.kingsrook.qbits.alerts.model.AlertsMonitor;
+import com.kingsrook.qbits.alerts.spi.AlertLink;
 import com.kingsrook.qbits.alerts.spi.AlertMonitorKind;
+import com.kingsrook.qbits.alerts.spi.NotificationAudience;
+import com.kingsrook.qbits.alerts.spi.NotificationCategory;
+import com.kingsrook.qbits.alerts.spi.NotificationSeverity;
+import com.kingsrook.qbits.alerts.spi.NotificationSignal;
 import com.kingsrook.qqq.backend.core.actions.processes.listeners.ProcessLifecycleListenerInterface;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
 import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
 import com.kingsrook.qqq.backend.core.utils.ValueUtils;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
 
 /*******************************************************************************
- ** Feeds PROCESS_FAILED and process HEARTBEAT monitors from QQQ's process
- ** lifecycle events (no ESB needed): a failed run raises, a completed run
- ** resolves and beats the process's heartbeat key. Only processes some
- ** monitor watches are handled (the watched names are cached for a minute),
- ** so other process runs cost one set lookup. Never affects the process.
+ ** Feeds alerts and notifications from QQQ's process lifecycle events (no ESB
+ ** needed). Never affects the process.
+ ** - PROCESS_FAILED and process HEARTBEAT monitors: a failed run raises, a
+ **   completed run resolves and beats the process's heartbeat key. Only
+ **   processes some monitor watches are handled (the watched names are
+ **   cached for a minute).
+ ** - Job notifications (docs/NOTIFICATIONS.md 6.1): when a run started by a
+ **   person fails, or completes with error summary lines (a partial run),
+ **   that person gets a JOB notification linked to the run. System runs and
+ **   the qbit's own processes are skipped. Off with notifyFailedJobs(false).
  *******************************************************************************/
 public class AlertsProcessListener implements ProcessLifecycleListenerInterface
 {
@@ -64,6 +84,16 @@ public class AlertsProcessListener implements ProcessLifecycleListenerInterface
       {
          return (false);
       }
+      return (notifiesJobs(runtime, processName) || isWatched(runtime, processName));
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether a monitor watches this process.
+    *******************************************************************************/
+   private static boolean isWatched(AlertsRuntime runtime, String processName)
+   {
       for(String pattern : watched(runtime))
       {
          if(processName.equals(pattern) || processName.matches(pattern.replace(".", "\\.").replace("*", ".*")))
@@ -102,8 +132,17 @@ public class AlertsProcessListener implements ProcessLifecycleListenerInterface
       }
       try
       {
-         runtime.getEngine().heartbeat(MonitorEvaluator.processHeartbeatKey(input.getProcessName()), runtime.now());
-         runtime.getEvaluator().processCompleted(input.getProcessName());
+         if(isWatched(runtime, input.getProcessName()))
+         {
+            runtime.getEngine().heartbeat(MonitorEvaluator.processHeartbeatKey(input.getProcessName()), runtime.now());
+            runtime.getEvaluator().processCompleted(input.getProcessName());
+         }
+         int errors = errorCount(output);
+         if(errors > 0 && notifiesJobs(runtime, input.getProcessName()))
+         {
+            notifyStarter(runtime, input, NotificationSeverity.WARNING, "%s finished with errors",
+               (errors == 1 ? "1 record" : errors + " records") + " had errors. Open the run's results for the details.");
+         }
       }
       catch(Exception e)
       {
@@ -126,9 +165,17 @@ public class AlertsProcessListener implements ProcessLifecycleListenerInterface
       }
       try
       {
-         QProcessMetaData process = QContext.getQInstance().getProcess(input.getProcessName());
-         runtime.getEvaluator().processFailed(input.getProcessName(), process == null ? null : process.getLabel(),
-            exception == null ? null : exception.getClass().getSimpleName() + ": " + exception.getMessage());
+         if(isWatched(runtime, input.getProcessName()))
+         {
+            QProcessMetaData process = QContext.getQInstance().getProcess(input.getProcessName());
+            runtime.getEvaluator().processFailed(input.getProcessName(), process == null ? null : process.getLabel(),
+               exception == null ? null : exception.getClass().getSimpleName() + ": " + exception.getMessage());
+         }
+         if(notifiesJobs(runtime, input.getProcessName()))
+         {
+            String message = exception == null || exception.getMessage() == null ? "It stopped with an error." : exception.getMessage();
+            notifyStarter(runtime, input, NotificationSeverity.ERROR, "%s failed", StringUtils.safeTruncate(message, 1000));
+         }
       }
       catch(Exception e)
       {
@@ -138,6 +185,92 @@ public class AlertsProcessListener implements ProcessLifecycleListenerInterface
 
 
 
+   /*******************************************************************************
+    ** Whether runs of this process notify the person who started them.
+    *******************************************************************************/
+   private static boolean notifiesJobs(AlertsRuntime runtime, String processName)
+   {
+      return (Boolean.TRUE.equals(runtime.getConfig().getNotifyFailedJobs()) && !processName.startsWith(AlertsQBitConfig.PERMISSION_BASE_NAME));
+   }
+
+
+
+   /*******************************************************************************
+    ** How many records a completed run reported as errors (its ERROR summary
+    ** lines), or 1 when the output carries an exception.
+    *******************************************************************************/
+   public static int errorCount(RunProcessOutput output)
+   {
+      if(output == null)
+      {
+         return (0);
+      }
+      if(output.getException().isPresent())
+      {
+         return (1);
+      }
+      int    errors  = 0;
+      Object summary = output.getValues() == null ? null : output.getValues().get(StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY);
+      if(summary instanceof List<?> lines)
+      {
+         for(Object line : lines)
+         {
+            if(line instanceof ProcessSummaryLineInterface summaryLine && summaryLine.getStatus() == Status.ERROR)
+            {
+               Integer count = line instanceof ProcessSummaryLine counted ? counted.getCount() : null;
+               errors += count == null || count < 1 ? 1 : count;
+            }
+         }
+      }
+      return (errors);
+   }
+
+
+
+   /*******************************************************************************
+    ** Send a JOB notification to the person who started the run (nothing for
+    ** system runs or sessions without a user).
+    *******************************************************************************/
+   private static void notifyStarter(AlertsRuntime runtime, RunProcessInput input, NotificationSeverity severity, String titleFormat, String body)
+   {
+      QSession session = QContext.getQSession();
+      String   userId  = session == null || session instanceof QSystemUserSession ? null : AlertsSecurity.userId(session);
+      if(userId == null)
+      {
+         return;
+      }
+      QProcessMetaData process   = QContext.getQInstance().getProcess(input.getProcessName());
+      String           label     = process == null || process.getLabel() == null ? input.getProcessName() : process.getLabel();
+      String           tableName = process == null ? null : process.getTableName();
+      runtime.sink().notify(new NotificationSignal(String.format(titleFormat, label))
+         .withBody(body)
+         .withSeverity(severity)
+         .withCategory(NotificationCategory.JOB)
+         .withSourceName("processes")
+         .withSourceKey("process:" + input.getProcessUUID())
+         .withSourceTableName(tableName)
+         .withLink(jobLink(runtime, input.getProcessName(), input.getProcessUUID(), tableName))
+         .withAudience(NotificationAudience.user(userId)));
+   }
+
+
+
+   /*******************************************************************************
+    ** Where a job notification links: the configured run link pattern, else
+    ** the process's table, else nowhere.
+    *******************************************************************************/
+   public static AlertLink jobLink(AlertsRuntime runtime, String processName, String processUUID, String tableName)
+   {
+      String pattern = runtime.getConfig().getJobRunLinkPattern();
+      if(StringUtils.hasContent(pattern))
+      {
+         return (AlertLink.toPath(pattern.replace("{processName}", processName).replace("{processUUID}", processUUID == null ? "" : processUUID), "Open the run"));
+      }
+      return (tableName == null ? null : AlertLink.toPath("/app/" + tableName, "Open " + tableName));
+   }
+
+
+
    /*******************************************************************************
     ** Forget the cached names (after monitors change; tests).
     *******************************************************************************/

src/test/java/com/kingsrook/qbits/alerts/demo/AlertsDemoServer.java

@@ -39,6 +39,7 @@ import com.kingsrook.qbits.alerts.model.AlertsMonitor;
 import com.kingsrook.qbits.alerts.model.AlertsSubscription;
 import com.kingsrook.qbits.alerts.model.AlertsType;
 import com.kingsrook.qbits.alerts.model.AlertsUserPreference;
+import com.kingsrook.qbits.alerts.notifications.NotificationService;
 import com.kingsrook.qbits.alerts.spi.AlertDefaultMonitor;
 import com.kingsrook.qbits.alerts.spi.AlertLink;
 import com.kingsrook.qbits.alerts.spi.AlertMonitorKind;
@@ -46,6 +47,10 @@ import com.kingsrook.qbits.alerts.spi.AlertRecordRef;
 import com.kingsrook.qbits.alerts.spi.AlertSeverity;
 import com.kingsrook.qbits.alerts.spi.AlertSignal;
 import com.kingsrook.qbits.alerts.spi.AlertsInstanceMetaData;
+import com.kingsrook.qbits.alerts.spi.NotificationAudience;
+import com.kingsrook.qbits.alerts.spi.NotificationCategory;
+import com.kingsrook.qbits.alerts.spi.NotificationSeverity;
+import com.kingsrook.qbits.alerts.spi.NotificationSignal;
 import com.kingsrook.qbits.alerts.spi.QAlertTypeMetaData;
 import com.kingsrook.qbits.alerts.spi.ResolveMode;
 import com.kingsrook.qqq.backend.core.context.QContext;
@@ -169,6 +174,55 @@ public class AlertsDemoServer extends AbstractQQQApplication
 
 
 
+   /*******************************************************************************
+    ** User notifications for the bell, the page and the admin screens: an
+    ** announcement, a sticky security notice, a failed job, a share, a
+    ** marketing note, a draft and a scheduled one.
+    *******************************************************************************/
+   static void seedNotifications(AlertsRuntime runtime, BaseAlertsTest.MutableClock clock) throws QException
+   {
+      Instant now = Instant.now();
+      runtime.asSystem(() ->
+      {
+         AlertsDb.insert("alerts_user_preference", new QRecord().withValue("userId", "sam").withValue("userLabel", "Sam Rivera").withValue("inAppEnabled", true)
+            .withValue("emailEnabled", false).withValue("digestHour", 8));
+         return (null);
+      });
+      NotificationService service = runtime.getNotifications();
+      clock.set(now.minus(Duration.ofDays(3)));
+      service.notify(new NotificationSignal("Version 4.1 is out").withBody("**What's new:** a notification center in the top bar, record pickers with thumbnails, "
+            + "and dark mode for the SQL Workbench. " + "\n\n" + "See the release notes for everything else.")
+         .withSeverity(NotificationSeverity.SUCCESS).withCategory(NotificationCategory.RELEASE).withSourceName("admin")
+         .withLink(AlertLink.toUrl("https://github.com/QRun-IO/qqq/releases", "Release notes")).withAudience(NotificationAudience.allUsers()));
+      clock.set(now.minus(Duration.ofDays(1)));
+      service.notify(new NotificationSignal("Spring content review starts Monday").withBody("Editors: please review your top 10 posts' excerpts and images before the campaign.")
+         .withCategory(NotificationCategory.MARKETING).withSourceName("qbit-marketing").withAudience(NotificationAudience.allUsers()));
+      clock.set(now.minus(Duration.ofHours(5)));
+      service.notify(new NotificationSignal("Sam shared the notebook \"Q3 traffic\" with you").withCategory(NotificationCategory.SHARE).withSourceName("qbit-sql-workbench")
+         .withLink(AlertLink.toPath("/app/blog_post", "Open notebook")).withAudience(NotificationAudience.user(ADMIN)));
+      clock.set(now.minus(Duration.ofMinutes(40)));
+      service.notify(new NotificationSignal("Blog post import finished with errors").withBody("3 rows had errors: the publish date was not a date. The other 117 rows were imported.")
+         .withSeverity(NotificationSeverity.WARNING).withCategory(NotificationCategory.JOB).withSourceName("processes").withSourceKey("process:demo-import")
+         .withLink(AlertLink.toPath("/app/blog_post", "Open Blog Posts")).withAudience(NotificationAudience.user(ADMIN)));
+      clock.set(now.minus(Duration.ofMinutes(8)));
+      service.notify(new NotificationSignal("New sign-in to your account").withBody("A sign-in from Firefox on Windows near Columbus, OH. If this wasn't you, change your password now.")
+         .withSeverity(NotificationSeverity.CRITICAL).withCategory(NotificationCategory.SECURITY).withSourceName("security").withSticky(true)
+         .withAudience(NotificationAudience.user(ADMIN)));
+      clock.set(now);
+      runtime.asSystem(() ->
+      {
+         AlertsDb.insert("alerts_notification", new QRecord().withValue("title", "Planned maintenance Saturday 06:00 UTC").withValue("body", "The admin is read-only for about 15 minutes.")
+            .withValue("severity", "WARNING").withValue("category", "SYSTEM").withValue("status", "DRAFT").withValue("audienceType", "ALL_USERS").withValue("authorUserId", ADMIN)
+            .withValue("sourceName", "admin").withValue("sticky", false));
+         AlertsDb.insert("alerts_notification", new QRecord().withValue("title", "Holiday hours").withValue("body", "Support answers within one business day over the holidays.")
+            .withValue("severity", "INFO").withValue("category", "SYSTEM").withValue("status", "SCHEDULED").withValue("audienceType", "ROLES").withValue("audienceValue", "staff")
+            .withValue("publishAt", now.plus(Duration.ofDays(10))).withValue("authorUserId", ADMIN).withValue("sourceName", "admin").withValue("sticky", false));
+         return (null);
+      });
+   }
+
+
+
    /*******************************************************************************
     ** Create the tables and produce 30 days of history through the engine.
     *******************************************************************************/
@@ -292,6 +346,7 @@ public class AlertsDemoServer extends AbstractQQQApplication
          runtime.getEngine().comment(acked.getValueInteger("id"), ADMIN, "Asked the author for an excerpt.");
       }
       runtime.getDeliveryWorker().deliverDue();
+      seedNotifications(runtime, clock);
       runtime.setClock(Clock.systemUTC());
       QContext.clear();
    }

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

@@ -0,0 +1,171 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package com.kingsrook.qbits.alerts.notifications;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.alerts.AlertsSecurity;
+import com.kingsrook.qbits.alerts.BaseAlertsTest;
+import com.kingsrook.qbits.alerts.TestSteps;
+import com.kingsrook.qbits.alerts.api.AlertsApi;
+import com.kingsrook.qbits.alerts.engine.AlertsDb;
+import com.kingsrook.qbits.alerts.engine.AlertsProcessListener;
+import com.kingsrook.qbits.alerts.model.AlertsNotification;
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Job notifications: a person's failed or partial process runs reach their
+ ** bell (docs/NOTIFICATIONS.md 6.1).
+ *******************************************************************************/
+class JobNotificationsTest extends BaseAlertsTest
+{
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   @AfterEach
+   void resetFlaky()
+   {
+      TestSteps.fail = false;
+   }
+
+
+
+   /*******************************************************************************
+    ** Run the test process as the current session.
+    *******************************************************************************/
+   private RunProcessOutput runNightly() throws Exception
+   {
+      return (new RunProcessAction().execute(new RunProcessInput().withProcessName("nightlyThing").withFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.SKIP)));
+   }
+
+
+
+   /*******************************************************************************
+    ** A failed run notifies the person who started it, once per run.
+    *******************************************************************************/
+   @Test
+   void testFailedRunNotifiesStarter() throws Exception
+   {
+      TestSteps.fail = true;
+      loginAs("viewer-1");
+      try
+      {
+         runNightly();
+      }
+      catch(Exception e)
+      {
+         ////////////////////////////////////////
+         // the run's own failure is expected //
+         ////////////////////////////////////////
+      }
+
+      List<QRecord> notifications = runtime.asSystem(() -> all(AlertsNotification.TABLE_NAME));
+      assertThat(notifications).singleElement().satisfies(notification ->
+      {
+         assertThat(notification.getValueString("title")).isEqualTo("Nightly thing failed");
+         assertThat(notification.getValueString("body")).contains("disk full");
+         assertThat(notification.getValueString("category")).isEqualTo("JOB");
+         assertThat(notification.getValueString("severity")).isEqualTo("ERROR");
+         assertThat(notification.getValueString("sourceKey")).startsWith("process:");
+         assertThat(notification.getValueString("audienceValue")).isEqualTo("viewer-1");
+      });
+
+      Map<String, Object> summary = new AlertsApi(runtime).inboxSummary();
+      assertThat(summary.get("unreadCount")).isEqualTo(1);
+
+      //////////////////////////////////////////////////////////////
+      // system runs and a successful run notify nobody           //
+      //////////////////////////////////////////////////////////////
+      QContext.init(qInstance, AlertsSecurity.systemSession());
+      try
+      {
+         runNightly();
+      }
+      catch(Exception e)
+      {
+         ////////////////////////////////////////
+         // the run's own failure is expected //
+         ////////////////////////////////////////
+      }
+      TestSteps.fail = false;
+      loginAs("viewer-1");
+      runNightly();
+      assertThat(runtime.asSystem(() -> all(AlertsNotification.TABLE_NAME))).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Turned off, nothing is sent.
+    *******************************************************************************/
+   @Test
+   void testOff() throws Exception
+   {
+      config.setNotifyFailedJobs(false);
+      TestSteps.fail = true;
+      loginAs("viewer-1");
+      try
+      {
+         runNightly();
+      }
+      catch(Exception e)
+      {
+         ////////////////////////////////////////
+         // the run's own failure is expected //
+         ////////////////////////////////////////
+      }
+      assertThat(runtime.asSystem(() -> AlertsDb.count(AlertsNotification.TABLE_NAME, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter()))).isEqualTo(0);
+   }
+
+
+
+   /*******************************************************************************
+    ** Partial runs: ERROR summary lines count; links follow the pattern.
+    *******************************************************************************/
+   @Test
+   void testPartialRunsAndLinks()
+   {
+      RunProcessOutput output = new RunProcessOutput();
+      assertThat(AlertsProcessListener.errorCount(output)).isEqualTo(0);
+      ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+      lines.add(new ProcessSummaryLine(Status.OK, 10, "rows were loaded"));
+      lines.add(new ProcessSummaryLine(Status.ERROR, 3, "rows had bad dates"));
+      output.addValue(StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY, lines);
+      assertThat(AlertsProcessListener.errorCount(output)).isEqualTo(3);
+      assertThat(AlertsProcessListener.errorCount(null)).isEqualTo(0);
+
+      assertThat(AlertsProcessListener.jobLink(runtime, "p", "u-1", "widget").getPath()).isEqualTo("/app/widget");
+      assertThat(AlertsProcessListener.jobLink(runtime, "p", "u-1", null)).isNull();
+      config.setJobRunLinkPattern("/runs/{processName}/{processUUID}");
+      assertThat(AlertsProcessListener.jobLink(runtime, "p", "u-1", null).getPath()).isEqualTo("/runs/p/u-1");
+   }
+}