Add AggregateNotFoundError for commands/events on non-existent aggregates

main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java

@@ -24,6 +24,7 @@
 import org.elasticsoftware.akces.commands.Command;
 import org.elasticsoftware.akces.commands.CommandBus;
 import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent;
+import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent;
 import org.elasticsoftware.akces.errors.CommandExecutionErrorEvent;
 import org.elasticsoftware.akces.events.DomainEvent;
 import org.elasticsoftware.akces.events.ErrorEvent;
@@ -57,11 +58,15 @@ public class AggregateBeanFactoryPostProcessor implements BeanFactoryPostProcess
             new DomainEventType<>("CommandExecutionError", 1, CommandExecutionErrorEvent.class, false, false, true, false)
     );
     public static final List<DomainEventType<? extends DomainEvent>> COMMAND_HANDLER_SYSTEM_ERRORS = List.of(
+            new DomainEventType<>("AggregateNotFoundError", 1, AggregateNotFoundErrorEvent.class, false, false, true, false),
             new DomainEventType<>("CommandExecutionError", 1, CommandExecutionErrorEvent.class, false, false, true, false)
     );
     public static final List<DomainEventType<? extends DomainEvent>> EVENT_HANDLER_CREATE_SYSTEM_ERRORS = List.of(
             new DomainEventType<>("AggregateAlreadyExistsError", 1, AggregateAlreadyExistsErrorEvent.class, false, false, true, false)
     );
+    public static final List<DomainEventType<? extends DomainEvent>> EVENT_HANDLER_SYSTEM_ERRORS = List.of(
+            new DomainEventType<>("AggregateNotFoundError", 1, AggregateNotFoundErrorEvent.class, false, false, true, false)
+    );
 
     @Override
     @SuppressWarnings("unchecked")
