Complete notification publishing migration

* Enables notification router and removes feature flag.
* Implements dex workflow for publishing.
* Removes direct publishing to Kafka.
* Adds partial rule filtering at notification emission time.

Signed-off-by: nscuro <nscuro@protonmail.com>

Author: nscuro

api/src/main/openapi/components/schemas/secrets/create-secret-request.yaml

@@ -24,7 +24,7 @@ properties:
   value:
     type: string
     minLength: 1
-    maxLength: 1024
+    maxLength: 4096
 required:
 - name
 - value

api/src/main/openapi/components/schemas/secrets/update-secret-request.yaml

@@ -26,4 +26,4 @@ properties:
     description: >-
       The new value. Omit this field to retain the current value.
     minLength: 1
-    maxLength: 1024
+    maxLength: 4096

apiserver/pom.xml

@@ -590,6 +590,7 @@
                         --add-opens java.base/java.net=ALL-UNNAMED
                         --add-opens java.base/java.lang=ALL-UNNAMED
                         --add-opens java.base/java.util=ALL-UNNAMED
+                        --add-opens java.base/java.util.concurrent=ALL-UNNAMED
                         -javaagent:${settings.localRepository}/org/mockito/mockito-core/${lib.mockito.version}/mockito-core-${lib.mockito.version}.jar
                         -Xshare:off
                     </argLine>

apiserver/src/main/java/org/dependencytrack/common/MdcKeys.java

