add aggregate support

Author: vanrogu

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/aggregates/Aggregate.java

@@ -0,0 +1,26 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.aggregates;
+
+import org.sliceworkz.eventstore.events.EventHandler;
+
+public interface Aggregate<DOMAIN_EVENT_TYPE> extends EventHandler<DOMAIN_EVENT_TYPE> {
+	
+	void setContext ( AggregateContext<DOMAIN_EVENT_TYPE> aggregateContext );
+
+}

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/aggregates/AggregateCapability.java

@@ -0,0 +1,26 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.aggregates;
+
+import org.sliceworkz.eventstore.events.Tags;
+
+public interface AggregateCapability<DOMAIN_EVENT_TYPE> {
+	
+	<T extends Aggregate<DOMAIN_EVENT_TYPE>> T aggregate ( Class<T> aggregateClass, Tags identity );
+
+}

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/aggregates/AggregateContext.java

@@ -0,0 +1,38 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.aggregates;
+
+import java.util.List;
+
+import org.sliceworkz.eventstore.events.Tags;
+
+public interface AggregateContext<DOMAIN_EVENT_TYPE> {
+	
+	Tags identity ( );
+	
+	void raiseEvent ( DOMAIN_EVENT_TYPE event );
+
+	void raiseEvent ( DOMAIN_EVENT_TYPE event, String idempotencyKey );
+
+	void raiseEvents ( List<DOMAIN_EVENT_TYPE> events );
+	
+	AggregateEventAppender<DOMAIN_EVENT_TYPE> eventAppender ( );
+	
+	void updateFromStream ( );
+
+}

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/aggregates/AggregateEventAppender.java

@@ -0,0 +1,28 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.aggregates;
+
+public interface AggregateEventAppender<DOMAIN_EVENT_TYPE> {
+	
+	AggregateEventAppender<DOMAIN_EVENT_TYPE> add ( DOMAIN_EVENT_TYPE event );
+	
+	AggregateEventAppender<DOMAIN_EVENT_TYPE> add ( DOMAIN_EVENT_TYPE event, String idempotencyKey );
+
+	void append ( );
+	
+}

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/boundedcontext/AllCapabilities.java

@@ -17,9 +17,12 @@
  */
 package org.sliceworkz.eventmodeling.boundedcontext;
 
+import org.sliceworkz.eventmodeling.aggregates.AggregateCapability;
+
 public interface AllCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> extends 
 	CQRSCapabilities<DOMAIN_EVENT_TYPE>, 
-	DCBCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE>, 
+	DCBCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE>,
+	AggregateCapability<DOMAIN_EVENT_TYPE>,
 	FeatureSliceCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> {
 
 }

sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/boundedcontext/BoundedContextBuilder.java

@@ -19,6 +19,7 @@
 
 import java.util.function.Predicate;
 
+import org.sliceworkz.eventmodeling.aggregates.Aggregate;
 import org.sliceworkz.eventmodeling.automation.Automation;
 import org.sliceworkz.eventmodeling.events.Instance;
 import org.sliceworkz.eventmodeling.inbound.Translator;
@@ -50,6 +51,8 @@ BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE
 
 	BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> rootPackage(Package rootPackage);
 
+	BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> aggregate(Class<? extends Aggregate<DOMAIN_EVENT_TYPE>> aggregateClass);
+
 	LiveModelSpecification<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> readmodel(Class<? extends ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> readModelClass);
 
 	LongLivedReadModelSpecification<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> readmodel(ReadModelWithMetaData<DOMAIN_EVENT_TYPE> readModel);

sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/aggregates/AggregateContextImpl.java