@@ -469,7 +474,7 @@ private List<DomainEventType<?>> generateDomainEventTypes(Class<? extends Domain
     }
 
     private List<DomainEventType<?>> generateEventHandlerErrorEventTypes(Class<? extends DomainEvent>[] domainEventClasses, boolean isCreate) {
-        Stream<DomainEventType<? extends DomainEvent>> systemErrorEvents = (isCreate) ? EVENT_HANDLER_CREATE_SYSTEM_ERRORS.stream() : Stream.empty();
+        Stream<DomainEventType<? extends DomainEvent>> systemErrorEvents = (isCreate) ? EVENT_HANDLER_CREATE_SYSTEM_ERRORS.stream() : EVENT_HANDLER_SYSTEM_ERRORS.stream();
         return Stream.concat(Arrays.stream(domainEventClasses).map(eventClass -> {
             DomainEventInfo eventInfo = eventClass.getAnnotation(DomainEventInfo.class);
             return new DomainEventType<>(eventInfo.type(), eventInfo.version(), eventClass, false, false, true, hasPIIDataAnnotation(eventClass));

main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java

@@ -27,6 +27,7 @@
 import org.elasticsoftware.akces.commands.Command;
 import org.elasticsoftware.akces.commands.CommandBus;
 import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent;
+import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent;
 import org.elasticsoftware.akces.errors.CommandExecutionErrorEvent;
 import org.elasticsoftware.akces.events.DomainEvent;
 import org.elasticsoftware.akces.events.ErrorEvent;
@@ -161,6 +162,22 @@ private void aggregateAlreadyExists(ProtocolRecord commandOrDomainEventRecord,
         protocolRecordConsumer.accept(eventRecord);
     }
 
+    private void aggregateNotFound(ProtocolRecord commandOrDomainEventRecord,
+                                   Consumer<ProtocolRecord> protocolRecordConsumer) {
+        AggregateNotFoundErrorEvent errorEvent = new AggregateNotFoundErrorEvent(commandOrDomainEventRecord.aggregateId(), this.getName());
+        DomainEventType<?> type = getDomainEventType(errorEvent.getClass());
+        DomainEventRecord eventRecord = new DomainEventRecord(
+                commandOrDomainEventRecord.tenantId(),
+                type.typeName(),
+                type.version(),
+                serialize(errorEvent),
+                getEncoding(type),
+                errorEvent.getAggregateId(),
+                commandOrDomainEventRecord.correlationId(),
+                -1L); // ErrorEvents have no generation number because they don't alter the state
+        protocolRecordConsumer.accept(eventRecord);
+    }
+
     private void commandExecutionError(CommandRecord commandRecord,
                                        Consumer<ProtocolRecord> protocolRecordConsumer,
                                        Throwable exception) {
@@ -265,6 +282,15 @@ private void handleCommand(CommandType<?> commandType,
         // does have PIIData. in that case we need to load the proper GDPRContext
         Command command = materialize(commandType, commandRecord);
         AggregateStateRecord currentStateRecord = stateRecordSupplier.get();
+        // if we don't have state, this is an error and the aggregate does not exist
+        if (currentStateRecord == null) {
+            log.warn("Command {} wants to update a {} Aggregate with id {}, but the state does not exist. Generating a AggregateNotFoundError",
+                    commandRecord.name(),
+                    getName(),
+                    commandRecord.aggregateId());
+            aggregateNotFound(commandRecord, protocolRecordConsumer);
+            return;
+        }
         AggregateState currentState = materialize(currentStateRecord);
         Stream<DomainEvent> domainEvents = commandHandlers.get(commandType).apply(command, currentState);
         for (DomainEvent domainEvent : domainEvents.toList())
@@ -333,6 +359,15 @@ private void handleEvent(DomainEventType<?> eventType,
         // materialize the event
         DomainEvent externalEvent = materialize(eventType, domainEventRecord);
         AggregateStateRecord currentStateRecord = stateRecordSupplier.get();
+        // if we don't have state, this is an error and the aggregate does not exist
+        if (currentStateRecord == null) {
+            log.warn("External DomainEvent {} wants to update a {} Aggregate with id {}, but the state does not exist. Generating a AggregateNotFoundError",
+                    domainEventRecord.name(),
+                    getName(),
+                    domainEventRecord.aggregateId());
+            aggregateNotFound(domainEventRecord, protocolRecordConsumer);
+            return;
+        }
         AggregateState currentState = materialize(currentStateRecord);
         Stream<DomainEvent> domainEvents = eventHandlers.get(eventType).apply(externalEvent, currentState);
         for (DomainEvent domainEvent : domainEvents.toList())

main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java

@@ -48,6 +48,7 @@
 import org.elasticsoftware.akces.control.AggregateServiceRecord;
 import org.elasticsoftware.akces.control.AkcesControlRecord;
 import org.elasticsoftware.akces.errors.AggregateAlreadyExistsErrorEvent;
+import org.elasticsoftware.akces.errors.AggregateNotFoundErrorEvent;
 import org.elasticsoftware.akces.events.DomainEvent;
 import org.elasticsoftware.akces.gdpr.jackson.AkcesGDPRModule;
 import org.elasticsoftware.akces.protocol.*;
@@ -881,6 +882,29 @@ public void testAggregateAlreadyExistsErrorWithAkcesClient() throws ExecutionExc
 
     }
 
+    @Test
+    @Order(12)
+    public void testAggregateNotFoundErrorWithAkcesClient() throws ExecutionException, InterruptedException, TimeoutException {
+        // wait until the akces controller is running
+        while (!walletAggregateController.isRunning() ||
+                !accountAggregateController.isRunning() ||
+                !orderProcessManagerAggregateController.isRunning() ||
+                !akcesClient.isRunning()) {
+            Thread.onSpinWait();
+        }
+
+        // Generate a random wallet ID that doesn't exist
+        String nonExistentWalletId = UUID.randomUUID().toString();
+
+        // Try to credit a wallet that doesn't exist
+        CreditWalletCommand command = new CreditWalletCommand(nonExistentWalletId, "USD", new BigDecimal("100.00"));
+        List<DomainEvent> result = akcesClient.send("TEST_TENANT", command).toCompletableFuture().get(10, TimeUnit.SECONDS);
+
+        Assertions.assertNotNull(result);
+        Assertions.assertEquals(1, result.size());
+        assertInstanceOf(AggregateNotFoundErrorEvent.class, result.getFirst());
+    }
+
     public TopicDescription getTopicDescription(String topic) {
         try {
             return adminClient.describeTopics(topic).get(topic);

main/shared/src/main/java/org/elasticsoftware/akces/errors/AggregateNotFoundErrorEvent.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2022 - 2026 The Original Authors
+ *
+ * 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 org.elasticsoftware.akces.errors;
+
+import jakarta.validation.constraints.NotNull;
+import org.elasticsoftware.akces.annotations.DomainEventInfo;
+import org.elasticsoftware.akces.events.ErrorEvent;
+
+@DomainEventInfo(type = "AggregateNotFoundError")
+public record AggregateNotFoundErrorEvent(
+        @NotNull String aggregateId,
+        @NotNull String aggregateName
+) implements ErrorEvent {
+    @Override
+    public String getAggregateId() {
+        return aggregateId();
+    }
+}