diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/AggregateStateType.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/AggregateStateType.java index 4c8249d2..ef2cdab9 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/AggregateStateType.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/AggregateStateType.java @@ -25,5 +25,5 @@ public record AggregateStateType( boolean indexed, String indexName, boolean piiData -) { +) implements ProtocolRecordType { } diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/CommandType.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/CommandType.java index 954353f4..dd9917a4 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/CommandType.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/CommandType.java @@ -27,7 +27,7 @@ public record CommandType( boolean create, boolean external, boolean piiData -) implements SchemaType { +) implements SchemaType { @Override public String getSchemaPrefix() { diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/DomainEventType.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/DomainEventType.java index 330e495b..270abbbb 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/DomainEventType.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/DomainEventType.java @@ -28,7 +28,7 @@ public record DomainEventType( boolean external, boolean error, boolean piiData -) implements SchemaType { +) implements SchemaType { @Override public String getSchemaPrefix() { diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/ProtocolRecordType.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/ProtocolRecordType.java new file mode 100644 index 00000000..4a572538 --- /dev/null +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/ProtocolRecordType.java @@ -0,0 +1,26 @@ +/* + * Copyright 2022 - 2025 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.aggregate; + +public interface ProtocolRecordType { + String typeName(); + + int version(); + + Class typeClass(); +} diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/SchemaType.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/SchemaType.java index a3336c07..dc10ed68 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/SchemaType.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/SchemaType.java @@ -17,7 +17,7 @@ package org.elasticsoftware.akces.aggregate; -public interface SchemaType { +public interface SchemaType extends ProtocolRecordType { String getSchemaPrefix(); default String getSchemaName() { @@ -26,11 +26,5 @@ default String getSchemaName() { boolean relaxExternalValidation(); - String typeName(); - - int version(); - - Class typeClass(); - boolean external(); } diff --git a/main/api/src/main/java/org/elasticsoftware/akces/aggregate/UpcastingHandlerFunction.java b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/UpcastingHandlerFunction.java new file mode 100644 index 00000000..64c0b8be --- /dev/null +++ b/main/api/src/main/java/org/elasticsoftware/akces/aggregate/UpcastingHandlerFunction.java @@ -0,0 +1,35 @@ +/* + * Copyright 2022 - 2025 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.aggregate; + +@FunctionalInterface +public interface UpcastingHandlerFunction, RR extends ProtocolRecordType> { + R apply(T t); + + default TR getInputType() { + throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getInputType()"); + } + + default RR getOutputType() { + throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getOutputType()"); + } + + default Aggregate getAggregate() { + throw new UnsupportedOperationException("When implementing UpcastingHandlerFunction directly, you must override getAggregate()"); + } +} diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateInfo.java index 3e887c46..46382aa1 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateInfo.java @@ -32,7 +32,7 @@ @AliasFor(annotation = Component.class) String value(); - int version() default 1; + int stateVersion() default 1; boolean generateGDPRKeyOnCreate() default false; diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateStateInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateStateInfo.java index 83a8a41e..305b5aaf 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateStateInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/AggregateStateInfo.java @@ -17,6 +17,13 @@ package org.elasticsoftware.akces.annotations; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.TYPE}) public @interface AggregateStateInfo { String type(); diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/CommandInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/CommandInfo.java index f42bff8b..7df2816f 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/CommandInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/CommandInfo.java @@ -17,10 +17,13 @@ package org.elasticsoftware.akces.annotations; +import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; @Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.TYPE}) public @interface CommandInfo { String type(); diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/DomainEventInfo.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/DomainEventInfo.java index 735ba581..4144b490 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/annotations/DomainEventInfo.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/DomainEventInfo.java @@ -17,10 +17,13 @@ package org.elasticsoftware.akces.annotations; +import java.lang.annotation.ElementType; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; @Retention(RetentionPolicy.RUNTIME) +@Target({ElementType.TYPE}) public @interface DomainEventInfo { String type(); diff --git a/main/api/src/main/java/org/elasticsoftware/akces/annotations/UpcastingHandler.java b/main/api/src/main/java/org/elasticsoftware/akces/annotations/UpcastingHandler.java new file mode 100644 index 00000000..04b15606 --- /dev/null +++ b/main/api/src/main/java/org/elasticsoftware/akces/annotations/UpcastingHandler.java @@ -0,0 +1,28 @@ +/* + * Copyright 2022 - 2025 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.annotations; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Target({ElementType.METHOD}) +@Retention(RetentionPolicy.RUNTIME) +public @interface UpcastingHandler { +} diff --git a/main/api/src/main/java/org/elasticsoftware/akces/commands/CommandBusHolder.java b/main/api/src/main/java/org/elasticsoftware/akces/commands/CommandBusHolder.java index a95c3469..cb0629c1 100644 --- a/main/api/src/main/java/org/elasticsoftware/akces/commands/CommandBusHolder.java +++ b/main/api/src/main/java/org/elasticsoftware/akces/commands/CommandBusHolder.java @@ -25,6 +25,7 @@ public class CommandBusHolder { protected CommandBusHolder() { } + @SuppressWarnings("rawtypes") public static CommandBus getCommandBus(Class aggregateClass) { return commandBusThreadLocal.get(); } diff --git a/main/client/src/test/java/org/elasticsoftware/akces/client/AkcesClientTests.java b/main/client/src/test/java/org/elasticsoftware/akces/client/AkcesClientTests.java index 9c9b1cd8..2e3a640c 100644 --- a/main/client/src/test/java/org/elasticsoftware/akces/client/AkcesClientTests.java +++ b/main/client/src/test/java/org/elasticsoftware/akces/client/AkcesClientTests.java @@ -120,6 +120,7 @@ public class AkcesClientTests { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); diff --git a/main/query-support/src/main/java/org/elasticsoftware/akces/query/database/beans/DatabaseModelEventHandlerFunctionAdapter.java b/main/query-support/src/main/java/org/elasticsoftware/akces/query/database/beans/DatabaseModelEventHandlerFunctionAdapter.java index 93b54d73..032fd3d4 100644 --- a/main/query-support/src/main/java/org/elasticsoftware/akces/query/database/beans/DatabaseModelEventHandlerFunctionAdapter.java +++ b/main/query-support/src/main/java/org/elasticsoftware/akces/query/database/beans/DatabaseModelEventHandlerFunctionAdapter.java @@ -1,21 +1,21 @@ /* - * Copyright 2022 - 2025 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. - * - */ + * Copyright 2022 - 2025 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.query.database.beans; + package org.elasticsoftware.akces.query.database.beans; import jakarta.validation.constraints.NotNull; import org.elasticsoftware.akces.aggregate.DomainEventType; @@ -24,70 +24,68 @@ import org.elasticsoftware.akces.query.DatabaseModel; import org.elasticsoftware.akces.query.DatabaseModelEventHandlerFunction; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; -public class DatabaseModelEventHandlerFunctionAdapter - implements DatabaseModelEventHandlerFunction { - private final DatabaseModel databaseModel; - private final String adapterMethodName; - private final Class domainEventClass; - private final DomainEventType domainEventType; - private Method adapterMethod; + public class DatabaseModelEventHandlerFunctionAdapter + implements DatabaseModelEventHandlerFunction { + private final DatabaseModel databaseModel; + private final String adapterMethodName; + private final Class domainEventClass; + private final DomainEventType domainEventType; + private MethodHandle adapterMethodHandle; - public DatabaseModelEventHandlerFunctionAdapter(DatabaseModel databaseModel, - String adapterMethodName, - Class domainEventClass, - String typeName, - int version) { - this.databaseModel = databaseModel; - this.adapterMethodName = adapterMethodName; - this.domainEventClass = domainEventClass; - this.domainEventType = new DomainEventType<>( - typeName, - version, - domainEventClass, - false, - true, - false, - GDPRAnnotationUtils.hasPIIDataAnnotation(domainEventClass)); - } + public DatabaseModelEventHandlerFunctionAdapter(DatabaseModel databaseModel, + String adapterMethodName, + Class domainEventClass, + String typeName, + int version) { + this.databaseModel = databaseModel; + this.adapterMethodName = adapterMethodName; + this.domainEventClass = domainEventClass; + this.domainEventType = new DomainEventType<>( + typeName, + version, + domainEventClass, + false, + true, + false, + GDPRAnnotationUtils.hasPIIDataAnnotation(domainEventClass)); + } - @SuppressWarnings("unused") - public void init() { - try { - adapterMethod = databaseModel.getClass().getMethod(adapterMethodName, domainEventClass); - } catch (NoSuchMethodException e) { - throw new RuntimeException(e); - } - } + @SuppressWarnings("unused") + public void init() { + try { + MethodHandles.Lookup lookup = MethodHandles.lookup(); + MethodType methodType = MethodType.methodType(void.class, domainEventClass); + adapterMethodHandle = lookup.findVirtual(databaseModel.getClass(), adapterMethodName, methodType); + } catch (NoSuchMethodException | IllegalAccessException e) { + throw new RuntimeException("Failed to find method " + adapterMethodName + " on " + + databaseModel.getClass().getName(), e); + } + } - @Override - public void accept(@NotNull E event) { - try { - adapterMethod.invoke(databaseModel, event); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } - } else { - throw new RuntimeException(e); - } - } - } + @Override + public void accept(@NotNull E event) { + try { + adapterMethodHandle.invoke(databaseModel, event); + } catch (Throwable e) { + if (e instanceof RuntimeException) { + throw (RuntimeException) e; + } else { + throw new RuntimeException("Error invoking event handler", e); + } + } + } - @Override - public DomainEventType getEventType() { - return domainEventType; - } + @Override + public DomainEventType getEventType() { + return domainEventType; + } - @Override - public DatabaseModel getDatabaseModel() { - return databaseModel; - } -} + @Override + public DatabaseModel getDatabaseModel() { + return databaseModel; + } + } \ No newline at end of file diff --git a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/AkcesQueryModelController.java b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/AkcesQueryModelController.java index 76d51c90..b0530423 100644 --- a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/AkcesQueryModelController.java +++ b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/AkcesQueryModelController.java @@ -160,9 +160,9 @@ private void process(Consumer indexConsumer) { if(!newExecutions.isEmpty()){ logger.info("Processing {} new HydrationExecutions", newExecutions.size()); // we need to get the endoffsets for all the partitions - indexConsumer.endOffsets(newExecutions.keySet()).forEach((partition, endOffset) -> { - hydrationExecutions.computeIfPresent(partition, (topicPartition, hydrationExecution) -> hydrationExecution.withEndOffset(endOffset)); - }); + indexConsumer.endOffsets(newExecutions.keySet()).forEach((partition, endOffset) -> + hydrationExecutions.computeIfPresent(partition, (topicPartition, hydrationExecution) -> + hydrationExecution.withEndOffset(endOffset))); } // seek to the correct offset for the new executions newExecutions.forEach((partition, execution) -> { diff --git a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeFactory.java b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeFactory.java index 64a1d9e3..5513d14c 100644 --- a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeFactory.java +++ b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeFactory.java @@ -65,7 +65,7 @@ private QueryModelRuntime createRuntime(QueryModel queryModel) { QueryModelInfo queryModelInfo = queryModel.getClass().getAnnotation(QueryModelInfo.class); if (queryModelInfo != null) { - runtimeBuilder.setStateType(new QueryModelStateType( + runtimeBuilder.setStateType(new QueryModelStateType<>( queryModelInfo.value(), queryModelInfo.version(), queryModel.getStateClass(), diff --git a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/beans/QueryModelEventHandlerFunctionAdapter.java b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/beans/QueryModelEventHandlerFunctionAdapter.java index e0e442b6..2135955a 100644 --- a/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/beans/QueryModelEventHandlerFunctionAdapter.java +++ b/main/query-support/src/main/java/org/elasticsoftware/akces/query/models/beans/QueryModelEventHandlerFunctionAdapter.java @@ -24,8 +24,9 @@ import org.elasticsoftware.akces.query.QueryModelEventHandlerFunction; import org.elasticsoftware.akces.query.QueryModelState; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; import static org.elasticsoftware.akces.gdpr.GDPRAnnotationUtils.hasPIIDataAnnotation; @@ -37,7 +38,7 @@ public class QueryModelEventHandlerFunctionAdapter stateClass; private final boolean create; private final DomainEventType domainEventType; - private Method adapterMethod; + private MethodHandle adapterMethodHandle; public QueryModelEventHandlerFunctionAdapter(QueryModel queryModel, String adapterMethodName, @@ -64,27 +65,24 @@ public QueryModelEventHandlerFunctionAdapter(QueryModel queryModel, @SuppressWarnings("unused") public void init() { try { - adapterMethod = queryModel.getClass().getMethod(adapterMethodName, domainEventClass, stateClass); - } catch (NoSuchMethodException e) { - throw new RuntimeException(e); + MethodHandles.Lookup lookup = MethodHandles.lookup(); + MethodType methodType = MethodType.methodType(stateClass, domainEventClass, stateClass); + adapterMethodHandle = lookup.findVirtual(queryModel.getClass(), adapterMethodName, methodType); + } catch (NoSuchMethodException | IllegalAccessException e) { + throw new RuntimeException("Failed to find method " + adapterMethodName + " on " + + queryModel.getClass().getName(), e); } } @Override public @NotNull S apply(@NotNull E event, S state) { try { - return (S) adapterMethod.invoke(queryModel, event, state); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } + return (S) adapterMethodHandle.invoke(queryModel, event, state); + } catch (Throwable e) { + if (e instanceof RuntimeException) { + throw (RuntimeException) e; } else { - throw new RuntimeException(e); + throw new RuntimeException("Error invoking event handler", e); } } } @@ -103,4 +101,4 @@ public QueryModel getQueryModel() { public boolean isCreate() { return create; } -} +} \ No newline at end of file diff --git a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelRuntimeTests.java b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelRuntimeTests.java index 302a366b..9019bbbc 100644 --- a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelRuntimeTests.java +++ b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelRuntimeTests.java @@ -116,6 +116,7 @@ public class DatabaseModelRuntimeTests { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); diff --git a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelTestConfiguration.java b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelTestConfiguration.java index cd449151..09e40485 100644 --- a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelTestConfiguration.java +++ b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/DatabaseModelTestConfiguration.java @@ -20,7 +20,6 @@ import liquibase.integration.spring.SpringLiquibase; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.Configuration; import javax.sql.DataSource; diff --git a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/model/DefaultJdbcModel.java b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/model/DefaultJdbcModel.java index cc11e376..e47e472a 100644 --- a/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/model/DefaultJdbcModel.java +++ b/main/query-support/src/test/java/org/elasticsoftware/akces/query/database/model/DefaultJdbcModel.java @@ -17,7 +17,6 @@ package org.elasticsoftware.akces.query.database.model; -import jakarta.inject.Inject; import org.elasticsoftware.akces.annotations.DatabaseModelEventHandler; import org.elasticsoftware.akces.annotations.DatabaseModelInfo; import org.elasticsoftware.akces.query.database.jdbc.JdbcDatabaseModel; diff --git a/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeTests.java b/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeTests.java index 155c127b..3f4dbd70 100644 --- a/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeTests.java +++ b/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelRuntimeTests.java @@ -39,7 +39,6 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.elasticsoftware.akces.AggregateServiceApplication; import org.elasticsoftware.akces.AkcesAggregateController; -import org.elasticsoftware.akces.annotations.DatabaseModelInfo; import org.elasticsoftware.akces.annotations.DomainEventInfo; import org.elasticsoftware.akces.client.AkcesClientController; import org.elasticsoftware.akces.control.AggregateServiceRecord; @@ -59,14 +58,11 @@ import org.elasticsoftware.akcestest.aggregate.wallet.*; import org.junit.jupiter.api.*; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.boot.test.autoconfigure.filter.TypeExcludeFilters; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextException; import org.springframework.context.ApplicationContextInitializer; import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.FilterType; import org.springframework.http.converter.json.Jackson2ObjectMapperBuilder; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaAdmin; @@ -125,6 +121,7 @@ public class QueryModelRuntimeTests { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); @@ -558,7 +555,7 @@ public void testAccountQueryModelUsingCache() { Assertions.assertNotNull(result); Assertions.assertEquals(1, result.size()); - assertInstanceOf(BalanceCreatedEvent.class, result.get(0)); + assertInstanceOf(BalanceCreatedEvent.class, result.getFirst()); CompletableFuture walletStateFuture4 = akcesQueryModelController.getHydratedState(WalletQueryModel.class, userId) .toCompletableFuture(); @@ -576,9 +573,9 @@ public void testAccountQueryModelUsingCache() { Assertions.assertNotNull(result); Assertions.assertEquals(1, result.size()); - assertInstanceOf(WalletCreditedEvent.class, result.get(0)); - assertEquals("EUR", ((WalletCreditedEvent) result.get(0)).currency()); - assertEquals(new BigDecimal("1000.00"), ((WalletCreditedEvent) result.get(0)).amount()); + assertInstanceOf(WalletCreditedEvent.class, result.getFirst()); + assertEquals("EUR", ((WalletCreditedEvent) result.getFirst()).currency()); + assertEquals(new BigDecimal("1000.00"), ((WalletCreditedEvent) result.getFirst()).amount()); CompletableFuture walletStateFuture5 = akcesQueryModelController.getHydratedState(WalletQueryModel.class, userId) .toCompletableFuture(); diff --git a/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelTestConfiguration.java b/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelTestConfiguration.java index ffac3696..66ad3d25 100644 --- a/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelTestConfiguration.java +++ b/main/query-support/src/test/java/org/elasticsoftware/akces/query/models/QueryModelTestConfiguration.java @@ -18,11 +18,8 @@ package org.elasticsoftware.akces.query.models; import org.elasticsoftware.akces.annotations.DatabaseModelInfo; -import org.springframework.boot.test.autoconfigure.filter.TypeExcludeFilters; import org.springframework.context.annotation.ComponentScan; -import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.FilterType; -import org.springframework.test.context.ContextConfiguration; //@Configuration @ComponentScan(basePackages = { diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/AkcesAggregateController.java b/main/runtime/src/main/java/org/elasticsoftware/akces/AkcesAggregateController.java index 245a39a6..26dfe1de 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/AkcesAggregateController.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/AkcesAggregateController.java @@ -323,7 +323,7 @@ private void process() { } } - private boolean protocolRecordTypeNotYetProduced(SchemaType schemaType, + private boolean protocolRecordTypeNotYetProduced(SchemaType schemaType, BiFunction createTopicPartition) { try (Consumer consumer = consumerFactory.createConsumer( aggregateRuntime.getName() + "-Akces-Control-TypeCheck", @@ -508,7 +508,7 @@ public CommandType resolveType(@Nonnull Class commandClass .filter(commandServiceRecord -> supportsCommand(commandServiceRecord.supportedCommands(), commandInfo)) .toList(); if (services.size() == 1) { - AggregateServiceRecord aggregateServiceRecord = services.get(0); + AggregateServiceRecord aggregateServiceRecord = services.getFirst(); if (aggregateRuntime.getName().equals(aggregateServiceRecord.aggregateName())) { // this is a local command (will be sent to self) return aggregateRuntime.getLocalCommandType(commandInfo.type(), commandInfo.version()); diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/aggregate/AggregateRuntime.java b/main/runtime/src/main/java/org/elasticsoftware/akces/aggregate/AggregateRuntime.java index 99165978..a238458e 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/aggregate/AggregateRuntime.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/aggregate/AggregateRuntime.java @@ -36,7 +36,7 @@ public interface AggregateRuntime { String getName(); - Class getAggregateClass(); + Class> getAggregateClass(); void handleCommandRecord(CommandRecord commandRecord, Consumer protocolRecordConsumer, diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java index f9545165..b66df4f6 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateBeanFactoryPostProcessor.java @@ -71,7 +71,7 @@ public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) logger.info("Processing Aggregate bean {}", beanName); BeanDefinition bd = beanFactory.getBeanDefinition(beanName); try { - Class aggregateClass = Class.forName(bd.getBeanClassName()); + final Class aggregateClass = Class.forName(bd.getBeanClassName()); List commandHandlers = Arrays.stream(aggregateClass.getMethods()) .filter(method -> method.isAnnotationPresent(CommandHandler.class)) .toList(); @@ -84,6 +84,9 @@ public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) List eventBridgeHandlers = Arrays.stream(aggregateClass.getMethods()) .filter(method -> method.isAnnotationPresent(EventBridgeHandler.class)) .toList(); + List upcastingHandlers = Arrays.stream(aggregateClass.getMethods()) + .filter(method -> method.isAnnotationPresent(UpcastingHandler.class)) + .toList(); commandHandlers.forEach(commandHandlerMethod -> processCommandHandler(beanName, commandHandlerMethod, bdr)); eventHandlers.forEach(eventHandlerMethod -> @@ -92,6 +95,8 @@ public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) processEventSourcingHandler(beanName, eventSourcingHandlerMethod, bdr)); eventBridgeHandlers.forEach(eventBridgeHandlerMethod -> processEventBridgeHandler(beanName, eventBridgeHandlerMethod, bdr)); + upcastingHandlers.forEach(upcastingHandlerMethod -> + processUpcastingHandler(beanName, aggregateClass, upcastingHandlerMethod, bdr)); } catch (ClassNotFoundException e) { throw new ApplicationContextException("Unable to load class for bean " + beanName, e); } @@ -231,6 +236,109 @@ private void processEventBridgeHandler(String aggregateBeanName, Method eventBri } } + private void processUpcastingHandler(String aggregateBeanName, + Class aggregateClass, + Method upcastingHandlerMethod, + BeanDefinitionRegistry bdr) { + // Handle domain event upcasting + if (upcastingHandlerMethod.getParameterCount() == 1 && + DomainEvent.class.isAssignableFrom(upcastingHandlerMethod.getParameterTypes()[0]) && + DomainEvent.class.isAssignableFrom(upcastingHandlerMethod.getReturnType())) { + + Class inputEventClass = upcastingHandlerMethod.getParameterTypes()[0]; + Class outputEventClass = upcastingHandlerMethod.getReturnType(); + + DomainEventInfo inputEventInfo = inputEventClass.getAnnotation(DomainEventInfo.class); + if (inputEventInfo == null) { + throw new IllegalArgumentException("Input event class " + inputEventClass.getName() + + " must be annotated with @DomainEventInfo"); + } + + DomainEventInfo outputEventInfo = outputEventClass.getAnnotation(DomainEventInfo.class); + if (outputEventInfo == null) { + throw new IllegalArgumentException("Output event class " + outputEventClass.getName() + + " must be annotated with @DomainEventInfo"); + } + // see if this is a valid upcaster configuration + if (!inputEventInfo.type().equals(outputEventInfo.type())) { + throw new IllegalArgumentException("Input event type " + inputEventInfo.type() + + " does not match output event type " + outputEventInfo.type()); + } + if(outputEventInfo.version() - inputEventInfo.version() != 1) { + throw new IllegalArgumentException("Output event version " + outputEventInfo.version() + + " must be one greater than input event version " + inputEventInfo.version()); + } + + // Generate bean name based on method name and event info + String beanName = aggregateBeanName + "_duh_" + upcastingHandlerMethod.getName() + + "_" + inputEventInfo.type() + "_" + inputEventInfo.version() + + "_to_" + outputEventInfo.version(); + + // we need to figure out if the event is internal or external + // we need to find the DomainEvent that is actually handled by either an @EventHandler or @EventSourcingHandler + boolean externalEvent = Arrays.stream(aggregateClass.getMethods()) + .filter(method -> method.isAnnotationPresent(EventSourcingHandler.class)) + .map(method -> method.getParameterTypes()[0]) + .noneMatch(eventClass -> eventClass.getAnnotation(DomainEventInfo.class).type().equals(outputEventInfo.type())); + + bdr.registerBeanDefinition(beanName, + BeanDefinitionBuilder.genericBeanDefinition(DomainEventUpcastingHandlerFunctionAdapter.class) + .addConstructorArgReference(aggregateBeanName) + .addConstructorArgValue(upcastingHandlerMethod.getName()) + .addConstructorArgValue(inputEventClass) + .addConstructorArgValue(outputEventClass) + .addConstructorArgValue(externalEvent) + .setInitMethodName("init") + .getBeanDefinition()); + + // Handle aggregate state upcasting + } else if (upcastingHandlerMethod.getParameterCount() == 1 && + AggregateState.class.isAssignableFrom(upcastingHandlerMethod.getParameterTypes()[0]) && + AggregateState.class.isAssignableFrom(upcastingHandlerMethod.getReturnType())) { + + Class inputStateClass = upcastingHandlerMethod.getParameterTypes()[0]; + Class outputStateClass = upcastingHandlerMethod.getReturnType(); + + AggregateStateInfo inputStateInfo = inputStateClass.getAnnotation(AggregateStateInfo.class); + if (inputStateInfo == null) { + throw new IllegalArgumentException("Input state class " + inputStateClass.getName() + + " must be annotated with @AggregateStateInfo"); + } + + AggregateStateInfo outputStateInfo = outputStateClass.getAnnotation(AggregateStateInfo.class); + if (outputStateInfo == null) { + throw new IllegalArgumentException("Output state class " + outputStateClass.getName() + + " must be annotated with @AggregateStateInfo"); + } + + // see if this is a valid upcaster configuration + if (!inputStateInfo.type().equals(outputStateInfo.type())) { + throw new IllegalArgumentException("Input state type " + inputStateInfo.type() + + " does not match output state type " + outputStateInfo.type()); + } + if(outputStateInfo.version() - inputStateInfo.version() != 1) { + throw new IllegalArgumentException("Output state version " + outputStateInfo.version() + + " must be one greater than input state version " + inputStateInfo.version()); + } + + // Generate bean name based on method name and state info + String beanName = aggregateBeanName + "_suh_" + upcastingHandlerMethod.getName() + + "_" + inputStateInfo.type() + "_" + inputStateInfo.version() + + "_to_" + outputStateInfo.version(); + + bdr.registerBeanDefinition(beanName, + BeanDefinitionBuilder.genericBeanDefinition(AggregateStateUpcastingHandlerFunctionAdapter.class) + .addConstructorArgReference(aggregateBeanName) + .addConstructorArgValue(upcastingHandlerMethod.getName()) + .addConstructorArgValue(inputStateClass) + .addConstructorArgValue(outputStateClass) + .setInitMethodName("init") + .getBeanDefinition()); + } else { + throw new ApplicationContextException("Invalid UpcastingHandler method signature: " + upcastingHandlerMethod); + } + } + private List> generateDomainEventTypes(Class[] domainEventClasses, boolean isCreate) { return Arrays.stream(domainEventClasses).map(eventClass -> { diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateStateUpcastingHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateStateUpcastingHandlerFunctionAdapter.java new file mode 100644 index 00000000..b50c7e88 --- /dev/null +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/AggregateStateUpcastingHandlerFunctionAdapter.java @@ -0,0 +1,131 @@ +/* + * Copyright 2022 - 2025 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.beans; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.aggregate.Aggregate; +import org.elasticsoftware.akces.aggregate.AggregateState; +import org.elasticsoftware.akces.aggregate.AggregateStateType; +import org.elasticsoftware.akces.aggregate.UpcastingHandlerFunction; +import org.elasticsoftware.akces.annotations.AggregateInfo; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; + +import static org.elasticsoftware.akces.gdpr.GDPRAnnotationUtils.hasPIIDataAnnotation; + +public class AggregateStateUpcastingHandlerFunctionAdapter + implements UpcastingHandlerFunction, AggregateStateType> { + + private final Aggregate aggregate; + private final String adapterMethodName; + private final Class inputStateClass; + private final Class outputStateClass; + private final AggregateStateType inputStateType; + private final AggregateStateType outputStateType; + private MethodHandle methodHandle; + + public AggregateStateUpcastingHandlerFunctionAdapter( + Aggregate aggregate, + String adapterMethodName, + Class inputStateClass, + Class outputStateClass) { + AggregateStateInfo inputStateInfo = inputStateClass.getAnnotation(AggregateStateInfo.class); + if (inputStateInfo == null) { + throw new IllegalArgumentException("Input state class " + inputStateClass.getName() + + " must be annotated with @AggregateStateInfo"); + } + AggregateStateInfo outputStateInfo = outputStateClass.getAnnotation(AggregateStateInfo.class); + if (outputStateInfo == null) { + throw new IllegalArgumentException("Output state class " + outputStateClass.getName() + + " must be annotated with @AggregateStateInfo"); + } + AggregateInfo aggregateInfo = aggregate.getClass().getAnnotation(AggregateInfo.class); + if (aggregateInfo == null) { + throw new IllegalArgumentException("Aggregate class " + aggregate.getClass().getName() + + " must be annotated with @AggregateInfo"); + } + this.aggregate = aggregate; + this.adapterMethodName = adapterMethodName; + this.inputStateClass = inputStateClass; + this.outputStateClass = outputStateClass; + this.inputStateType = new AggregateStateType<>( + inputStateInfo.type(), + inputStateInfo.version(), + inputStateClass, + aggregateInfo.generateGDPRKeyOnCreate(), + aggregateInfo.indexed(), + aggregateInfo.indexName(), + hasPIIDataAnnotation(inputStateClass) + ); + this.outputStateType = new AggregateStateType<>( + outputStateInfo.type(), + outputStateInfo.version(), + outputStateClass, + aggregateInfo.generateGDPRKeyOnCreate(), + aggregateInfo.indexed(), + aggregateInfo.indexName(), + hasPIIDataAnnotation(outputStateClass) + ); + } + + @SuppressWarnings("unused") + public void init() { + try { + methodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(outputStateClass, inputStateClass)); + } catch (NoSuchMethodException | IllegalAccessException e) { + throw new RuntimeException(e); + } + } + + @Override + @SuppressWarnings("unchecked") + public R apply(@NotNull T state) { + try { + return (R) methodHandle.invoke(aggregate, state); + } catch (WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException runtimeException) { + throw runtimeException; + } + throw new RuntimeException(e); + } + } + + @Override + public AggregateStateType getInputType() { + return inputStateType; + } + + @Override + public AggregateStateType getOutputType() { + return outputStateType; + } + + @Override + public Aggregate getAggregate() { + return aggregate; + } +} diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/CommandHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/CommandHandlerFunctionAdapter.java index 948cfb5f..fe895028 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/CommandHandlerFunctionAdapter.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/CommandHandlerFunctionAdapter.java @@ -21,8 +21,10 @@ import org.elasticsoftware.akces.commands.Command; import org.elasticsoftware.akces.events.DomainEvent; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; import java.util.List; import java.util.stream.Stream; @@ -38,7 +40,7 @@ public class CommandHandlerFunctionAdapter commandType; private final List> producedDomainEventTypes; private final List> errorEventTypes; - private Method adapterMethod; + private MethodHandle adapterMethodHandle; public CommandHandlerFunctionAdapter(Aggregate aggregate, String adapterMethodName, @@ -67,25 +69,25 @@ public CommandHandlerFunctionAdapter(Aggregate aggregate, @SuppressWarnings("unused") public void init() { try { - adapterMethod = aggregate.getClass().getMethod(adapterMethodName, commandClass, stateClass); - } catch (NoSuchMethodException e) { + adapterMethodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(Stream.class, commandClass, stateClass)); + } catch (NoSuchMethodException | IllegalAccessException e) { throw new RuntimeException(e); } } @Override + @SuppressWarnings("unchecked") public Stream apply(C command, S state) { try { - return (Stream) adapterMethod.invoke(aggregate, command, state); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } + return (Stream) adapterMethodHandle.invoke(aggregate, command, state); + } catch(WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException) { + throw (RuntimeException) e; } else { throw new RuntimeException(e); } diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/DomainEventUpcastingHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/DomainEventUpcastingHandlerFunctionAdapter.java new file mode 100644 index 00000000..f3298d5a --- /dev/null +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/DomainEventUpcastingHandlerFunctionAdapter.java @@ -0,0 +1,124 @@ +/* + * Copyright 2022 - 2025 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.beans; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.aggregate.Aggregate; +import org.elasticsoftware.akces.aggregate.AggregateState; +import org.elasticsoftware.akces.aggregate.DomainEventType; +import org.elasticsoftware.akces.aggregate.UpcastingHandlerFunction; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.events.DomainEvent; +import org.elasticsoftware.akces.events.ErrorEvent; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; + +import static org.elasticsoftware.akces.gdpr.GDPRAnnotationUtils.hasPIIDataAnnotation; + +public class DomainEventUpcastingHandlerFunctionAdapter + implements UpcastingHandlerFunction, DomainEventType> { + private final Aggregate aggregate; + private final String adapterMethodName; + private final Class inputEventClass; + private final Class outputEventClass; + private final DomainEventType inputEventType; + private final DomainEventType outputEventType; + private MethodHandle methodHandle; + + public DomainEventUpcastingHandlerFunctionAdapter(Aggregate aggregate, + String adapterMethodName, + Class inputEventClass, + Class outputEventClass, + boolean external) { + DomainEventInfo inputEventInfo = inputEventClass.getAnnotation(DomainEventInfo.class); + if (inputEventInfo == null) { + throw new IllegalArgumentException("Input event class " + inputEventClass.getName() + + " must be annotated with @DomainEventInfo"); + } + + DomainEventInfo outputEventInfo = outputEventClass.getAnnotation(DomainEventInfo.class); + if (outputEventInfo == null) { + throw new IllegalArgumentException("Output event class " + outputEventClass.getName() + + " must be annotated with @DomainEventInfo"); + } + + this.aggregate = aggregate; + this.adapterMethodName = adapterMethodName; + this.inputEventClass = inputEventClass; + this.inputEventType = new DomainEventType<>( + inputEventInfo.type(), + inputEventInfo.version(), + inputEventClass, + false, + external, + ErrorEvent.class.isAssignableFrom(inputEventClass), + hasPIIDataAnnotation(inputEventClass)); + this.outputEventClass = outputEventClass; + this.outputEventType = new DomainEventType<>( + outputEventInfo.type(), + outputEventInfo.version(), + outputEventClass, + false, + external, + ErrorEvent.class.isAssignableFrom(outputEventClass), + hasPIIDataAnnotation(outputEventClass)); + } + + @SuppressWarnings("unused") + public void init() { + try { + methodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(outputEventClass, inputEventClass)); + } catch (NoSuchMethodException | IllegalAccessException e) { + throw new RuntimeException(e); + } + } + + @Override + @SuppressWarnings("unchecked") + public R apply(@NotNull T event) { + try { + return (R) methodHandle.invoke(aggregate, event); + } catch (WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException runtimeException) { + throw runtimeException; + } + throw new RuntimeException(e); + } + } + + public DomainEventType getInputType() { + return inputEventType; + } + + public DomainEventType getOutputType() { + return outputEventType; + } + + @Override + public Aggregate getAggregate() { + return aggregate; + } +} \ No newline at end of file diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventBridgeHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventBridgeHandlerFunctionAdapter.java index 2675b5e4..e91cb98f 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventBridgeHandlerFunctionAdapter.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventBridgeHandlerFunctionAdapter.java @@ -26,8 +26,10 @@ import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akces.events.ErrorEvent; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; import static org.elasticsoftware.akces.gdpr.GDPRAnnotationUtils.hasPIIDataAnnotation; @@ -36,7 +38,7 @@ public class EventBridgeHandlerFunctionAdapter inputEventClass; private final DomainEventType domainEventType; - private Method adapterMethod; + private MethodHandle methodHandle; public EventBridgeHandlerFunctionAdapter(Aggregate aggregate, String adapterMethodName, @@ -59,8 +61,11 @@ public EventBridgeHandlerFunctionAdapter(Aggregate aggregate, @SuppressWarnings("unused") public void init() { try { - adapterMethod = aggregate.getClass().getMethod(adapterMethodName, inputEventClass, CommandBus.class); - } catch (NoSuchMethodException e) { + methodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(void.class, inputEventClass, CommandBus.class)); + } catch (NoSuchMethodException | IllegalAccessException e) { throw new RuntimeException(e); } } @@ -68,19 +73,14 @@ public void init() { @Override public void apply(@NotNull E event, CommandBus commandBus) { try { - adapterMethod.invoke(aggregate, event, commandBus); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } - } else { - throw new RuntimeException(e); + methodHandle.invoke(aggregate, event, commandBus); + } catch (WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException runtimeException) { + throw runtimeException; } + throw new RuntimeException(e); } } @@ -93,5 +93,4 @@ public DomainEventType getEventType() { public Aggregate getAggregate() { return aggregate; } - -} +} \ No newline at end of file diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventHandlerFunctionAdapter.java index 4ecbf48e..699f3504 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventHandlerFunctionAdapter.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventHandlerFunctionAdapter.java @@ -25,8 +25,10 @@ import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akces.events.ErrorEvent; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; import java.util.List; import java.util.stream.Stream; @@ -41,7 +43,7 @@ public class EventHandlerFunctionAdapter> producedDomainEventTypes; private final List> errorEventTypes; private final DomainEventType domainEventType; - private Method adapterMethod; + private MethodHandle methodHandle; public EventHandlerFunctionAdapter(Aggregate aggregate, String adapterMethodName, @@ -72,28 +74,27 @@ public EventHandlerFunctionAdapter(Aggregate aggregate, @SuppressWarnings("unused") public void init() { try { - adapterMethod = aggregate.getClass().getMethod(adapterMethodName, inputEventClass, stateClass); - } catch (NoSuchMethodException e) { + methodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(Stream.class, inputEventClass, stateClass)); + } catch (NoSuchMethodException | IllegalAccessException e) { throw new RuntimeException(e); } } @Override + @SuppressWarnings("unchecked") public Stream apply(@NotNull InputEvent event, S state) { try { - return (Stream) adapterMethod.invoke(aggregate, event, state); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } - } else { - throw new RuntimeException(e); + return (Stream) methodHandle.invoke(aggregate, event, state); + } catch (WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException runtimeException) { + throw runtimeException; } + throw new RuntimeException(e); } } @@ -121,4 +122,4 @@ public List> getProducedDomainEventTypes() { public List> getErrorEventTypes() { return errorEventTypes; } -} +} \ No newline at end of file diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventSourcingHandlerFunctionAdapter.java b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventSourcingHandlerFunctionAdapter.java index 19ea4ec5..909a43ed 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventSourcingHandlerFunctionAdapter.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/beans/EventSourcingHandlerFunctionAdapter.java @@ -24,8 +24,10 @@ import org.elasticsoftware.akces.aggregate.EventSourcingHandlerFunction; import org.elasticsoftware.akces.events.DomainEvent; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.lang.invoke.WrongMethodTypeException; import static org.elasticsoftware.akces.gdpr.GDPRAnnotationUtils.hasPIIDataAnnotation; @@ -36,7 +38,7 @@ public class EventSourcingHandlerFunctionAdapter stateClass; private final boolean create; private final DomainEventType domainEventType; - private Method adapterMethod; + private MethodHandle methodHandle; public EventSourcingHandlerFunctionAdapter(Aggregate aggregate, String adapterMethodName, @@ -62,28 +64,27 @@ public EventSourcingHandlerFunctionAdapter(Aggregate aggregate, @SuppressWarnings("unused") public void init() { try { - adapterMethod = aggregate.getClass().getMethod(adapterMethodName, domainEventClass, stateClass); - } catch (NoSuchMethodException e) { + methodHandle = MethodHandles.lookup().findVirtual( + aggregate.getClass(), + adapterMethodName, + MethodType.methodType(stateClass, domainEventClass, stateClass)); + } catch (NoSuchMethodException | IllegalAccessException e) { throw new RuntimeException(e); } } @Override + @SuppressWarnings("unchecked") public @NotNull S apply(@NotNull E event, S state) { try { - return (S) adapterMethod.invoke(aggregate, event, state); - } catch (IllegalAccessException e) { - throw new RuntimeException(e); - } catch (InvocationTargetException e) { - if (e.getCause() != null) { - if (e.getCause() instanceof RuntimeException) { - throw (RuntimeException) e.getCause(); - } else { - throw new RuntimeException(e.getCause()); - } - } else { - throw new RuntimeException(e); + return (S) methodHandle.invoke(aggregate, event, state); + } catch (WrongMethodTypeException | ClassCastException e) { + throw e; + } catch (Throwable e) { + if (e instanceof RuntimeException runtimeException) { + throw runtimeException; } + throw new RuntimeException(e); } } @@ -101,4 +102,4 @@ public Aggregate getAggregate() { public boolean isCreate() { return create; } -} +} \ No newline at end of file diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregatePartition.java b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregatePartition.java index aab953a1..5c788f89 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregatePartition.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregatePartition.java @@ -517,7 +517,7 @@ private void processRecords(ConsumerRecords allRecords) List> stateRecords = allRecords.records(statePartition); if (!stateRecords.isEmpty()) { stateRepository.process(stateRecords); - offsets.put(statePartition, stateRecords.get(stateRecords.size() - 1).offset()); + offsets.put(statePartition, stateRecords.getLast().offset()); } // then internal events (ignore?) allRecords.records(domainEventPartition) diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregateRuntimeFactory.java b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregateRuntimeFactory.java index c54a258c..3bac0186 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregateRuntimeFactory.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/AggregateRuntimeFactory.java @@ -20,6 +20,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import org.elasticsoftware.akces.aggregate.*; import org.elasticsoftware.akces.annotations.AggregateInfo; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; import org.elasticsoftware.akces.schemas.KafkaSchemaRegistry; import org.springframework.beans.BeansException; import org.springframework.beans.factory.FactoryBean; @@ -48,7 +49,7 @@ public void setApplicationContext(ApplicationContext applicationContext) throws } @Override - public AggregateRuntime getObject() throws Exception { + public AggregateRuntime getObject(){ return createRuntime(aggregate); } @@ -61,22 +62,31 @@ private KafkaAggregateRuntime createRuntime(Aggregate aggregate) { KafkaAggregateRuntime.Builder runtimeBuilder = new KafkaAggregateRuntime.Builder(); AggregateInfo aggregateInfo = aggregate.getClass().getAnnotation(AggregateInfo.class); - - if (aggregateInfo != null) { - runtimeBuilder.setStateType(new AggregateStateType<>( - aggregateInfo.value(), - aggregateInfo.version(), - aggregate.getStateClass(), - aggregateInfo.generateGDPRKeyOnCreate(), - aggregateInfo.indexed(), - aggregateInfo.indexName(), - hasPIIDataAnnotation(aggregate.getStateClass()) - )); - } else { + AggregateStateInfo aggregateStateInfo = aggregate.getStateClass().getAnnotation(AggregateStateInfo.class); + if (aggregateStateInfo == null) { + throw new IllegalStateException("Aggregate state class " + aggregate.getStateClass().getName() + + " must be annotated with @AggregateStateInfo"); + } + // ensure aggregateInfo is not null + if (aggregateInfo == null) { throw new IllegalStateException("Class implementing Aggregate must be annotated with @AggregateInfo"); } - runtimeBuilder - .setAggregateClass(aggregate.getClass()) + + // ensure the aggregate class references the correct state class version + if (aggregateStateInfo.version() != aggregateInfo.stateVersion()) { + throw new IllegalStateException("Aggregate state class version " + aggregateStateInfo.version() + + " does not match the aggregate stateVersion " + aggregateInfo.stateVersion()); + } + + runtimeBuilder.setStateType(new AggregateStateType<>( + aggregateInfo.value(), + aggregateStateInfo.version(), + aggregate.getStateClass(), + aggregateInfo.generateGDPRKeyOnCreate(), + aggregateInfo.indexed(), + aggregateInfo.indexName(), + hasPIIDataAnnotation(aggregate.getStateClass()))) + .setAggregateClass((Class>) aggregate.getClass()) .setObjectMapper(objectMapper) .setGenerateGDPRKeyOnCreate(aggregateInfo.generateGDPRKeyOnCreate()); @@ -147,6 +157,19 @@ private KafkaAggregateRuntime createRuntime(Aggregate aggregate) { .addEventBridgeHandler(type, adapter) .addDomainEvent(type); }); + // Add state upcasting handlers + applicationContext.getBeansOfType(UpcastingHandlerFunction.class).values().stream() + .filter(adapter -> adapter.getAggregate().equals(aggregate)) + .forEach(adapter -> { + if(adapter.getInputType() instanceof AggregateStateType stateType) { + runtimeBuilder + .addStateUpcastingHandler(stateType, adapter); + } else if(adapter.getInputType() instanceof DomainEventType eventType) { + runtimeBuilder + .addEventUpcastingHandler(eventType, adapter) + .addDomainEvent(eventType); + } + }); return runtimeBuilder.setSchemaRegistry(schemaRegistry).build(); } diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java index 4600472c..8a466b10 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/kafka/KafkaAggregateRuntime.java @@ -50,8 +50,8 @@ public class KafkaAggregateRuntime implements AggregateRuntime { private static final Logger log = LoggerFactory.getLogger(KafkaAggregateRuntime.class); - private final AggregateStateType type; - private final Class aggregateClass; + private final AggregateStateType stateType; + private final Class> aggregateClass; private final CommandHandlerFunction commandCreateHandler; private final EventHandlerFunction eventCreateHandler; private final EventSourcingHandlerFunction createStateHandler; @@ -61,6 +61,8 @@ public class KafkaAggregateRuntime implements AggregateRuntime { private final Map, EventHandlerFunction> eventHandlers; private final Map, EventSourcingHandlerFunction> eventSourcingHandlers; private final Map, EventBridgeHandlerFunction> eventBridgeHandlers; + private final Map, UpcastingHandlerFunction, AggregateStateType>> stateUpcastingHandlers; + private final Map, UpcastingHandlerFunction, DomainEventType>> eventUpcastingHandlers; private final boolean generateGDPRKeyOnCreate; private final boolean shouldHandlePIIData; private final KafkaSchemaRegistry schemaRegistry; @@ -71,7 +73,7 @@ public class KafkaAggregateRuntime implements AggregateRuntime { private KafkaAggregateRuntime(KafkaSchemaRegistry schemaRegistry, ObjectMapper objectMapper, AggregateStateType stateType, - Class aggregateClass, + Class> aggregateClass, CommandHandlerFunction commandCreateHandler, EventHandlerFunction eventCreateHandler, EventSourcingHandlerFunction createStateHandler, @@ -80,10 +82,10 @@ private KafkaAggregateRuntime(KafkaSchemaRegistry schemaRegistry, Map, CommandHandlerFunction> commandHandlers, Map, EventHandlerFunction> eventHandlers, Map, EventSourcingHandlerFunction> eventSourcingHandlers, - Map, EventBridgeHandlerFunction> eventBridgeHandlers, + Map, EventBridgeHandlerFunction> eventBridgeHandlers, Map, UpcastingHandlerFunction, AggregateStateType>> stateUpcastingHandlers, Map, UpcastingHandlerFunction, DomainEventType>> eventUpcastingHandlers, boolean generateGDPRKeyOnCreate, boolean shouldHandlePIIData) { - this.type = stateType; + this.stateType = stateType; this.aggregateClass = aggregateClass; this.commandCreateHandler = commandCreateHandler; this.eventCreateHandler = eventCreateHandler; @@ -94,6 +96,8 @@ private KafkaAggregateRuntime(KafkaSchemaRegistry schemaRegistry, this.eventHandlers = eventHandlers; this.eventSourcingHandlers = eventSourcingHandlers; this.eventBridgeHandlers = eventBridgeHandlers; + this.stateUpcastingHandlers = stateUpcastingHandlers; + this.eventUpcastingHandlers = eventUpcastingHandlers; this.generateGDPRKeyOnCreate = generateGDPRKeyOnCreate; this.shouldHandlePIIData = shouldHandlePIIData; this.schemaRegistry = schemaRegistry; @@ -102,11 +106,11 @@ private KafkaAggregateRuntime(KafkaSchemaRegistry schemaRegistry, @Override public String getName() { - return type.typeName(); + return stateType.typeName(); } @Override - public Class getAggregateClass() { + public Class> getAggregateClass() { return aggregateClass; } @@ -188,18 +192,17 @@ private void commandExecutionError(CommandRecord commandRecord, } private CommandType getCommandType(CommandRecord commandRecord) { - CommandType commandType = commandTypes.getOrDefault(commandRecord.name(), emptyList()).stream() + return commandTypes.getOrDefault(commandRecord.name(), emptyList()).stream() .filter(ct -> ct.version() == commandRecord.version()) .findFirst().orElseThrow(RuntimeException::new); // TODO: replace with specific exception - return commandType; } private void indexDomainEventIfRequired(DomainEventRecord domainEventRecord, AggregateState state, BiConsumer domainEventIndexer, boolean createIndex) { - if (type.indexed()) { - domainEventIndexer.accept(domainEventRecord, new IndexParams(type.indexName(), state.getIndexKey(), createIndex)); + if (stateType.indexed()) { + domainEventIndexer.accept(domainEventRecord, new IndexParams(stateType.indexName(), state.getIndexKey(), createIndex)); } } @@ -214,27 +217,30 @@ private void handleCreateCommand(CommandType commandType, // always treat the first event as a create event Iterator itr = domainEvents.iterator(); DomainEvent domainEvent = itr.next(); + // we may need to upcast the event + DomainEventType domainEventType = getDomainEventType(domainEvent.getClass()); // create the state - AggregateState state = createStateHandler.apply(domainEvent, null); + AggregateState state = (createStateHandler.getEventType().equals(domainEventType)) ? + createStateHandler.apply(domainEvent, null) : + createStateHandler.apply(upcast(domainEvent, domainEventType), null); // store the state, generation is 1 because it is the first record AggregateStateRecord stateRecord = new AggregateStateRecord( commandRecord.tenantId(), - type.typeName(), - type.version(), + stateType.typeName(), + stateType.version(), serialize(state), - getEncoding(type), + getEncoding(stateType), state.getAggregateId(), commandRecord.correlationId(), 1L); protocolRecordConsumer.accept(stateRecord); // store the domain event - DomainEventType type = getDomainEventType(domainEvent.getClass()); DomainEventRecord eventRecord = new DomainEventRecord( commandRecord.tenantId(), - type.typeName(), - type.version(), + domainEventType.typeName(), + domainEventType.version(), serialize(domainEvent), - getEncoding(type), + getEncoding(domainEventType), domainEvent.getAggregateId(), commandRecord.correlationId(), stateRecord.generation()); @@ -288,10 +294,10 @@ private void handleCreateEvent(DomainEventType eventType, // store the state, generation is 1 because it is the first record AggregateStateRecord stateRecord = new AggregateStateRecord( domainEventRecord.tenantId(), - type.typeName(), - type.version(), + stateType.typeName(), + stateType.version(), serialize(state), - getEncoding(type), + getEncoding(stateType), state.getAggregateId(), domainEventRecord.correlationId(), 1L); @@ -359,14 +365,19 @@ private AggregateStateRecord processDomainEvent(String correlationId, DomainEventType domainEventType = getDomainEventType(domainEvent.getClass()); // error events don't change the state if (!(domainEvent instanceof ErrorEvent)) { - AggregateState nextState = eventSourcingHandlers.get(domainEventType).apply(domainEvent, currentState); + // we may have to upcast the event + // TODO: we're upcasting twice here + AggregateState nextState = eventSourcingHandlers.containsKey(domainEventType) ? + eventSourcingHandlers.get(domainEventType).apply(domainEvent, currentState) : + eventSourcingHandlers.get(getDomainEventType(upcast(domainEvent, domainEventType).getClass())) + .apply(upcast(domainEvent, domainEventType), currentState); // store the state, increasing the generation by 1 AggregateStateRecord nextStateRecord = new AggregateStateRecord( currentStateRecord.tenantId(), // inherit tenantId from the state record - type.typeName(), - type.version(), + stateType.typeName(), + stateType.version(), serialize(nextState), - getEncoding(type), + getEncoding(stateType), currentStateRecord.aggregateId(), correlationId, currentStateRecord.generation() + 1L); @@ -407,11 +418,12 @@ public void handleExternalDomainEventRecord(DomainEventRecord eventRecord, BiConsumer domainEventIndexer, Supplier stateRecordSupplier, CommandBus commandBus) throws IOException { - // determine the type to use for the external event - DomainEventType domainEventType = getDomainEventType(eventRecord); + // determine the type to use for the external event (this can be different from the eventRecord type) + DomainEventType eventRecordDomainEventType = getDomainEventType(eventRecord); // with external domainevents we should look at the handler and not at the type of the external event - if (domainEventType != null) { - if (eventCreateHandler != null && eventCreateHandler.getEventType().equals(domainEventType)) { + if (eventRecordDomainEventType != null) { + DomainEventType actualDomainEventType = upcastedType(eventRecordDomainEventType); + if (eventCreateHandler != null && eventCreateHandler.getEventType().equals(upcastedType(actualDomainEventType))) { // if the state already exists, this is an error. if (stateRecordSupplier.get() != null) { // this is an error, log it and generate a AggregateAlreadyExistsError @@ -421,14 +433,14 @@ public void handleExternalDomainEventRecord(DomainEventRecord eventRecord, eventRecord.aggregateId()); aggregateAlreadyExists(eventRecord, protocolRecordConsumer); } else { - handleCreateEvent(domainEventType, eventRecord, protocolRecordConsumer, domainEventIndexer); + handleCreateEvent(eventRecordDomainEventType, eventRecord, protocolRecordConsumer, domainEventIndexer); } } else { // only process the event if we have a handler for it - if (eventHandlers.containsKey(domainEventType)) { - handleEvent(domainEventType, eventRecord, protocolRecordConsumer, domainEventIndexer, stateRecordSupplier); - } else if(eventBridgeHandlers.containsKey(domainEventType)) { // it can also be an event bridge - handleBridgedEvent(domainEventType, eventRecord, commandBus); + if (eventHandlers.containsKey(actualDomainEventType)) { + handleEvent(eventRecordDomainEventType, eventRecord, protocolRecordConsumer, domainEventIndexer, stateRecordSupplier); + } else if(eventBridgeHandlers.containsKey(actualDomainEventType)) { // it can also be an event bridge + handleBridgedEvent(eventRecordDomainEventType, eventRecord, commandBus); } } } // ignore if we don't have an external domainevent registered @@ -505,13 +517,13 @@ public boolean requiresGDPRContext(DomainEventRecord eventRecord) { if(domainEventType == null) { return false; } else { - return domainEventType.piiData() || (!eventBridgeHandlers.containsKey(domainEventType) && type.piiData()); + return domainEventType.piiData() || (!eventBridgeHandlers.containsKey(domainEventType) && stateType.piiData()); } } @Override public boolean requiresGDPRContext(CommandRecord commandRecord) { - return this.type.piiData() || getCommandType(commandRecord).piiData(); + return this.stateType.piiData() || getCommandType(commandRecord).piiData(); } @Override @@ -542,11 +554,49 @@ public Command materialize(CommandType type, CommandRecord commandRecord) thr } private DomainEvent materialize(DomainEventType domainEventType, DomainEventRecord eventRecord) throws IOException { - return objectMapper.readValue(eventRecord.payload(), domainEventType.typeClass()); + DomainEvent domainEvent = objectMapper.readValue(eventRecord.payload(), domainEventType.typeClass()); + // check if we need to upcast the event + if(eventUpcastingHandlers.containsKey(domainEventType)) { + return upcast(domainEvent, domainEventType); + } else { + return domainEvent; + } + } + + private DomainEvent upcast(DomainEvent domainEvent, DomainEventType domainEventType) { + final UpcastingHandlerFunction, DomainEventType> upcastingHandlerFunction = eventUpcastingHandlers.get(domainEventType); + DomainEvent upcastedEvent = upcastingHandlerFunction.apply(domainEvent); + // see if we need to upcast the event + if(eventUpcastingHandlers.containsKey(upcastingHandlerFunction.getOutputType())) { + return upcast(upcastedEvent, upcastingHandlerFunction.getOutputType()); + } else { + return upcastedEvent; + } + } + + private DomainEventType upcastedType(DomainEventType type) { + return eventUpcastingHandlers.containsKey(type) ? upcastedType(eventUpcastingHandlers.get(type).getOutputType()) : type; } private AggregateState materialize(AggregateStateRecord stateRecord) throws IOException { - return objectMapper.readValue(stateRecord.payload(), getAggregateStateType(stateRecord).typeClass()); + AggregateStateType stateType = getAggregateStateType(stateRecord); + AggregateState state = objectMapper.readValue(stateRecord.payload(), getAggregateStateType(stateRecord).typeClass()); + // see if we need to upcast the state + if(!stateType.equals(this.stateType)) { + return upcast(state, stateType); + } else { + return state; + } + } + + private AggregateState upcast(AggregateState state, AggregateStateType inputStateType) { + final UpcastingHandlerFunction, AggregateStateType> upcastingHandlerFunction = stateUpcastingHandlers.get(inputStateType); + AggregateState upcastedState = upcastingHandlerFunction.apply(state); + if(!upcastingHandlerFunction.getOutputType().equals(stateType)) { + return upcast(upcastedState, upcastingHandlerFunction.getOutputType()); + } else { + return upcastedState; + } } private byte[] serialize(AggregateState state) throws IOException { @@ -592,8 +642,17 @@ private PayloadEncoding getEncoding(AggregateStateType type) { } private AggregateStateType getAggregateStateType(AggregateStateRecord record) { - // TODO: add support for more state versions - return type; + // there is only one active state type, however there can be multiple versions. if so, there must be an upcaster + // for the older versions to the current version + // TODO: this should be checked at initialization time and the runtime should fail to start! + if(stateType.typeName().equals(record.name()) && stateType.version() == record.version()) { + return stateType; + } else { + // it's an older version. we need to get the upcaster for this version + return stateUpcastingHandlers.keySet().stream() + .filter(aggregateStateType -> aggregateStateType.version() == record.version()) + .findAny().orElseThrow(() -> new IllegalStateException("Aggregate state type for " + record.name() + " with version " + record.version() +" does not exist")); + } } private void addCommand(CommandType commandType) { @@ -604,16 +663,18 @@ public static class Builder { private KafkaSchemaRegistry schemaRegistry; private ObjectMapper objectMapper; private AggregateStateType stateType; - private Class aggregateClass; + private Class> aggregateClass; private CommandHandlerFunction commandCreateHandler; private EventHandlerFunction eventCreateHandler; private EventSourcingHandlerFunction createStateHandler; - private Map, DomainEventType> domainEvents = new HashMap<>(); - private Map>> commandTypes = new HashMap<>(); - private Map, CommandHandlerFunction> commandHandlers = new HashMap<>(); - private Map, EventHandlerFunction> eventHandlers = new HashMap<>(); - private Map, EventSourcingHandlerFunction> eventSourcingHandlers = new HashMap<>(); - private Map, EventBridgeHandlerFunction> eventBridgeHandlers = new HashMap<>(); + private final Map, DomainEventType> domainEvents = new HashMap<>(); + private final Map>> commandTypes = new HashMap<>(); + private final Map, CommandHandlerFunction> commandHandlers = new HashMap<>(); + private final Map, EventHandlerFunction> eventHandlers = new HashMap<>(); + private final Map, EventSourcingHandlerFunction> eventSourcingHandlers = new HashMap<>(); + private final Map, EventBridgeHandlerFunction> eventBridgeHandlers = new HashMap<>(); + private final Map, UpcastingHandlerFunction, AggregateStateType>> stateUpcastingHandlers = new HashMap<>(); + private final Map, UpcastingHandlerFunction, DomainEventType>> eventUpcastingHandlers = new HashMap<>(); private boolean generateGDPRKeyOnCreate = false; public Builder setSchemaRegistry(KafkaSchemaRegistry schemaRegistry) { @@ -631,7 +692,7 @@ public Builder setStateType(AggregateStateType stateType) { return this; } - public Builder setAggregateClass(Class aggregateClass) { + public Builder setAggregateClass(Class> aggregateClass) { this.aggregateClass = aggregateClass; return this; } @@ -681,6 +742,18 @@ public Builder addEventBridgeHandler(DomainEventType eventType, EventBridgeHa return this; } + public Builder addStateUpcastingHandler(AggregateStateType inputType, + UpcastingHandlerFunction, AggregateStateType> upcastingHandler) { + this.stateUpcastingHandlers.put(inputType, upcastingHandler); + return this; + } + + public Builder addEventUpcastingHandler(DomainEventType inputType, + UpcastingHandlerFunction, DomainEventType> upcastingHandler) { + this.eventUpcastingHandlers.put(inputType, upcastingHandler); + return this; + } + public Builder setGenerateGDPRKeyOnCreate(boolean generateGDPRKeyOnCreate) { this.generateGDPRKeyOnCreate = generateGDPRKeyOnCreate; @@ -708,6 +781,8 @@ public KafkaAggregateRuntime build() { eventHandlers, eventSourcingHandlers, eventBridgeHandlers, + stateUpcastingHandlers, + eventUpcastingHandlers, generateGDPRKeyOnCreate, shouldHandlePIIData); } diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/state/InMemoryAggregateStateRepository.java b/main/runtime/src/main/java/org/elasticsoftware/akces/state/InMemoryAggregateStateRepository.java index 565cc49c..969ff8fa 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/state/InMemoryAggregateStateRepository.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/state/InMemoryAggregateStateRepository.java @@ -45,7 +45,7 @@ public void close() { @Override public void prepare(AggregateStateRecord record, Future recordMetadataFuture) { - transactionStateRecordMap.put(record.aggregateId(), new RecordAndMetadata(record, recordMetadataFuture)); + transactionStateRecordMap.put(record.aggregateId(), new RecordAndMetadata<>(record, recordMetadataFuture)); } @Override diff --git a/main/runtime/src/main/java/org/elasticsoftware/akces/state/RocksDBAggregateStateRepository.java b/main/runtime/src/main/java/org/elasticsoftware/akces/state/RocksDBAggregateStateRepository.java index 2e272b50..4d0dc089 100644 --- a/main/runtime/src/main/java/org/elasticsoftware/akces/state/RocksDBAggregateStateRepository.java +++ b/main/runtime/src/main/java/org/elasticsoftware/akces/state/RocksDBAggregateStateRepository.java @@ -118,7 +118,7 @@ public long getOffset() { @Override public void prepare(AggregateStateRecord record, Future recordMetadataFuture) { checkAggregateIdType(record.aggregateId()); - transactionStateRecordMap.put(record.aggregateId(), new RecordAndMetadata(record, recordMetadataFuture)); + transactionStateRecordMap.put(record.aggregateId(), new RecordAndMetadata<>(record, recordMetadataFuture)); } @Override @@ -127,7 +127,7 @@ public void commit() { // start writing the transactions (no need to resolve the futures just yet) Transaction transaction = db.beginTransaction(new WriteOptions()); try { - for (RecordAndMetadata recordAndMetadata : transactionStateRecordMap.values()) { + for (RecordAndMetadata recordAndMetadata : transactionStateRecordMap.values()) { transaction.put(keyBytes(recordAndMetadata.record().aggregateId()), serializer.serialize(topicName, recordAndMetadata.record())); } // now we need to find the highest offset in this batch diff --git a/main/runtime/src/test/java/org/elasticsoftware/akces/beans/MinInsyncReplicasTest.java b/main/runtime/src/test/java/org/elasticsoftware/akces/beans/MinInsyncReplicasTest.java index 9bbe794b..e4a91892 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akces/beans/MinInsyncReplicasTest.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akces/beans/MinInsyncReplicasTest.java @@ -17,7 +17,6 @@ package org.elasticsoftware.akces.beans; -import org.elasticsoftware.akces.util.KafkaUtils; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountConfiguration.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountConfiguration.java new file mode 100644 index 00000000..2ad82102 --- /dev/null +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountConfiguration.java @@ -0,0 +1,63 @@ +/* + * Copyright 2022 - 2025 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.akcestest; + +import com.fasterxml.jackson.databind.ObjectMapper; +import io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import org.elasticsoftware.akces.beans.AggregateBeanFactoryPostProcessor; +import org.elasticsoftware.akces.gdpr.jackson.AkcesGDPRModule; +import org.elasticsoftware.akces.schemas.KafkaSchemaRegistry; +import org.elasticsoftware.akces.serialization.BigDecimalSerializer; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.autoconfigure.jackson.Jackson2ObjectMapperBuilderCustomizer; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; + +import java.math.BigDecimal; + +@EnableAutoConfiguration +@ComponentScan(basePackages = { + "org.elasticsoftware.akcestest.aggregate.account", +}) +public class AccountConfiguration { + @Bean(name = "aggregateServiceBeanFactoryPostProcessor") + public AggregateBeanFactoryPostProcessor aggregateBeanFactoryPostProcessor() { + return new AggregateBeanFactoryPostProcessor(); + } + + @Bean(name = "aggregateServiceJsonCustomizer") + public Jackson2ObjectMapperBuilderCustomizer jsonCustomizer() { + return builder -> { + builder.modulesToInstall(new AkcesGDPRModule()); + builder.serializerByType(BigDecimal.class, new BigDecimalSerializer()); + }; + } + + @Bean(name = "aggregateServiceSchemaRegistryClient") + public SchemaRegistryClient createSchemaRegistryClient() { + return new MockSchemaRegistryClient(); + } + + @Bean(name = "aggregateServiceSchemaRegistry") + public KafkaSchemaRegistry createSchemaRegistry(@Qualifier("aggregateServiceSchemaRegistryClient") SchemaRegistryClient src, + ObjectMapper objectMapper) { + return new KafkaSchemaRegistry(src, objectMapper); + } +} diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountTests.java new file mode 100644 index 00000000..13317966 --- /dev/null +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AccountTests.java @@ -0,0 +1,52 @@ +/* + * Copyright 2022 - 2025 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.akcestest; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import jakarta.inject.Inject; +import org.elasticsoftware.akces.schemas.KafkaSchemaRegistry; +import org.elasticsoftware.akcestest.aggregate.account.AccountState; +import org.elasticsoftware.akcestest.aggregate.account.PreviousAccountState; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.ApplicationContext; + +@SpringBootTest(classes = AccountConfiguration.class, properties = "spring.autoconfigure.exclude=org.elasticsoftware.akces.client.AkcesClientAutoConfiguration") +public class AccountTests { + @Inject + ApplicationContext applicationContext; + @Inject + ObjectMapper objectMapper; + @Inject + KafkaSchemaRegistry schemaRegistry; + @Inject + SchemaRegistryClient schemaRegistryClient; + + @Test + public void testAggregateStateSerializationWithChangingSchema() throws JsonProcessingException { + String userId = "a2b04808-19d7-4ec8-94db-1ce28ca517f1"; + String serializedState= objectMapper.writeValueAsString(new PreviousAccountState(userId, "US", "John", "Doe", "john.doe@example.com")); + AccountState accountState = objectMapper.readValue(serializedState, AccountState.class); + Assertions.assertEquals(userId, accountState.userId()); + Assertions.assertNotNull(accountState.twoFactorEnabled()); + Assertions.assertFalse(accountState.twoFactorEnabled()); + } +} diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/AggregateServiceApplicationTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AggregateServiceApplicationTests.java index eb214b92..c31876e3 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/AggregateServiceApplicationTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/AggregateServiceApplicationTests.java @@ -87,6 +87,7 @@ public class AggregateServiceApplicationTests { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java index 11fd3e63..28ef5ea5 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/RuntimeTests.java @@ -140,6 +140,7 @@ public class RuntimeTests { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); @@ -647,13 +648,13 @@ public void onPartitionsAssigned(Collection partitions) { assertEquals(2, allRecords.size()); // now see if the domain event and state are encrypted - assertTrue(allRecords.get(1) instanceof DomainEventRecord); + assertInstanceOf(DomainEventRecord.class, allRecords.get(1)); DomainEventRecord domainEventRecord = (DomainEventRecord) allRecords.get(1); AccountCreatedEvent accountCreatedEvent = objectMapper.readValue(domainEventRecord.payload(), AccountCreatedEvent.class); assertNotEquals("Fahim", accountCreatedEvent.firstName()); assertNotEquals("Zuijderwijk", accountCreatedEvent.lastName()); assertNotEquals("FahimZuijderwijk@jourrapide.com", accountCreatedEvent.email()); - AggregateStateRecord stateRecord = (AggregateStateRecord) allRecords.get(0); + AggregateStateRecord stateRecord = (AggregateStateRecord) allRecords.getFirst(); AccountState accountState = objectMapper.readValue(stateRecord.payload(), AccountState.class); assertNotEquals("Fahim", accountState.firstName()); assertNotEquals("Zuijderwijk", accountState.lastName()); diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/WalletTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/WalletTests.java index 33fd4a3b..0880e1e0 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/WalletTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/WalletTests.java @@ -57,6 +57,7 @@ public void testFindBeans() { assertEquals(4, applicationContext.getBeansOfType(CommandHandlerFunction.class).size()); assertEquals(1, applicationContext.getBeansOfType(EventHandlerFunction.class).size()); assertEquals(4, applicationContext.getBeansOfType(EventSourcingHandlerFunction.class).size()); + assertEquals(1, applicationContext.getBeansOfType(UpcastingHandlerFunction.class).size()); Assertions.assertNotNull(applicationContext.getBean("Wallet_ch_create_CreateWallet_1")); Assertions.assertNotNull(applicationContext.getBean("Wallet_ch_credit_CreditWallet_1")); Assertions.assertNotNull(applicationContext.getBean("Wallet_ch_makeReservation_ReserveAmount_1")); @@ -65,6 +66,7 @@ public void testFindBeans() { Assertions.assertNotNull(applicationContext.getBean("Wallet_esh_create_WalletCreated_1")); Assertions.assertNotNull(applicationContext.getBean("Wallet_esh_createBalance_BalanceCreated_1")); Assertions.assertNotNull(applicationContext.getBean("Wallet_esh_credit_WalletCredited_1")); + Assertions.assertNotNull(applicationContext.getBean("Wallet_suh_upcast_Wallet_1_to_2")); } @Test @@ -221,12 +223,12 @@ public void testCreateWalletByCommand() throws Exception { () -> null ); assertEquals(4, producedRecords.size()); - AggregateStateRecord actualRecord = (AggregateStateRecord) producedRecords.get(0); + AggregateStateRecord actualRecord = (AggregateStateRecord) producedRecords.getFirst(); AggregateStateRecord expectedRecord = new AggregateStateRecord( tenantId, "Wallet", - 1, - objectMapper.writeValueAsBytes(new WalletState(aggregateId, new ArrayList<>())), + 2, + objectMapper.writeValueAsBytes(new WalletStateV2(aggregateId, new ArrayList<>())), PayloadEncoding.JSON, aggregateId, correlationId, @@ -254,8 +256,8 @@ public void testCreateWalletByCommand() throws Exception { expectedRecord = new AggregateStateRecord( tenantId, "Wallet", - 1, - objectMapper.writeValueAsBytes(new WalletState(aggregateId, List.of(new WalletState.Balance("EUR", BigDecimal.ZERO)))), + 2, + objectMapper.writeValueAsBytes(new WalletStateV2(aggregateId, List.of(new WalletStateV2.Balance("EUR", BigDecimal.ZERO)))), PayloadEncoding.JSON, aggregateId, correlationId, @@ -313,7 +315,7 @@ public void testIndexWalletEventsFromCommand() throws Exception { ); // we should index 2 events: WalletCreated and BalanceCreated assertEquals(2, indexedEvents.size()); - DomainEventRecord actual = indexedEvents.get(0); + DomainEventRecord actual = indexedEvents.getFirst(); assertEquals(1, actual.generation()); assertEquals(aggregateId, actual.aggregateId()); @@ -323,7 +325,7 @@ public void testIndexWalletEventsFromCommand() throws Exception { assertEquals("WalletCreated", actual.name()); assertEquals(1, actual.version()); - actual = (DomainEventRecord) indexedEvents.get(1); + actual = indexedEvents.get(1); assertEquals(2, actual.generation()); assertEquals(aggregateId, actual.aggregateId()); @@ -367,12 +369,12 @@ public void testCreateWalletByExternalDomainEvent() throws Exception { null ); assertEquals(4, producedRecords.size()); - AggregateStateRecord actualRecord = (AggregateStateRecord) producedRecords.get(0); + AggregateStateRecord actualRecord = (AggregateStateRecord) producedRecords.getFirst(); AggregateStateRecord expectedRecord = new AggregateStateRecord( tenantId, "Wallet", - 1, - objectMapper.writeValueAsBytes(new WalletState(aggregateId, new ArrayList<>())), + 2, + objectMapper.writeValueAsBytes(new WalletStateV2(aggregateId, new ArrayList<>())), PayloadEncoding.JSON, aggregateId, correlationId, @@ -400,8 +402,8 @@ public void testCreateWalletByExternalDomainEvent() throws Exception { expectedRecord = new AggregateStateRecord( tenantId, "Wallet", - 1, - objectMapper.writeValueAsBytes(new WalletState(aggregateId, List.of(new WalletState.Balance("EUR", BigDecimal.ZERO)))), + 2, + objectMapper.writeValueAsBytes(new WalletStateV2(aggregateId, List.of(new WalletStateV2.Balance("EUR", BigDecimal.ZERO)))), PayloadEncoding.JSON, aggregateId, correlationId, @@ -461,4 +463,76 @@ public void testIndexWalletEventsByExternalDomainEvent() throws Exception { // we should index 2 events: WalletCreated and BalanceCreated assertEquals(2, indexedEvents.size()); } + + @Test + public void testWalletCreatedWithWalletStateV1andUpdatedWithWalletStateV2() throws Exception { + AggregateRuntime walletAggregate = applicationContext.getBean("WalletAggregateRuntimeFactory", AggregateRuntime.class); + String tenantId = "tenant1"; + String aggregateId = "d43a3afc-3e5a-11ed-b878-0242ac120002"; + String correlationId = "01e04622-3e5b-11ed-b878-0242ac120002"; + List producedRecords = new ArrayList<>(); + List indexedEvents = new ArrayList<>(); + + schemaRegistryClient.register("domainevents.AccountCreated", + schemaRegistry.generateJsonSchema(new DomainEventType<>("AccountCreated", 1, ExternalAccountCreatedEvent.class, true, true, false, false)), + 1, + -1); + schemaRegistryClient.register("domainevents.BalanceCreated", + schemaRegistry.generateJsonSchema(new DomainEventType<>("BalanceCreated", 1, BalanceCreatedEvent.class, false, false, false, false)), + 1, + -1); + for (DomainEventType domainEventType : walletAggregate.getAllDomainEventTypes()) { + walletAggregate.registerAndValidate(domainEventType); + } + + AggregateStateRecord v1StateRecord = new AggregateStateRecord( + tenantId, + "Wallet", + 1, + objectMapper.writeValueAsBytes(new WalletState(aggregateId, List.of(new WalletState.Balance("EUR", BigDecimal.ZERO)))), + PayloadEncoding.JSON, + aggregateId, + correlationId, + 2); + + walletAggregate.handleCommandRecord( + new CommandRecord( + tenantId, + "CreateBalance", + 1, + objectMapper.writeValueAsBytes( + new CreateBalanceCommand(aggregateId, "ETH")), + PayloadEncoding.JSON, + aggregateId, + correlationId, + null), + producedRecords::add, + (eventRecord, index) -> indexedEvents.add(eventRecord), + () -> v1StateRecord + ); + + assertEquals(2, producedRecords.size()); + + AggregateStateRecord expectedStateRecord = new AggregateStateRecord( + tenantId, + "Wallet", + 2, + objectMapper.writeValueAsBytes(new WalletStateV2(aggregateId, List.of( + new WalletStateV2.Balance("EUR", BigDecimal.ZERO), + new WalletStateV2.Balance("ETH", BigDecimal.ZERO)))), + PayloadEncoding.JSON, + aggregateId, + correlationId, + 3); + + AggregateStateRecord actualStateRecord = (AggregateStateRecord) producedRecords.getFirst(); + assertEquals(expectedStateRecord.generation(), actualStateRecord.generation()); + assertEquals(expectedStateRecord.aggregateId(), actualStateRecord.aggregateId()); + assertEquals(expectedStateRecord.correlationId(), actualStateRecord.correlationId()); + assertEquals(expectedStateRecord.encoding(), actualStateRecord.encoding()); + assertEquals(expectedStateRecord.version(), actualStateRecord.version()); + assertEquals(expectedStateRecord.tenantId(), actualStateRecord.tenantId()); + assertEquals(expectedStateRecord.name(), actualStateRecord.name()); + assertArrayEquals(expectedStateRecord.payload(), actualStateRecord.payload()); + } } diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/Account.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/Account.java index 50012841..32d34045 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/Account.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/Account.java @@ -22,6 +22,7 @@ import org.elasticsoftware.akces.annotations.AggregateInfo; import org.elasticsoftware.akces.annotations.CommandHandler; import org.elasticsoftware.akces.annotations.EventSourcingHandler; +import org.elasticsoftware.akces.annotations.UpcastingHandler; import java.util.stream.Stream; @@ -43,9 +44,14 @@ public Stream create(CreateAccountCommand cmd, AccountState return Stream.of(new AccountCreatedEvent(cmd.userId(), cmd.country(), cmd.firstName(), cmd.lastName(), cmd.email())); } + @UpcastingHandler + public AccountCreatedEventV2 cast(AccountCreatedEvent event) { + return new AccountCreatedEventV2(event.userId(), event.country(), event.firstName(), event.lastName(), event.email(), false); + } + @EventSourcingHandler(create = true) @NotNull - public AccountState create(@NotNull AccountCreatedEvent event, AccountState isNull) { + public AccountState create(@NotNull AccountCreatedEventV2 event, AccountState isNull) { return new AccountState(event.userId(), event.country(), event.firstName(), event.lastName(), event.email()); } } diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountCreatedEventV2.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountCreatedEventV2.java new file mode 100644 index 00000000..9012c3d4 --- /dev/null +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountCreatedEventV2.java @@ -0,0 +1,40 @@ +/* + * Copyright 2022 - 2025 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.akcestest.aggregate.account; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.annotations.AggregateIdentifier; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.annotations.PIIData; +import org.elasticsoftware.akces.events.DomainEvent; + + +@DomainEventInfo(type = "AccountCreated", version = 2) +public record AccountCreatedEventV2( + @AggregateIdentifier @NotNull String userId, + @NotNull String country, + @NotNull @PIIData String firstName, + @NotNull @PIIData String lastName, + @NotNull @PIIData String email, + @NotNull boolean twoFactorEnabled +) implements DomainEvent { + @Override + public String getAggregateId() { + return userId(); + } +} diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountState.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountState.java index b5dd2f70..e6a7507c 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountState.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/AccountState.java @@ -27,9 +27,26 @@ public record AccountState(@NotNull String userId, @NotNull String country, @NotNull @PIIData String firstName, @NotNull @PIIData String lastName, - @NotNull @PIIData String email) implements AggregateState { + @NotNull @PIIData String email, + Boolean twoFactorEnabled) implements AggregateState { + // Compact constructor to handle possible null values from deserialization + public AccountState { + if (twoFactorEnabled == null) { + twoFactorEnabled = false; + } + } + + // Default constructor with false for twoFactorEnabled + public AccountState(@NotNull String userId, + @NotNull String country, + @NotNull String firstName, + @NotNull String lastName, + @NotNull String email) { + this(userId, country, firstName, lastName, email, false); + } + @Override public String getAggregateId() { return userId(); } -} +} \ No newline at end of file diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/PreviousAccountState.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/PreviousAccountState.java new file mode 100644 index 00000000..149dd1ed --- /dev/null +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/account/PreviousAccountState.java @@ -0,0 +1,27 @@ +/* + * Copyright 2022 - 2025 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.akcestest.aggregate.account; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.annotations.PIIData; + +public record PreviousAccountState(@NotNull String userId, + @NotNull String country, + @NotNull @PIIData String firstName, + @NotNull @PIIData String lastName, + @NotNull @PIIData String email) {} diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/orders/OrderProcessManagerState.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/orders/OrderProcessManagerState.java index 9d391f94..34c64c72 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/orders/OrderProcessManagerState.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/orders/OrderProcessManagerState.java @@ -19,11 +19,13 @@ import jakarta.validation.constraints.NotNull; import org.elasticsoftware.akces.annotations.AggregateIdentifier; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; import org.elasticsoftware.akces.processmanager.ProcessManagerState; import org.elasticsoftware.akces.processmanager.UnknownAkcesProcessException; import java.util.List; +@AggregateStateInfo(type = "OrderProcessManager", version = 1) public record OrderProcessManagerState( @NotNull @AggregateIdentifier String userId, List runningProcesses diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/Wallet.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/Wallet.java index fb85e242..955784da 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/Wallet.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/Wallet.java @@ -19,10 +19,7 @@ import jakarta.validation.constraints.NotNull; import org.elasticsoftware.akces.aggregate.Aggregate; -import org.elasticsoftware.akces.annotations.AggregateInfo; -import org.elasticsoftware.akces.annotations.CommandHandler; -import org.elasticsoftware.akces.annotations.EventHandler; -import org.elasticsoftware.akces.annotations.EventSourcingHandler; +import org.elasticsoftware.akces.annotations.*; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.akcestest.aggregate.account.AccountCreatedEvent; @@ -32,32 +29,41 @@ import java.util.stream.Stream; -@AggregateInfo(value = "Wallet", version = 1, indexed = true, indexName = "Users") +@AggregateInfo(value = "Wallet", stateVersion = 2, indexed = true, indexName = "Users") @SuppressWarnings("unused") -public final class Wallet implements Aggregate { +public final class Wallet implements Aggregate { @Override public String getName() { return "Wallet"; } @Override - public Class getStateClass() { - return WalletState.class; + public Class getStateClass() { + return WalletStateV2.class; + } + + @UpcastingHandler + public WalletStateV2 upcast(WalletState state) { + // if there is any reservedAmount, we need to create a reservation for it + return new WalletStateV2(state.id(), state.balances().stream().map( + balance -> balance.reservedAmount().compareTo(BigDecimal.ZERO) > 0 ? + new WalletStateV2.Balance(balance.currency(), balance.amount(), List.of(new WalletStateV2.Reservation("v1-reservedAmount", balance.reservedAmount()))) : + new WalletStateV2.Balance(balance.currency(), balance.amount())).toList()); } @CommandHandler(create = true, produces = WalletCreatedEvent.class, errors = {}) - public @NotNull Stream create(@NotNull CreateWalletCommand cmd, WalletState isNull) { + public @NotNull Stream create(@NotNull CreateWalletCommand cmd, WalletStateV2 isNull) { return Stream.of(new WalletCreatedEvent(cmd.id()), new BalanceCreatedEvent(cmd.id(), cmd.currency())); } @EventHandler(create = true, produces = WalletCreatedEvent.class, errors = {}) - public @NotNull Stream create(@NotNull AccountCreatedEvent event, WalletState isNull) { + public @NotNull Stream create(@NotNull AccountCreatedEvent event, WalletStateV2 isNull) { // TODO: base the currency on the country return Stream.of(new WalletCreatedEvent(event.getAggregateId()), new BalanceCreatedEvent(event.getAggregateId(), "EUR")); } @CommandHandler(produces = BalanceCreatedEvent.class, errors = {BalanceAlreadyExistsErrorEvent.class}) - public @NotNull Stream createBalance(@NotNull CreateBalanceCommand cmd, @NotNull WalletState currentState) { + public @NotNull Stream createBalance(@NotNull CreateBalanceCommand cmd, @NotNull WalletStateV2 currentState) { boolean balanceExists = currentState.balances().stream() .anyMatch(balance -> balance.currency().equals(cmd.currency())); if (balanceExists) { @@ -68,8 +74,8 @@ public Class getStateClass() { @CommandHandler(produces = WalletCreditedEvent.class, errors = {InvalidCurrencyErrorEvent.class, InvalidAmountErrorEvent.class}) @NotNull - public Stream credit(@NotNull CreditWalletCommand cmd, @NotNull WalletState currentState) { - WalletState.Balance balance = currentState.balances().stream().filter(b -> b.currency().equals(cmd.currency())).findFirst().orElse(null); + public Stream credit(@NotNull CreditWalletCommand cmd, @NotNull WalletStateV2 currentState) { + WalletStateV2.Balance balance = currentState.balances().stream().filter(b -> b.currency().equals(cmd.currency())).findFirst().orElse(null); if (balance == null) { // TODO: add more detail to the error event return Stream.of(new InvalidCurrencyErrorEvent(cmd.id(), cmd.currency())); @@ -82,8 +88,8 @@ public Stream credit(@NotNull CreditWalletCommand cmd, @NotNull Wal } @CommandHandler(produces = AmountReservedEvent.class, errors = {InvalidCurrencyErrorEvent.class, InvalidAmountErrorEvent.class, InsufficientFundsErrorEvent.class}) - public Stream makeReservation(ReserveAmountCommand command, WalletState state) { - WalletState.Balance balance = state.balances().stream().filter(b -> b.currency().equals(command.currency())).findFirst().orElse(null); + public Stream makeReservation(ReserveAmountCommand command, WalletStateV2 state) { + WalletStateV2.Balance balance = state.balances().stream().filter(b -> b.currency().equals(command.currency())).findFirst().orElse(null); if (balance == null) { // TODO: add more detail to the error event return Stream.of(new InvalidCurrencyErrorEvent(command.userId(), command.currency(), command.referenceId())); @@ -101,22 +107,22 @@ public Stream makeReservation(ReserveAmountCommand command, WalletS } @EventSourcingHandler(create = true) - public @NotNull WalletState create(@NotNull WalletCreatedEvent event, WalletState isNull) { - return new WalletState(event.id(), new ArrayList<>()); + public @NotNull WalletStateV2 create(@NotNull WalletCreatedEvent event, WalletStateV2 isNull) { + return new WalletStateV2(event.id(), new ArrayList<>()); } @EventSourcingHandler - public @NotNull WalletState createBalance(@NotNull BalanceCreatedEvent event, WalletState state) { - List balances = new ArrayList<>(state.balances()); - balances.add(new WalletState.Balance(event.currency(), BigDecimal.ZERO)); - return new WalletState(state.id(), balances); + public @NotNull WalletStateV2 createBalance(@NotNull BalanceCreatedEvent event, WalletStateV2 state) { + List balances = new ArrayList<>(state.balances()); + balances.add(new WalletStateV2.Balance(event.currency(), BigDecimal.ZERO)); + return new WalletStateV2(state.id(), balances); } @EventSourcingHandler - public @NotNull WalletState credit(@NotNull WalletCreditedEvent event, @NotNull WalletState state) { - return new WalletState(state.id(), state.balances().stream().map(b -> { + public @NotNull WalletStateV2 credit(@NotNull WalletCreditedEvent event, @NotNull WalletStateV2 state) { + return new WalletStateV2(state.id(), state.balances().stream().map(b -> { if (b.currency().equals(event.currency())) { - return new WalletState.Balance(b.currency(), b.amount().add(event.amount())); + return new WalletStateV2.Balance(b.currency(), b.amount().add(event.amount())); } else { return b; } @@ -124,10 +130,12 @@ public Stream makeReservation(ReserveAmountCommand command, WalletS } @EventSourcingHandler - public @NotNull WalletState reserveAmount(@NotNull AmountReservedEvent event, @NotNull WalletState state) { - return new WalletState(state.id(), state.balances().stream().map(b -> { + public @NotNull WalletStateV2 reserveAmount(@NotNull AmountReservedEvent event, @NotNull WalletStateV2 state) { + return new WalletStateV2(state.id(), state.balances().stream().map(b -> { if (b.currency().equals(event.currency())) { - return new WalletState.Balance(b.currency(), b.amount(), b.reservedAmount().add(event.amount())); + List updatedReservations = new ArrayList<>(b.reservations()); + updatedReservations.add(new WalletStateV2.Reservation(event.referenceId(), event.amount())); + return new WalletStateV2.Balance(b.currency(), b.amount(), updatedReservations); } else { return b; } diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/WalletStateV2.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/WalletStateV2.java new file mode 100644 index 00000000..0d6e18b3 --- /dev/null +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/aggregate/wallet/WalletStateV2.java @@ -0,0 +1,65 @@ +/* + * Copyright 2022 - 2025 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.akcestest.aggregate.wallet; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.aggregate.AggregateState; +import org.elasticsoftware.akces.annotations.AggregateIdentifier; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; + +import java.math.BigDecimal; +import java.util.List; + +@AggregateStateInfo(type = "Wallet", version = 2) +public record WalletStateV2( + @AggregateIdentifier @NotNull String id, + List balances +) implements AggregateState { + @Override + public String getAggregateId() { + return id(); + } + + public record Balance(@NotNull String currency,@NotNull BigDecimal amount,@NotNull List reservations) { + public Balance(@NotNull String currency) { + this(currency, BigDecimal.ZERO, List.of()); + } + + public Balance(@NotNull String currency, @NotNull BigDecimal amount) { + this(currency, amount, List.of()); + } + + @JsonIgnore + public BigDecimal getAvailableAmount() { + return amount.subtract(reservations().stream().map(Reservation::amount).reduce(BigDecimal.ZERO, BigDecimal::add)); + } + } + + public record Reservation( + @NotNull String referenceId, + @NotNull BigDecimal amount + ) { + public Reservation { + // Defensive validation + if (amount.compareTo(BigDecimal.ZERO) <= 0) { + throw new IllegalArgumentException("Reservation amount must be positive"); + } + } + } +} diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/control/AkcesAggregateControllerTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/control/AkcesAggregateControllerTests.java index 4c26a78c..ee268b33 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/control/AkcesAggregateControllerTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/control/AkcesAggregateControllerTests.java @@ -72,7 +72,7 @@ public void testDeserialization() throws JsonProcessingException { ObjectMapper objectMapper = new ObjectMapper(); AkcesControlRecord deserialized = objectMapper.readValue(serializedRecord, AggregateServiceRecord.class); assertNotNull(deserialized); - assertTrue(deserialized instanceof AggregateServiceRecord); + assertInstanceOf(AggregateServiceRecord.class, deserialized); assertEquals("Account", ((AggregateServiceRecord) deserialized).aggregateName()); assertEquals("Account-Commands", ((AggregateServiceRecord) deserialized).commandTopic()); } @@ -92,7 +92,7 @@ public void testSerde() { AkcesControlRecord deserialized = serde.deserializer().deserialize("Akces-Control", serialized); assertNotNull(deserialized); - assertTrue(deserialized instanceof AggregateServiceRecord); + assertInstanceOf(AggregateServiceRecord.class, deserialized); assertEquals("Account", ((AggregateServiceRecord) deserialized).aggregateName()); } } diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEvent.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEvent.java index 5a2032eb..ad5ceee7 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEvent.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEvent.java @@ -22,8 +22,11 @@ import org.elasticsoftware.akces.events.DomainEvent; @DomainEventInfo(type = "AccountCreatedEvent", version = 1) -public record AccountCreatedEvent(@NotNull String userId, @NotNull String lastName, - @NotNull AccountTypeV1 type) implements DomainEvent { +public record AccountCreatedEvent( + @NotNull String userId, + @NotNull String lastName, + @NotNull AccountTypeV1 type +) implements DomainEvent { @Override public String getAggregateId() { return userId(); diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV2.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV2.java index 35c80190..fd1835f5 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV2.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV2.java @@ -26,7 +26,7 @@ public record AccountCreatedEventV2( @NotNull String userId, @NotNull String lastName, @NotNull AccountTypeV2 type, - String firstName, + @NotNull String firstName, String country ) implements DomainEvent { @Override diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV3.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV3.java index 6bb92300..f249237c 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV3.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/AccountCreatedEventV3.java @@ -26,7 +26,7 @@ public record AccountCreatedEventV3( @NotNull String userId, @NotNull String lastName, @NotNull AccountTypeV2 type, - String firstName, + @NotNull String firstName, String country, String city ) implements DomainEvent { diff --git a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/JsonSchemaTests.java b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/JsonSchemaTests.java index 753a3138..a4750636 100644 --- a/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/JsonSchemaTests.java +++ b/main/runtime/src/test/java/org/elasticsoftware/akcestest/schemas/JsonSchemaTests.java @@ -22,13 +22,14 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import com.github.victools.jsonschema.generator.*; import com.github.victools.jsonschema.module.jackson.JacksonModule; import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationModule; import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationOption; -import io.confluent.kafka.schemaregistry.CompatibilityLevel; -import io.confluent.kafka.schemaregistry.SimpleParsedSchemaHolder; import io.confluent.kafka.schemaregistry.json.JsonSchema; +import io.confluent.kafka.schemaregistry.json.diff.Difference; +import io.confluent.kafka.schemaregistry.json.diff.SchemaDiff; import org.elasticsoftware.akces.gdpr.jackson.AkcesGDPRModule; import org.elasticsoftware.akces.serialization.BigDecimalSerializer; import org.elasticsoftware.akcestest.aggregate.wallet.InvalidAmountErrorEvent; @@ -77,21 +78,34 @@ public class JsonSchemaTests { public void testSchemaCompatibility() throws IOException { SchemaGenerator generator = createSchemaGenerator(); - JsonNode schemaV1 = generator.generateSchema(AccountCreatedEvent.class); - JsonNode schemaV2 = generator.generateSchema(AccountCreatedEventV2.class); - - System.out.println(schemaV1.toString()); - System.out.println(schemaV2.toString()); + ObjectNode schemaV1 = generator.generateSchema(AccountCreatedEvent.class); + ObjectNode schemaV2 = generator.generateSchema(AccountCreatedEventV2.class); + ObjectNode schemaV3 = generator.generateSchema(AccountCreatedEventV3.class); JsonSchema schema1 = new JsonSchema(schemaV1); JsonSchema schema2 = new JsonSchema(schemaV2); + JsonSchema schema3 = new JsonSchema(schemaV3); + + List differencesV2 = SchemaDiff.compare(schema1.rawSchema(), schema2.rawSchema()) + .stream().filter(diff -> + !SchemaDiff.COMPATIBLE_CHANGES.contains(diff.getType()) && + !Difference.Type.REQUIRED_PROPERTY_ADDED_TO_UNOPEN_CONTENT_MODEL.equals(diff.getType())).toList(); + + assertEquals(0, differencesV2.size()); - assertEquals(schema2.isCompatible(CompatibilityLevel.BACKWARD_TRANSITIVE, List.of(new SimpleParsedSchemaHolder(schema1))).size(), 0); + List differencesV3 = SchemaDiff.compare(schema2.rawSchema(), schema3.rawSchema()) + .stream().filter(diff -> + !SchemaDiff.COMPATIBLE_CHANGES.contains(diff.getType()) && + !Difference.Type.REQUIRED_PROPERTY_ADDED_TO_UNOPEN_CONTENT_MODEL.equals(diff.getType())).toList(); - schema2.validate(schema2.toJson(new AccountCreatedEvent("1", "Musk", AccountTypeV1.PREMIUM))); + assertEquals(0, differencesV3.size()); + + schema1.validate(schema1.toJson(new AccountCreatedEvent("1", "Musk", AccountTypeV1.PREMIUM))); schema2.validate(schema2.toJson(new AccountCreatedEventV2("1", "Musk", AccountTypeV2.PREMIUM, "Elon", "US"))); + schema3.validate(schema3.toJson(new AccountCreatedEventV3("1", "Musk", AccountTypeV2.GOLD, "Elon", "NL", "Amsterdam"))); + // schema2.validate(new AccountCreatedEvent("1", null, AccountTypeV1.PREMIUM)); //Assert.assertEquals(schema2.isBackwardCompatible(schema1).size(), 0); diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/EncryptingGDPRContext.java b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/EncryptingGDPRContext.java index 434b120c..51abc67d 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/EncryptingGDPRContext.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/EncryptingGDPRContext.java @@ -118,7 +118,7 @@ public String decrypt(@Nullable String encryptedData) { @Override @Nonnull - public String getAggregateId() { + public String aggregateId() { return aggregateId; } diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/GDPRContext.java b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/GDPRContext.java index c74e65f8..fe2978d8 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/GDPRContext.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/GDPRContext.java @@ -29,7 +29,7 @@ public sealed interface GDPRContext permits NoopGDPRContext, EncryptingGDPRConte String decrypt(@Nullable String encryptedData); @Nonnull - String getAggregateId(); + String aggregateId(); @Nullable default byte[] getEncryptionKey() { diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/InMemoryGDPRContextRepository.java b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/InMemoryGDPRContextRepository.java index 8ffebe36..a9f1d22a 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/InMemoryGDPRContextRepository.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/InMemoryGDPRContextRepository.java @@ -23,9 +23,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.requests.ProduceResponse; +import org.elasticsoftware.akces.kafka.RecordAndMetadata; import org.elasticsoftware.akces.protocol.GDPRKeyRecord; import org.elasticsoftware.akces.protocol.ProtocolRecord; -import org.elasticsoftware.akces.kafka.RecordAndMetadata; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -79,7 +79,7 @@ public void commit() { .map(recordMetadata -> recordMetadata != null ? recordMetadata.offset() : ProduceResponse.INVALID_OFFSET) .max(Long::compareTo).orElse(ProduceResponse.INVALID_OFFSET); log.trace("Committing {} records and offset {}", transactionStateRecordMap.size(), this.offset); - transactionStateRecordMap.values().forEach(recordAndMetadata -> stateRecordMap.put(recordAndMetadata.record().aggregateId(), (GDPRKeyRecord) recordAndMetadata.record())); + transactionStateRecordMap.values().forEach(recordAndMetadata -> stateRecordMap.put(recordAndMetadata.record().aggregateId(), recordAndMetadata.record())); transactionStateRecordMap.clear(); } } diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/NoopGDPRContext.java b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/NoopGDPRContext.java index f532e33d..9db7b052 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/NoopGDPRContext.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/NoopGDPRContext.java @@ -20,9 +20,7 @@ import jakarta.annotation.Nonnull; import jakarta.annotation.Nullable; -public final class NoopGDPRContext implements GDPRContext { - private final String aggregateId; - +public record NoopGDPRContext(String aggregateId) implements GDPRContext { public NoopGDPRContext(@Nonnull String aggregateId) { this.aggregateId = aggregateId; } @@ -41,7 +39,7 @@ public String decrypt(@Nullable String encryptedData) { @Nonnull @Override - public String getAggregateId() { + public String aggregateId() { return aggregateId; } } diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/jackson/AkcesGDPRModule.java b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/jackson/AkcesGDPRModule.java index 4c8801a2..2e1466ad 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/jackson/AkcesGDPRModule.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/gdpr/jackson/AkcesGDPRModule.java @@ -63,7 +63,7 @@ public static Version generateVersion(Semver semver) { semver.getMajor(), semver.getMinor(), semver.getPatch(), - !semver.getPreRelease().isEmpty() ? semver.getPreRelease().get(0) : null, + !semver.getPreRelease().isEmpty() ? semver.getPreRelease().getFirst() : null, "org.elasticsoftwarefoundation.akces", "akces-runtime"); } else { diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/schemas/KafkaSchemaRegistry.java b/main/shared/src/main/java/org/elasticsoftware/akces/schemas/KafkaSchemaRegistry.java index c6a11c8d..597c3a4a 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/schemas/KafkaSchemaRegistry.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/schemas/KafkaSchemaRegistry.java @@ -24,9 +24,7 @@ import com.github.victools.jsonschema.module.jackson.JacksonModule; import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationModule; import com.github.victools.jsonschema.module.jakarta.validation.JakartaValidationOption; -import io.confluent.kafka.schemaregistry.CompatibilityLevel; import io.confluent.kafka.schemaregistry.ParsedSchema; -import io.confluent.kafka.schemaregistry.SimpleParsedSchemaHolder; import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; import io.confluent.kafka.schemaregistry.json.JsonSchema; @@ -43,7 +41,6 @@ import java.util.List; import java.util.Map; import java.util.Objects; -import java.util.stream.Collectors; public class KafkaSchemaRegistry { private static final Logger logger = LoggerFactory.getLogger(KafkaSchemaRegistry.class); @@ -64,7 +61,7 @@ public JsonSchema validate(DomainEventType domainEventType) throws SchemaExce return validate(domainEventType, false); } - private JsonSchema validate(SchemaType schemaType, boolean strict) throws SchemaException { + private JsonSchema validate(SchemaType schemaType, boolean strict) throws SchemaException { try { logger.info("Validating schema {} v{}", schemaType.getSchemaName(), schemaType.version()); JsonSchema localSchema = generateJsonSchema(schemaType); @@ -128,7 +125,7 @@ private JsonSchema validate(SchemaType schemaType, boolean strict) throws Schema } } - public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegisterOnIncompatibleSchema) throws SchemaException { + public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegisterOnIncompatibleSchema) throws SchemaException { try { // generate the local schema version JsonSchema localSchema = generateJsonSchema(schemaType); @@ -238,9 +235,9 @@ public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegist schemaType.typeClass()); } else { // ensure we have an ordered list of schemas - registeredSchemas.sort(Comparator.comparingInt(ParsedSchema::version)); + registeredSchemas.sort(Comparator.comparingInt(parsedSchema -> getSchemaVersion(schemaType, parsedSchema))); // see if the new version is exactly one higher than the last version - if (schemaType.version() != registeredSchemas.getLast().version() + 1) { + if (schemaType.version() != getSchemaVersion(schemaType,registeredSchemas.getLast()) + 1) { throw new InvalidSchemaVersionException( schemaName, registeredSchemas.getLast().version(), @@ -248,10 +245,13 @@ public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegist schemaType.typeClass()); } // see if the new schema is backwards compatible with the previous ones - List compatibilityErrors = localSchema.isCompatible(CompatibilityLevel.BACKWARD_TRANSITIVE, - registeredSchemas.stream().map(SimpleParsedSchemaHolder::new) - .collect(Collectors.toList())); - if (compatibilityErrors.isEmpty()) { + List differences = SchemaDiff.compare( + ((JsonSchema)registeredSchemas.getLast()).rawSchema(), + localSchema.rawSchema()) + .stream().filter(diff -> + !SchemaDiff.COMPATIBLE_CHANGES.contains(diff.getType()) && + !Difference.Type.REQUIRED_PROPERTY_ADDED_TO_UNOPEN_CONTENT_MODEL.equals(diff.getType())).toList(); + if (differences.isEmpty()) { // register the new schema schemaRegistryClient.register( schemaName, @@ -259,13 +259,13 @@ public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegist schemaType.version(), -1); } else { - // incomp + // incompatible throw new SchemaNotBackwardsCompatibleException( schemaName, - registeredSchemas.getLast().version(), + getSchemaVersion(schemaType,registeredSchemas.getLast()), schemaType.version(), schemaType.typeClass(), - compatibilityErrors); + differences); } } } @@ -280,11 +280,11 @@ public JsonSchema registerAndValidate(SchemaType schemaType, boolean forceRegist } } - public JsonSchema generateJsonSchema(SchemaType schemaType) { + public JsonSchema generateJsonSchema(SchemaType schemaType) { return new JsonSchema(schemaGeneratorTheadLocal.get().generateSchema(schemaType.typeClass()), List.of(), Map.of(), schemaType.version()); } - private int getSchemaVersion(SchemaType schemaType, ParsedSchema parsedSchema) { + private int getSchemaVersion(SchemaType schemaType, ParsedSchema parsedSchema) { try { return schemaRegistryClient.getVersion(schemaType.getSchemaName(), parsedSchema); } catch (IOException | RestClientException e) { diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/schemas/SchemaNotBackwardsCompatibleException.java b/main/shared/src/main/java/org/elasticsoftware/akces/schemas/SchemaNotBackwardsCompatibleException.java index 1e2d4cee..0d610307 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/schemas/SchemaNotBackwardsCompatibleException.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/schemas/SchemaNotBackwardsCompatibleException.java @@ -17,18 +17,20 @@ package org.elasticsoftware.akces.schemas; +import io.confluent.kafka.schemaregistry.json.diff.Difference; + import java.util.List; public class SchemaNotBackwardsCompatibleException extends SchemaException { private final int previousSchemaVersion; private final int schemaVersion; - private final List differences; + private final List differences; public SchemaNotBackwardsCompatibleException(String schemaIdentifier, int previousSchemaVersion, int schemaVersion, Class implementationClass, - List differences) { + List differences) { super("Schema not backwards compatible with previous version: " + previousSchemaVersion, schemaIdentifier, implementationClass); this.previousSchemaVersion = previousSchemaVersion; this.schemaVersion = schemaVersion; @@ -43,7 +45,7 @@ public int getSchemaVersion() { return schemaVersion; } - public List getDifferences() { + public List getDifferences() { return differences; } diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/serialization/AkcesControlRecordSerde.java b/main/shared/src/main/java/org/elasticsoftware/akces/serialization/AkcesControlRecordSerde.java index de3b563c..e542af72 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/serialization/AkcesControlRecordSerde.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/serialization/AkcesControlRecordSerde.java @@ -47,12 +47,7 @@ public Deserializer deserializer() { return deserializer; } - private static class SerializerImpl implements Serializer { - private final ObjectMapper objectMapper; - - private SerializerImpl(ObjectMapper objectMapper) { - this.objectMapper = objectMapper; - } + private record SerializerImpl(ObjectMapper objectMapper) implements Serializer { @Override public byte[] serialize(String topic, AkcesControlRecord data) { @@ -68,12 +63,7 @@ public byte[] serialize(String topic, AkcesControlRecord data) { } } - private static class DeserializerImpl implements Deserializer { - private final ObjectMapper objectMapper; - - private DeserializerImpl(ObjectMapper objectMapper) { - this.objectMapper = objectMapper; - } + private record DeserializerImpl(ObjectMapper objectMapper) implements Deserializer { @Override public AkcesControlRecord deserialize(String topic, byte[] data) { diff --git a/main/shared/src/main/java/org/elasticsoftware/akces/serialization/ProtocolRecordSerde.java b/main/shared/src/main/java/org/elasticsoftware/akces/serialization/ProtocolRecordSerde.java index 03c77b00..9feba4df 100644 --- a/main/shared/src/main/java/org/elasticsoftware/akces/serialization/ProtocolRecordSerde.java +++ b/main/shared/src/main/java/org/elasticsoftware/akces/serialization/ProtocolRecordSerde.java @@ -181,16 +181,6 @@ public ProtocolRecordSerde() { } } - @Override - public void configure(Map configs, boolean isKey) { - - } - - @Override - public void close() { - - } - @Override public Serializer serializer() { return serializer; @@ -201,25 +191,13 @@ public Deserializer deserializer() { return deserializer; } - private static class SerializerImpl implements Serializer { - private final ObjectWriter domainEventRecordWriter; - private final ObjectWriter aggregateStateRecordWriter; - private final ObjectWriter commandRecordWriter; - private final ObjectWriter gdprKeyRecordWriter; - private final ObjectWriter commandResponseRecordWriter; - - private SerializerImpl(ObjectWriter domainEventRecordWriter, - ObjectWriter aggregateStateRecordWriter, - ObjectWriter commandRecordWriter, - ObjectWriter gdprKeyRecordWriter, - ObjectWriter commandResponseRecordWriter) { - this.domainEventRecordWriter = domainEventRecordWriter; - this.aggregateStateRecordWriter = aggregateStateRecordWriter; - this.commandRecordWriter = commandRecordWriter; - this.gdprKeyRecordWriter = gdprKeyRecordWriter; - this.commandResponseRecordWriter = commandResponseRecordWriter; - } - + private record SerializerImpl( + ObjectWriter domainEventRecordWriter, + ObjectWriter aggregateStateRecordWriter, + ObjectWriter commandRecordWriter, + ObjectWriter gdprKeyRecordWriter, + ObjectWriter commandResponseRecordWriter + ) implements Serializer { @Override public byte[] serialize(String topic, ProtocolRecord data) { @@ -243,24 +221,13 @@ public byte[] serialize(String topic, ProtocolRecord data) { } } - private static class DeserializerImpl implements Deserializer { - private final ObjectReader domainEventRecordReader; - private final ObjectReader aggregateStateRecordReader; - private final ObjectReader commandRecordReader; - private final ObjectReader gdprKeyRecordReader; - private final ObjectReader commandResponseRecordReader; - - public DeserializerImpl(ObjectReader domainEventRecordReader, - ObjectReader aggregateStateRecordReader, - ObjectReader commandRecordReader, - ObjectReader gdprKeyRecordReader, - ObjectReader commandResponseRecordReader) { - this.domainEventRecordReader = domainEventRecordReader; - this.aggregateStateRecordReader = aggregateStateRecordReader; - this.commandRecordReader = commandRecordReader; - this.gdprKeyRecordReader = gdprKeyRecordReader; - this.commandResponseRecordReader = commandResponseRecordReader; - } + private record DeserializerImpl( + ObjectReader domainEventRecordReader, + ObjectReader aggregateStateRecordReader, + ObjectReader commandRecordReader, + ObjectReader gdprKeyRecordReader, + ObjectReader commandResponseRecordReader + ) implements Deserializer { @Override public ProtocolRecord deserialize(String topic, byte[] data) { diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/cryptomarket/CryptoMarket.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/cryptomarket/CryptoMarket.java index ccb2d471..29cd5090 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/cryptomarket/CryptoMarket.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/cryptomarket/CryptoMarket.java @@ -26,7 +26,10 @@ import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.commands.CreateCryptoMarketCommand; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.commands.PlaceMarketOrderCommand; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.data.Side; -import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.*; +import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.CryptoMarketCreatedEvent; +import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderFilledEvent; +import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderPlacedEvent; +import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderRejectedErrorEvent; import org.elasticsoftware.cryptotrading.services.coinbase.CoinbaseService; import org.elasticsoftware.cryptotrading.services.coinbase.Ticker; diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/BuyOrderProcess.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/BuyOrderProcess.java index f7ff6aee..004b2af0 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/BuyOrderProcess.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/BuyOrderProcess.java @@ -18,8 +18,8 @@ package org.elasticsoftware.cryptotrading.aggregates.orders; import org.elasticsoftware.cryptotrading.aggregates.orders.commands.RejectOrderCommand; -import org.elasticsoftware.cryptotrading.aggregates.orders.events.BuyOrderRejectedEvent; import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; +import org.elasticsoftware.cryptotrading.aggregates.orders.events.BuyOrderRejectedEvent; import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InsufficientFundsErrorEvent; import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InvalidCryptoCurrencyErrorEvent; diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcess.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcess.java index cb656fec..44bf3769 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcess.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcess.java @@ -21,8 +21,8 @@ import com.fasterxml.jackson.annotation.JsonTypeInfo; import org.elasticsoftware.akces.processmanager.AkcesProcess; import org.elasticsoftware.cryptotrading.aggregates.orders.commands.RejectOrderCommand; -import org.elasticsoftware.cryptotrading.aggregates.orders.events.BuyOrderRejectedEvent; import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; +import org.elasticsoftware.cryptotrading.aggregates.orders.events.BuyOrderRejectedEvent; import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InsufficientFundsErrorEvent; import org.elasticsoftware.cryptotrading.aggregates.wallet.events.InvalidCryptoCurrencyErrorEvent; diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManager.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManager.java index 0293ab0b..7b9063c9 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManager.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManager.java @@ -22,8 +22,8 @@ import org.elasticsoftware.akces.commands.CommandBus; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.cryptotrading.aggregates.account.events.AccountCreatedEvent; -import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.data.Side; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.commands.PlaceMarketOrderCommand; +import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.data.Side; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderFilledEvent; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.events.MarketOrderRejectedErrorEvent; import org.elasticsoftware.cryptotrading.aggregates.orders.commands.FillBuyOrderCommand; diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManagerState.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManagerState.java index 5abd7035..74e18fee 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManagerState.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/orders/OrderProcessManagerState.java @@ -19,11 +19,13 @@ import jakarta.validation.constraints.NotNull; import org.elasticsoftware.akces.annotations.AggregateIdentifier; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; import org.elasticsoftware.akces.processmanager.ProcessManagerState; import org.elasticsoftware.akces.processmanager.UnknownAkcesProcessException; import java.util.List; +@AggregateStateInfo(type = "OrderProcessManager", version = 1) public record OrderProcessManagerState( @NotNull @AggregateIdentifier String userId, List runningProcesses diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/Wallet.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/Wallet.java index 7750cbe2..534de922 100644 --- a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/Wallet.java +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/Wallet.java @@ -25,10 +25,7 @@ import org.elasticsoftware.akces.annotations.EventSourcingHandler; import org.elasticsoftware.akces.events.DomainEvent; import org.elasticsoftware.cryptotrading.aggregates.account.events.AccountCreatedEvent; -import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CreateBalanceCommand; -import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CreateWalletCommand; -import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CreditWalletCommand; -import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.ReserveAmountCommand; +import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.*; import org.elasticsoftware.cryptotrading.aggregates.wallet.events.*; import java.math.BigDecimal; @@ -37,7 +34,7 @@ import java.util.stream.Stream; -@AggregateInfo(value = "Wallet", version = 1, indexed = true, indexName = "Users") +@AggregateInfo(value = "Wallet", stateVersion = 1, indexed = true, indexName = "Users") @SuppressWarnings("unused") public final class Wallet implements Aggregate { @Override @@ -50,6 +47,10 @@ public Class getStateClass() { return WalletState.class; } + public WalletStateV2 upcast(WalletState state) { + return new WalletStateV2(state.id(), state.balances().stream().map(b -> new WalletStateV2.Balance(b.currency(), b.amount(), new ArrayList<>())).toList()); + } + @CommandHandler(create = true, produces = WalletCreatedEvent.class, errors = {}) public @NotNull Stream create(@NotNull CreateWalletCommand cmd, WalletState isNull) { return Stream.of(new WalletCreatedEvent(cmd.id()), new BalanceCreatedEvent(cmd.id(), cmd.currency())); @@ -75,6 +76,40 @@ public Stream credit(@NotNull CreditWalletCommand cmd, @NotNull Wal return Stream.of(new WalletCreditedEvent(currentState.id(), cmd.currency(), cmd.amount(), balance.amount().add(cmd.amount()))); } + @CommandHandler(produces = WalletDebitedEvent.class, errors = {InvalidCryptoCurrencyErrorEvent.class, InvalidAmountErrorEvent.class, InsufficientFundsErrorEvent.class}) + @NotNull + public Stream debit(@NotNull DebitWalletCommand cmd, @NotNull WalletState currentState) { + WalletState.Balance balance = currentState.balances().stream() + .filter(b -> b.currency().equals(cmd.currency())) + .findFirst() + .orElse(null); + + if (balance == null) { + return Stream.of(new InvalidCryptoCurrencyErrorEvent(cmd.id(), cmd.currency())); + } + + if (cmd.amount().compareTo(BigDecimal.ZERO) <= 0) { + return Stream.of(new InvalidAmountErrorEvent(cmd.id(), cmd.currency())); + } + + if (balance.getAvailableAmount().compareTo(cmd.amount()) < 0) { + return Stream.of(new InsufficientFundsErrorEvent( + cmd.id(), + cmd.currency(), + balance.getAvailableAmount(), + cmd.amount(), + null + )); + } + + return Stream.of(new WalletDebitedEvent( + currentState.id(), + cmd.currency(), + cmd.amount(), + balance.amount().subtract(cmd.amount()) + )); + } + @CommandHandler(produces = AmountReservedEvent.class, errors = {InvalidCryptoCurrencyErrorEvent.class, InvalidAmountErrorEvent.class, InsufficientFundsErrorEvent.class}) public Stream makeReservation(ReserveAmountCommand command, WalletState state) { WalletState.Balance balance = state.balances().stream().filter(b -> b.currency().equals(command.currency())).findFirst().orElse(null); diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/WalletStateV2.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/WalletStateV2.java new file mode 100644 index 00000000..9e7da474 --- /dev/null +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/WalletStateV2.java @@ -0,0 +1,65 @@ +/* + * Copyright 2022 - 2025 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.cryptotrading.aggregates.wallet; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.aggregate.AggregateState; +import org.elasticsoftware.akces.annotations.AggregateIdentifier; +import org.elasticsoftware.akces.annotations.AggregateStateInfo; + +import java.math.BigDecimal; +import java.util.List; + +@AggregateStateInfo(type = "Wallet", version = 2) +public record WalletStateV2( + @AggregateIdentifier @NotNull String id, + List balances +) implements AggregateState { + @Override + public String getAggregateId() { + return id(); + } + + public record Balance(@NotNull String currency,@NotNull BigDecimal amount,@NotNull List reservations) { + public Balance(@NotNull String currency) { + this(currency, BigDecimal.ZERO, List.of()); + } + + public Balance(@NotNull String currency, @NotNull BigDecimal amount) { + this(currency, amount, List.of()); + } + + @JsonIgnore + public BigDecimal getAvailableAmount() { + return amount.subtract(reservations().stream().map(Reservation::amount).reduce(BigDecimal.ZERO, BigDecimal::add)); + } + } + + public record Reservation( + @NotNull String referenceId, + @NotNull BigDecimal amount + ) { + public Reservation { + // Defensive validation + if (amount.compareTo(BigDecimal.ZERO) <= 0) { + throw new IllegalArgumentException("Reservation amount must be positive"); + } + } + } +} diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/commands/DebitWalletCommand.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/commands/DebitWalletCommand.java new file mode 100644 index 00000000..a5c05ac6 --- /dev/null +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/commands/DebitWalletCommand.java @@ -0,0 +1,36 @@ +/* + * Copyright 2022 - 2025 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.cryptotrading.aggregates.wallet.commands; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.annotations.CommandInfo; +import org.elasticsoftware.akces.commands.Command; + +import java.math.BigDecimal; + +@CommandInfo(type = "DebitWallet", version = 1) +public record DebitWalletCommand( + @NotNull String id, + @NotNull String currency, + @NotNull BigDecimal amount +) implements Command { + @Override + public String getAggregateId() { + return id(); + } +} diff --git a/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/events/WalletDebitedEvent.java b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/events/WalletDebitedEvent.java new file mode 100644 index 00000000..c04687dd --- /dev/null +++ b/test-apps/crypto-trading/aggregates/src/main/java/org/elasticsoftware/cryptotrading/aggregates/wallet/events/WalletDebitedEvent.java @@ -0,0 +1,37 @@ +/* + * Copyright 2022 - 2025 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.cryptotrading.aggregates.wallet.events; + +import jakarta.validation.constraints.NotNull; +import org.elasticsoftware.akces.annotations.DomainEventInfo; +import org.elasticsoftware.akces.events.DomainEvent; + +import java.math.BigDecimal; + +@DomainEventInfo(type = "WalletDebited", version = 1) +public record WalletDebitedEvent( + @NotNull String id, + @NotNull String currency, + @NotNull BigDecimal amount, + @NotNull BigDecimal newBalance +) implements DomainEvent { + @Override + public String getAggregateId() { + return id(); + } +} diff --git a/test-apps/crypto-trading/aggregates/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingApplicationTest.java b/test-apps/crypto-trading/aggregates/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingApplicationTest.java index c6d0762e..84c5750a 100644 --- a/test-apps/crypto-trading/aggregates/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingApplicationTest.java +++ b/test-apps/crypto-trading/aggregates/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingApplicationTest.java @@ -32,8 +32,8 @@ import org.elasticsoftware.akces.protocol.ProtocolRecord; import org.elasticsoftware.cryptotrading.aggregates.account.commands.CreateAccountCommand; import org.elasticsoftware.cryptotrading.aggregates.cryptomarket.commands.CreateCryptoMarketCommand; -import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; import org.elasticsoftware.cryptotrading.aggregates.orders.commands.PlaceBuyOrderCommand; +import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CreateBalanceCommand; import org.elasticsoftware.cryptotrading.aggregates.wallet.commands.CreditWalletCommand; import org.elasticsoftware.cryptotrading.services.coinbase.CoinbaseService; @@ -99,6 +99,7 @@ public class CryptoTradingApplicationTest { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); @@ -281,7 +282,6 @@ void testCreateAllEURMarketsAndMakeATrade() { new CreditWalletCommand(accountId, "EUR", new BigDecimal("1000")))).block(); - ; // place a buy order on BTC-EUR market String clientOrderId = "479ab2a4-d19e-4116-9f7e-cf13dca5763a"; diff --git a/test-apps/crypto-trading/commands/src/main/java/org/elasticsoftware/cryptotrading/web/dto/BuyOrderInput.java b/test-apps/crypto-trading/commands/src/main/java/org/elasticsoftware/cryptotrading/web/dto/BuyOrderInput.java index 155aab64..f80ec753 100644 --- a/test-apps/crypto-trading/commands/src/main/java/org/elasticsoftware/cryptotrading/web/dto/BuyOrderInput.java +++ b/test-apps/crypto-trading/commands/src/main/java/org/elasticsoftware/cryptotrading/web/dto/BuyOrderInput.java @@ -1,7 +1,7 @@ package org.elasticsoftware.cryptotrading.web.dto; -import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; import org.elasticsoftware.cryptotrading.aggregates.orders.commands.PlaceBuyOrderCommand; +import org.elasticsoftware.cryptotrading.aggregates.orders.data.CryptoMarket; import java.math.BigDecimal; diff --git a/test-apps/crypto-trading/commands/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingCommandApiTest.java b/test-apps/crypto-trading/commands/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingCommandApiTest.java index 97b6544a..891dc34b 100644 --- a/test-apps/crypto-trading/commands/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingCommandApiTest.java +++ b/test-apps/crypto-trading/commands/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingCommandApiTest.java @@ -91,6 +91,7 @@ public class CryptoTradingCommandApiTest { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka); diff --git a/test-apps/crypto-trading/queries/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingQueryApiTest.java b/test-apps/crypto-trading/queries/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingQueryApiTest.java index 2e6619eb..45caf30b 100644 --- a/test-apps/crypto-trading/queries/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingQueryApiTest.java +++ b/test-apps/crypto-trading/queries/src/test/java/org/elasticsoftware/cryptotrading/CryptoTradingQueryApiTest.java @@ -105,6 +105,7 @@ public class CryptoTradingQueryApiTest { .withNetwork(network) .withEnv("SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS", "kafka:9092") .withEnv("SCHEMA_REGISTRY_HOST_NAME", "localhost") + .withEnv("SCHEMA_REGISTRY_SCHEMA_COMPATIBILITY_LEVEL","none") .withExposedPorts(8081) .withNetworkAliases("schema-registry") .dependsOn(kafka);