From 62dfa5f3509b7ac57883476f38a8659de6bfb4b2 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 13 Oct 2026 10:06:27 -0500 Subject: [PATCH 1/2] feat(notifications): notify people of their failed and partial jobs (phase c) The process lifecycle listener now also sends a JOB notification to the person who started a run that fails, or that completes with ERROR summary lines (a partial run). System runs and the qbit's own processes are skipped; a repeat of the same run updates one notification (sourceKey process:). It links to jobRunLinkPattern or the process's table. Monitor heartbeats stay limited to watched processes. Co-Authored-By: Claude Opus 5.5 --- .../alerts/engine/AlertsProcessListener.java | 153 +++++++++++++++- .../notifications/JobNotificationsTest.java | 171 ++++++++++++++++++ 2 files changed, 314 insertions(+), 10 deletions(-) create mode 100644 src/test/java/com/kingsrook/qbits/alerts/notifications/JobNotificationsTest.java diff --git a/src/main/java/com/kingsrook/qbits/alerts/engine/AlertsProcessListener.java b/src/main/java/com/kingsrook/qbits/alerts/engine/AlertsProcessListener.java index 924ab81..d79ef27 100644 --- a/src/main/java/com/kingsrook/qbits/alerts/engine/AlertsProcessListener.java +++ b/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). *******************************************************************************/ diff --git a/src/test/java/com/kingsrook/qbits/alerts/notifications/JobNotificationsTest.java b/src/test/java/com/kingsrook/qbits/alerts/notifications/JobNotificationsTest.java new file mode 100644 index 0000000..1a7ef11 --- /dev/null +++ b/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 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 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 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"); + } +} -- 2.54.0 From 1e8f9aac25b2bb128678edb38f8cdecedcd03831 Mon Sep 17 00:00:00 2001 From: James Maes Date: Tue, 13 Oct 2026 10:08:51 -0500 Subject: [PATCH 2/2] test(demo): seed user notifications for the bell, page and admin screens Co-Authored-By: Claude Opus 5.5 --- .../qbits/alerts/demo/AlertsDemoServer.java | 55 +++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/src/test/java/com/kingsrook/qbits/alerts/demo/AlertsDemoServer.java b/src/test/java/com/kingsrook/qbits/alerts/demo/AlertsDemoServer.java index b339d73..33e9fde 100644 --- a/src/test/java/com/kingsrook/qbits/alerts/demo/AlertsDemoServer.java +++ b/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(); } -- 2.54.0