Merge pull request #1515 from DependencyTrack/notification-router

Author: nscuro

apiserver/src/main/java/org/dependencytrack/model/NotificationRule.java

@@ -37,6 +37,7 @@
 import javax.jdo.annotations.Column;
 import javax.jdo.annotations.Element;
 import javax.jdo.annotations.Extension;
+import javax.jdo.annotations.Extensions;
 import javax.jdo.annotations.ForeignKey;
 import javax.jdo.annotations.ForeignKeyAction;
 import javax.jdo.annotations.IdGeneratorStrategy;
@@ -110,6 +111,10 @@ public class NotificationRule implements Serializable {
 
     @Persistent(defaultFetchGroup = "true")
     @Column(name = "NOTIFICATION_LEVEL", jdbcType = "VARCHAR")
+    @Extensions(value = {
+            @Extension(vendorName = "datanucleus", key = "insert-function", value = "CAST(? AS notification_level)"),
+            @Extension(vendorName = "datanucleus", key = "update-function", value = "CAST(? AS notification_level)")
+    })
     private NotificationLevel notificationLevel;
 
     @Persistent(table = "NOTIFICATIONRULE_PROJECTS", defaultFetchGroup = "true")

apiserver/src/main/java/org/dependencytrack/notification/NotificationFactory.java

@@ -18,9 +18,6 @@
  */
 package org.dependencytrack.notification;
 
-import alpine.model.LdapUser;
-import alpine.model.ManagedUser;
-import alpine.model.OidcUser;
 import com.fasterxml.uuid.Generators;
 import com.fasterxml.uuid.impl.TimeBasedEpochRandomGenerator;
 import com.google.protobuf.Any;
@@ -46,6 +43,7 @@
 import org.dependencytrack.proto.notification.v1.ProjectVulnAnalysisCompleteSubject;
 import org.dependencytrack.proto.notification.v1.ProjectVulnAnalysisStatus;
 import org.dependencytrack.proto.notification.v1.Scope;
+import org.dependencytrack.proto.notification.v1.UserSubject;
 import org.dependencytrack.proto.notification.v1.VexConsumedOrProcessedSubject;
 import org.dependencytrack.proto.notification.v1.Vulnerability;
 import org.dependencytrack.proto.notification.v1.VulnerabilityAnalysis;
@@ -453,33 +451,29 @@ public static Notification createUserCreatedNotification(
             final alpine.model.User user) {
         requireNonNull(user, "user must not be null");
 
+        return createUserCreatedNotification(convert(user));
+    }
+
+    static Notification createUserCreatedNotification(final UserSubject user) {
         return newNotificationBuilder(SCOPE_SYSTEM, GROUP_USER_CREATED, LEVEL_INFORMATIONAL)
                 .setTitle("User Created")
-                .setContent(switch (user) {
-                    case LdapUser ignored -> "LDAP";
-                    case ManagedUser ignored -> "Managed";
-                    case OidcUser ignored -> "OpenID Connect";
-                    default -> throw new IllegalStateException(
-                            "Unexpected user type: " + user.getClass());
-                } + " user created")
-                .setSubject(Any.pack(convert(user)))
+                .setContent("User %s was created".formatted(user.getUsername()))
+                .setSubject(Any.pack(user))
                 .build();
     }
 
     public static Notification createUserDeletedNotification(
             final alpine.model.User user) {
         requireNonNull(user, "user must not be null");
 
+        return createUserDeletedNotification(convert(user));
+    }
+
+    static Notification createUserDeletedNotification(final UserSubject user) {
         return newNotificationBuilder(SCOPE_SYSTEM, GROUP_USER_DELETED, LEVEL_INFORMATIONAL)
                 .setTitle("User Deleted")
-                .setContent(switch (user) {
-                    case LdapUser ignored -> "LDAP";
-                    case ManagedUser ignored -> "Managed";
-                    case OidcUser ignored -> "OpenID Connect";
-                    default -> throw new IllegalStateException(
-                            "Unexpected user type: " + user.getClass());
-                } + " user deleted")
-                .setSubject(Any.pack(convert(user)))
+                .setContent("User %s was deleted".formatted(user.getUsername()))
+                .setSubject(Any.pack(user))
                 .build();
     }
 

apiserver/src/main/java/org/dependencytrack/notification/NotificationOutboxRelay.java

@@ -69,6 +69,7 @@ final class NotificationOutboxRelay implements Closeable {
 
     private final KafkaEventDispatcher delegateDispatcher;
     private final MeterRegistry meterRegistry;
+    private final boolean routerEnabled;
     private final long pollIntervalMillis;
     private final int batchSize;
     private final BlockingQueue<Notification> currentBatch;
@@ -83,6 +84,7 @@ final class NotificationOutboxRelay implements Closeable {
     public NotificationOutboxRelay(
             final KafkaEventDispatcher delegateDispatcher,
             final MeterRegistry meterRegistry,
+            final boolean routerEnabled,
             final long pollIntervalMillis,
             final int batchSize) {
         this.delegateDispatcher = requireNonNull(delegateDispatcher, "delegate dispatcher must not be null");
@@ -93,6 +95,7 @@ public NotificationOutboxRelay(
         if (batchSize <= 0) {
             throw new IllegalArgumentException("batchSize must be greater than 0");
         }
+        this.routerEnabled = routerEnabled;
         this.pollIntervalMillis = pollIntervalMillis;
         this.batchSize = batchSize;
         this.currentBatch = new ArrayBlockingQueue<>(batchSize);
@@ -222,6 +225,18 @@ private RelayCycleOutcome executeRelayCycle() {
                 return RelayCycleOutcome.COMPLETED;
             }
 
+            if (routerEnabled) {
+                try {
+                    final List<NotificationPublishTask> publishTasks =
+                            new NotificationRouter(handle, meterRegistry).route(currentBatch);
+                    LOGGER.debug("Router generated {} publish tasks", publishTasks.size());
+                } catch (RuntimeException e) {
+                    LOGGER.warn("""
+                            Router failed, but since routing results are not currently used,
+                            the failure is ignored. If it continues to fail, consider disabling the router.""", e);
+                }
+            }
+
             final Timer.Sample sendLatencySample = Timer.start();
             try {
                 sendAll(currentBatch);

apiserver/src/main/java/org/dependencytrack/notification/NotificationPublishTask.java

@@ -0,0 +1,35 @@
+/*
+ * This file is part of Dependency-Track.
+ *
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright (c) OWASP Foundation. All Rights Reserved.
+ */
+package org.dependencytrack.notification;
+
+import org.dependencytrack.proto.notification.v1.Notification;
+
+/**
+ * Unit of work for publishing a notification.
+ *
+ * @param ruleId       ID of the applicable notification rule.
+ * @param ruleName     Name of the applicable notification rule.
+ * @param notification The notification to publish.
+ * @since 5.7.0
+ */
+record NotificationPublishTask(
+        long ruleId,
+        String ruleName,
+        Notification notification) {
+}