@@ -38,6 +38,11 @@ public final class MdcKeys {
     public static final String MDC_KAFKA_RECORD_PARTITION = "kafkaRecordPartition";
     public static final String MDC_KAFKA_RECORD_OFFSET = "kafkaRecordOffset";
     public static final String MDC_KAFKA_RECORD_KEY = "kafkaRecordKey";
+    public static final String MDC_NOTIFICATION_GROUP = "notificationGroup";
+    public static final String MDC_NOTIFICATION_ID = "notificationId";
+    public static final String MDC_NOTIFICATION_LEVEL = "notificationLevel";
+    public static final String MDC_NOTIFICATION_RULE_NAME = "notificationRuleName";
+    public static final String MDC_NOTIFICATION_SCOPE = "notificationScope";
     public static final String MDC_PLUGIN = "plugin";
     public static final String MDC_PROJECT_NAME = "projectName";
     public static final String MDC_PROJECT_UUID = "projectUuid";

apiserver/src/main/java/org/dependencytrack/dev/DevServicesInitializer.java

@@ -154,20 +154,6 @@ public void contextInitialized(final ServletContextEvent event) {
         }
 
         final var topicsToCreate = new ArrayList<>(List.of(
-                new NewTopic(KafkaTopics.NOTIFICATION_ANALYZER.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_BOM.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_CONFIGURATION.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_DATASOURCE_MIRRORING.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_FILE_SYSTEM.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_INTEGRATION.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_NEW_VULNERABILITY.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_NEW_VULNERABLE_DEPENDENCY.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_POLICY_VIOLATION.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_PROJECT_AUDIT_CHANGE.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_PROJECT_CREATED.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_REPOSITORY.name(), 1, (short) 1),
-                new NewTopic(KafkaTopics.NOTIFICATION_VEX.name(), 1, (short) 1),
                 new NewTopic(KafkaTopics.REPO_META_ANALYSIS_COMMAND.name(), 1, (short) 1),
                 new NewTopic(KafkaTopics.REPO_META_ANALYSIS_RESULT.name(), 1, (short) 1),
                 new NewTopic(KafkaTopics.VULN_ANALYSIS_COMMAND.name(), 1, (short) 1),

apiserver/src/main/java/org/dependencytrack/dex/DexEngineInitializer.java

@@ -26,9 +26,22 @@
 import org.dependencytrack.common.EncryptedPageTokenEncoder;
 import org.dependencytrack.common.datasource.DataSourceRegistry;
 import org.dependencytrack.common.health.HealthCheckRegistry;
+import org.dependencytrack.dex.activity.DeleteFilesActivity;
 import org.dependencytrack.dex.engine.api.DexEngine;
 import org.dependencytrack.dex.engine.api.DexEngineConfig;
 import org.dependencytrack.dex.engine.api.DexEngineFactory;
+import org.dependencytrack.dex.engine.api.TaskType;
+import org.dependencytrack.dex.engine.api.TaskWorkerOptions;
+import org.dependencytrack.dex.engine.api.request.CreateTaskQueueRequest;
+import org.dependencytrack.notification.PublishNotificationActivity;
+import org.dependencytrack.notification.PublishNotificationWorkflow;
+import org.dependencytrack.notification.templating.pebble.PebbleNotificationTemplateRendererFactory;
+import org.dependencytrack.persistence.jdbi.ConfigPropertyDao;
+import org.dependencytrack.plugin.PluginManager;
+import org.dependencytrack.proto.internal.workflow.v1.DeleteFilesArgument;
+import org.dependencytrack.proto.internal.workflow.v1.PublishNotificationActivityArg;
+import org.dependencytrack.proto.internal.workflow.v1.PublishNotificationWorkflowArg;
+import org.dependencytrack.secret.management.SecretManager;
 import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.config.ConfigProvider;
 import org.jspecify.annotations.Nullable;
@@ -38,10 +51,19 @@
 import javax.sql.DataSource;
 import java.io.IOException;
 import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.ServiceLoader;
+import java.util.regex.Pattern;
+import java.util.stream.StreamSupport;
 
 import static java.util.Objects.requireNonNull;
+import static org.dependencytrack.dex.api.payload.PayloadConverters.protoConverter;
+import static org.dependencytrack.dex.api.payload.PayloadConverters.voidConverter;
+import static org.dependencytrack.model.ConfigPropertyConstants.GENERAL_BASE_URL;
+import static org.dependencytrack.persistence.jdbi.JdbiFactory.withJdbiHandle;
 
 /**
  * @since 5.7.0
@@ -74,14 +96,69 @@ public void contextInitialized(ServletContextEvent event) {
         final var healthCheckRegistry = (HealthCheckRegistry) servletContext.getAttribute(HealthCheckRegistry.class.getName());
         requireNonNull(healthCheckRegistry, "healthCheckRegistry has not been initialized");
 
+        final var pluginManager = (PluginManager) servletContext.getAttribute(PluginManager.class.getName());
+        requireNonNull(pluginManager, "pluginManager has not been initialized");
+
+        final var secretManager = (SecretManager) servletContext.getAttribute(SecretManager.class.getName());
+        requireNonNull(pluginManager, "secretManager has not been initialized");
+
+        final var templateRendererFactory = new PebbleNotificationTemplateRendererFactory(
+                Map.of("baseUrl", () -> withJdbiHandle(
+                        handle -> handle
+                                .attach(ConfigPropertyDao.class)
+                                .getOptionalValue(GENERAL_BASE_URL)
+                                .orElse(null))));
+
         final var engineFactory = ServiceLoader.load(DexEngineFactory.class).findFirst().orElseThrow();
         engine = engineFactory.create(engineConfig);
 
-        // Register workflows and activities here.
+        engine.registerWorkflow(
+                new PublishNotificationWorkflow(),
+                protoConverter(PublishNotificationWorkflowArg.class),
+                voidConverter(),
+                Duration.ofMinutes(1));
+        engine.registerActivity(
+                new DeleteFilesActivity(pluginManager),
+                protoConverter(DeleteFilesArgument.class),
+                voidConverter(),
+                Duration.ofMinutes(1));
+        engine.registerActivity(
+                new PublishNotificationActivity(
+                        pluginManager,
+                        secretManager::getSecretValue,
+                        templateRendererFactory),
+                protoConverter(PublishNotificationActivityArg.class),
+                voidConverter(),
+                Duration.ofMinutes(1));
+
+        ensureTaskQueues(engine, List.of(
+                new CreateTaskQueueRequest(TaskType.WORKFLOW, "default", 1000),
+                new CreateTaskQueueRequest(TaskType.ACTIVITY, "default", 1000),
+                new CreateTaskQueueRequest(TaskType.ACTIVITY, "notifications", 25)));
+
+        for (final String workerName : getWorkflowWorkerNames(config)) {
+            if (!isTaskWorkerEnabled(config, TaskType.WORKFLOW, workerName)) {
+                LOGGER.info("Not registering workflow worker '{}' because it is disabled", workerName);
+                continue;
+            }
+            LOGGER.info("Registering workflow worker '{}'", workerName);
+
+            final TaskWorkerOptions workerOptions =
+                    getTaskWorkerOptions(config, TaskType.WORKFLOW, workerName);
+            engine.registerTaskWorker(workerOptions);
+        }
 
-        // Create task queues here.
+        for (final String workerName : getActivityWorkerNames(config)) {
+            if (!isTaskWorkerEnabled(config, TaskType.ACTIVITY, workerName)) {
+                LOGGER.info("Not registering activity worker '{}' because it is disabled", workerName);
+                continue;
+            }
+            LOGGER.info("Registering activity worker '{}'", workerName);
 
-        // Register task workers here.
+            final TaskWorkerOptions workerOptions =
+                    getTaskWorkerOptions(config, TaskType.ACTIVITY, workerName);
+            engine.registerTaskWorker(workerOptions);
+        }
 
         LOGGER.info("Starting durable execution engine");
         healthCheckRegistry.addCheck(new DexEngineHealthCheck(engine));
@@ -177,6 +254,77 @@ private DexEngineConfig createEngineConfig() {
         return engineConfig;
     }
 
+    private void ensureTaskQueues(DexEngine engine, Collection<CreateTaskQueueRequest> requests) {
+        for (final var request : requests) {
+            final boolean created = engine.createTaskQueue(request);
+            if (created) {
+                LOGGER.info(
+                        "Created {} task queue '{}' with capacity {}",
+                        request.type().name().toLowerCase(),
+                        request.name(),
+                        request.capacity());
+            }
+        }
+    }
+
+    private static final Pattern WORKFLOW_WORKER_PROPERTY_PATTERN =
+            Pattern.compile("^dt\\.dex-engine\\.workflow-worker\\..+\\..+$");
+
+    private static List<String> getWorkflowWorkerNames(Config config) {
+        return StreamSupport.stream(config.getPropertyNames().spliterator(), false)
+                .filter(name -> WORKFLOW_WORKER_PROPERTY_PATTERN.matcher(name).matches())
+                .map(name -> name.split("\\.", 5)[3])
+                .distinct()
+                .toList();
+    }
+
+    private static final Pattern ACTIVITY_WORKER_PROPERTY_PATTERN =
+            Pattern.compile("^dt\\.dex-engine\\.activity-worker\\..+\\..+$");
+
+    private static List<String> getActivityWorkerNames(Config config) {
+        return StreamSupport.stream(config.getPropertyNames().spliterator(), false)
+                .filter(name -> ACTIVITY_WORKER_PROPERTY_PATTERN.matcher(name).matches())
+                .map(name -> name.split("\\.", 5)[3])
+                .distinct()
+                .toList();
+    }
+
+    private static boolean isTaskWorkerEnabled(Config config, TaskType taskType, String name) {
+        return config
+                .getOptionalValue(
+                        switch (taskType) {
+                            case ACTIVITY -> "dt.dex-engine.activity-worker.%s.enabled".formatted(name);
+                            case WORKFLOW -> "dt.dex-engine.workflow-worker.%s.enabled".formatted(name);
+                        },
+                        boolean.class)
+                .orElse(true);
+    }
+
+    private static TaskWorkerOptions getTaskWorkerOptions(Config config, TaskType type, String name) {
+        final var prefix = switch (type) {
+            case ACTIVITY -> "dt.dex-engine.activity-worker.%s.".formatted(name);
+            case WORKFLOW -> "dt.dex-engine.workflow-worker.%s.".formatted(name);
+        };
+
+        final var queueName = config.getValue(prefix + "queue-name", String.class);
+        final var maxConcurrency = config.getValue(prefix + "max-concurrency", int.class);
+        final var minPollInterval = config
+                .getOptionalValue(prefix + "min-poll-interval-ms", long.class)
+                .map(Duration::ofMillis)
+                .orElse(null);
+        final IntervalFunction pollBackoffFunction = getBackoffFunction(config, prefix).orElse(null);
+
+        var options = new TaskWorkerOptions(type, name, queueName, maxConcurrency);
+        if (minPollInterval != null) {
+            options = options.withMinPollInterval(minPollInterval);
+        }
+        if (pollBackoffFunction != null) {
+            options = options.withPollBackoffFunction(pollBackoffFunction);
+        }
+
+        return options;
+    }
+
     private static Optional<IntervalFunction> getBackoffFunction(Config config, String prefix) {
         final Optional<Long> initialDelayMillis = config.getOptionalValue(prefix + ".initial-delay-ms", long.class);
         final Optional<Double> multiplier = config.getOptionalValue(prefix + ".multiplier", double.class);

apiserver/src/main/java/org/dependencytrack/dex/activity/DeleteFilesActivity.java

@@ -0,0 +1,80 @@
+/*
+ * 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.dex.activity;
+
+import org.dependencytrack.dex.api.Activity;
+import org.dependencytrack.dex.api.ActivityContext;
+import org.dependencytrack.dex.api.ActivitySpec;
+import org.dependencytrack.dex.api.failure.TerminalApplicationFailureException;
+import org.dependencytrack.filestorage.api.FileStorage;
+import org.dependencytrack.filestorage.proto.v1.FileMetadata;
+import org.dependencytrack.plugin.PluginManager;
+import org.dependencytrack.proto.internal.workflow.v1.DeleteFilesArgument;
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @since 5.7.0
+ */
+@ActivitySpec(name = "delete-files")
+public final class DeleteFilesActivity implements Activity<DeleteFilesArgument, Void> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(DeleteFilesActivity.class);
+
+    private final PluginManager pluginManager;
+
+    public DeleteFilesActivity(PluginManager pluginManager) {
+        this.pluginManager = pluginManager;
+    }
+
+    @Override
+    public @Nullable Void execute(
+            ActivityContext ctx,
+            @Nullable DeleteFilesArgument argument) throws Exception {
+        if (argument == null) {
+            throw new TerminalApplicationFailureException("No argument provided");
+        }
+        if (argument.getFileMetadataCount() == 0) {
+            return null;
+        }
+
+        final Map<String, List<FileMetadata>> fileMetadataByProvider =
+                argument.getFileMetadataList().stream()
+                        .collect(Collectors.groupingBy(FileMetadata::getProviderName));
+
+        for (final String providerName : fileMetadataByProvider.keySet()) {
+            try (final var fileStorage = pluginManager.getExtension(FileStorage.class, providerName)) {
+
+                // TODO: Call fileStorage#deleteMany here once available.
+                for (final FileMetadata fileMetadata : fileMetadataByProvider.get(providerName)) {
+                    LOGGER.debug("Deleting file {}", fileMetadata.getLocation());
+                    fileStorage.delete(fileMetadata);
+                }
+            }
+        }
+
+        return null;
+    }
+
+}

apiserver/src/main/java/org/dependencytrack/dex/activity/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+@NullMarked
+package org.dependencytrack.dex.activity;
+
+import org.jspecify.annotations.NullMarked;