@@ -0,0 +1,81 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.module.aggregates;
+
+import java.util.List;
+
+import org.sliceworkz.eventmodeling.aggregates.Aggregate;
+import org.sliceworkz.eventmodeling.aggregates.AggregateContext;
+import org.sliceworkz.eventmodeling.aggregates.AggregateEventAppender;
+import org.sliceworkz.eventstore.events.EventReference;
+import org.sliceworkz.eventstore.events.Tags;
+import org.sliceworkz.eventstore.projection.Projection;
+import org.sliceworkz.eventstore.projection.Projector;
+import org.sliceworkz.eventstore.stream.EventStream;
+
+public class AggregateContextImpl<DOMAIN_EVENT_TYPE> implements AggregateContext<DOMAIN_EVENT_TYPE> {
+
+	private Tags identity;
+	private Aggregate<DOMAIN_EVENT_TYPE> aggregate;
+	private EventStream<DOMAIN_EVENT_TYPE> eventStream;
+	private Projection<DOMAIN_EVENT_TYPE> projectionTowardsAggregate;
+	private EventReference lastEventReference;
+	private AggregateEventAppender<DOMAIN_EVENT_TYPE> aggregateEventAppender;
+	
+	public AggregateContextImpl ( Tags identity, Aggregate<DOMAIN_EVENT_TYPE> aggregate, EventStream<DOMAIN_EVENT_TYPE> eventStream ) {
+		this.identity = identity;
+		this.aggregate = aggregate;
+		this.eventStream = eventStream;
+		this.projectionTowardsAggregate = new ProjectionTowardsAggregate<>(aggregate, identity);
+		this.aggregateEventAppender = new AggregateEventAppenderImpl<>(eventStream, aggregate, identity, null);
+	}
+	
+	@Override
+	public Tags identity() {
+		return identity;
+	}
+
+	@Override
+	public void raiseEvent(DOMAIN_EVENT_TYPE event) {
+		raiseEvent(event, null);
+	}
+
+	@Override
+	public void raiseEvent(DOMAIN_EVENT_TYPE event, String idempotencyKey) {
+		eventAppender().add(event, idempotencyKey).append();
+	}
+
+	@Override
+	public void raiseEvents(List<DOMAIN_EVENT_TYPE> events) {
+		var eventAppender = eventAppender();
+		events.stream().map(eventAppender::add);
+		eventAppender.append();
+	}
+
+	@Override
+	public AggregateEventAppender<DOMAIN_EVENT_TYPE> eventAppender() {
+		return aggregateEventAppender;
+	}
+
+	@Override
+	public void updateFromStream() {
+		this.lastEventReference = Projector.from(eventStream).towards(projectionTowardsAggregate).startingAfter(lastEventReference).build().run().lastEventReference();
+		this.aggregateEventAppender = new AggregateEventAppenderImpl<>(eventStream, aggregate, identity, lastEventReference);
+	}
+	
+}

sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/aggregates/AggregateEventAppenderImpl.java

@@ -0,0 +1,69 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.module.aggregates;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import org.sliceworkz.eventmodeling.aggregates.Aggregate;
+import org.sliceworkz.eventmodeling.aggregates.AggregateEventAppender;
+import org.sliceworkz.eventstore.events.EphemeralEvent;
+import org.sliceworkz.eventstore.events.Event;
+import org.sliceworkz.eventstore.events.EventReference;
+import org.sliceworkz.eventstore.events.Tags;
+import org.sliceworkz.eventstore.query.EventQuery;
+import org.sliceworkz.eventstore.query.EventTypesFilter;
+import org.sliceworkz.eventstore.stream.AppendCriteria;
+import org.sliceworkz.eventstore.stream.EventStream;
+
+public class AggregateEventAppenderImpl<DOMAIN_EVENT_TYPE> implements AggregateEventAppender<DOMAIN_EVENT_TYPE> {
+
+	private List<EphemeralEvent<? extends DOMAIN_EVENT_TYPE>> events = new ArrayList<>();
+	private Aggregate<DOMAIN_EVENT_TYPE> aggregate;
+	private EventStream<DOMAIN_EVENT_TYPE> eventStream;
+	private Tags identity;
+	private EventReference lastReference;
+	
+	public AggregateEventAppenderImpl ( EventStream<DOMAIN_EVENT_TYPE> eventStream, Aggregate<DOMAIN_EVENT_TYPE> aggregate, Tags identity, EventReference lastReference ) {
+		this.eventStream = eventStream;
+		this.aggregate = aggregate;
+		this.identity = identity;
+		this.lastReference = lastReference; 
+	}
+	
+	@Override
+	public AggregateEventAppenderImpl<DOMAIN_EVENT_TYPE> add(DOMAIN_EVENT_TYPE event) {
+		return add(event, null);
+	}
+
+	@Override
+	public AggregateEventAppenderImpl<DOMAIN_EVENT_TYPE> add(DOMAIN_EVENT_TYPE event, String idempotencyKey) {
+		events.add(Event.of(event, identity).withIdempotencyKey(idempotencyKey));
+		return this;
+	}
+
+	@Override
+	public void append() {
+		eventStream.append(
+				AppendCriteria.of(EventQuery.forEvents(EventTypesFilter.any(), identity), Optional.ofNullable(lastReference)),
+				events).stream().map(e->{this.lastReference=e.reference();return e;}).forEach(aggregate::when);
+		events.clear();
+	}
+	
+}

sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/aggregates/AggregateModule.java

@@ -0,0 +1,86 @@
+/*
+ * Sliceworkz Event Modeling - an opinionated Event Modeling framework in Java
+ * Copyright © 2025 Sliceworkz / XTi ([email protected])
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.sliceworkz.eventmodeling.module.aggregates;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.sliceworkz.eventmodeling.aggregates.Aggregate;
+import org.sliceworkz.eventmodeling.aggregates.AggregateCapability;
+import org.sliceworkz.eventstore.events.Tags;
+import org.sliceworkz.eventstore.stream.EventStream;
+
+// TODO add micrometer for observability
+public class AggregateModule<DOMAIN_EVENT_TYPE> implements AggregateCapability<DOMAIN_EVENT_TYPE> {
+	
+	private static final Logger LOGGER = LoggerFactory.getLogger(AggregateModule.class);
+
+	private Map<Class<? extends Aggregate<DOMAIN_EVENT_TYPE>>,Constructor<? extends Aggregate<DOMAIN_EVENT_TYPE>>> aggregateClassesWithConstructor = new HashMap<>();
+	private EventStream<DOMAIN_EVENT_TYPE> domainEventStream;
+	private String boundedContext;
+	
+	public AggregateModule ( String boundedContext, Collection<Class<? extends Aggregate<DOMAIN_EVENT_TYPE>>> aggregateClasses, EventStream<DOMAIN_EVENT_TYPE> domainEventStream ) {
+		this.boundedContext = boundedContext;
+		this.domainEventStream = domainEventStream;
+		aggregateClasses.forEach(aggregateClass->{
+			try {
+				aggregateClassesWithConstructor.put(aggregateClass, aggregateClass.getDeclaredConstructor(new Class[] {}));
+			} catch (NoSuchMethodException | SecurityException e) {
+				LOGGER.error(e.getMessage(), e);
+				throw new RuntimeException(e);
+			}
+		});
+		
+		LOGGER.info("aggregates: %s".formatted(aggregateClasses));
+	}
+	
+	@SuppressWarnings("unchecked")
+	@Override
+	public <T extends Aggregate<DOMAIN_EVENT_TYPE>> T aggregate(Class<T> aggregateClass, Tags identity) {
+		
+		if ( aggregateClassesWithConstructor.containsKey(aggregateClass)) {
+			
+			if ( identity == null || identity.tags().size() < 1 ) {
+				throw new IllegalArgumentException("aggregate identity needs at least one tag, got '%s'".formatted(identity));
+			}
+			
+			T result;
+			try {
+				result = (T) aggregateClassesWithConstructor.get(aggregateClass).newInstance(new Object[] {});
+				
+				AggregateContextImpl<DOMAIN_EVENT_TYPE> aci = new AggregateContextImpl<DOMAIN_EVENT_TYPE> (identity, result, domainEventStream);
+				result.setContext(aci);
+				aci.updateFromStream();
+				
+			} catch (InstantiationException | IllegalAccessException | IllegalArgumentException
+					| InvocationTargetException | SecurityException e) {
+				LOGGER.error(e.getMessage(), e);
+				throw new RuntimeException(e);
+			}
+			return result;
+		} else {
+			throw new IllegalArgumentException("aggregate class '%s' not registered in bounded context '%s'".formatted(aggregateClass, boundedContext));
+		}
+	}
+
+}