Merge branch 'release/0.5' into develop

Author: vanrogu

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/events/EphemeralEvent.java

@@ -54,7 +54,7 @@
  * @see Event
  * @see Tags
  */
-public record EphemeralEvent<DOMAIN_EVENT_TYPE> ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags ) {
+public record EphemeralEvent<DOMAIN_EVENT_TYPE> ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags, String idempotencyKey ) {
 
 	/**
 	 * Constructs an EphemeralEvent with validation of all required fields.
@@ -67,7 +67,7 @@ public record EphemeralEvent<DOMAIN_EVENT_TYPE> ( EventType type, DOMAIN_EVENT_T
 	 * @param tags the tags for this event (required, use Tags.none() if no tags)
 	 * @throws IllegalArgumentException if any required parameter is null
 	 */
-	public EphemeralEvent ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags ) {
+	public EphemeralEvent ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags, String idempotencyKey ) {
 		if ( type == null ) {
 			throw new IllegalArgumentException("type is required on event");
 		}
@@ -80,6 +80,7 @@ public EphemeralEvent ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags ) {
 		this.type = type;
 		this.data = data;
 		this.tags = tags;
+		this.idempotencyKey = idempotencyKey;
 	}
 
 	/**
@@ -92,7 +93,26 @@ public EphemeralEvent ( EventType type, DOMAIN_EVENT_TYPE data, Tags tags ) {
 	 * @return a new EphemeralEvent instance with the specified tags
 	 */
 	public EphemeralEvent<DOMAIN_EVENT_TYPE> withTags ( Tags tags ) {
-		return new EphemeralEvent<>(type, data, tags);
+		return new EphemeralEvent<>(type, data, tags, idempotencyKey);
+	}
+
+	/**
+	 * Creates a copy of this ephemeral event with a different idempotency key.
+	 * <p>
+	 * All other properties remain unchanged. An idempotency key ensures that the same event
+	 * is not appended multiple times if the append operation is retried. When an event with
+	 * an idempotency key is appended a second time, the storage will silently ignore it
+	 * rather than creating a duplicate event.
+	 * <p>
+	 * <b>Important:</b> When appending multiple events in a single batch, none of them may
+	 * have an idempotency key. Idempotency keys can only be used with single-event appends.
+	 *
+	 * @param idempotencyKey the idempotency key to attach to the event, or null for no idempotency check
+	 * @return a new EphemeralEvent instance with the specified idempotency key
+	 * @see org.sliceworkz.eventstore.stream.EventStream#append(org.sliceworkz.eventstore.stream.AppendCriteria, java.util.List)
+	 */
+	public EphemeralEvent<DOMAIN_EVENT_TYPE> withIdempotencyKey ( String idempotencyKey ) {
+		return new EphemeralEvent<>(type, data, tags, idempotencyKey);
 	}
 
 	/**
@@ -107,7 +127,7 @@ public EphemeralEvent<DOMAIN_EVENT_TYPE> withTags ( Tags tags ) {
 	 * @return a new EphemeralEvent instance ready to be appended
 	 */
 	public static final <DOMAIN_EVENT_TYPE> EphemeralEvent<DOMAIN_EVENT_TYPE> of ( DOMAIN_EVENT_TYPE data, Tags tags) {
-		return new EphemeralEvent<DOMAIN_EVENT_TYPE>(EventType.of(data), data, tags);
+		return new EphemeralEvent<DOMAIN_EVENT_TYPE>(EventType.of(data), data, tags, null);
 	}
 
 	/**

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/events/Event.java

@@ -170,7 +170,7 @@ public <OTHER_DOMAIN_EVENT_TYPE extends DOMAIN_EVENT_TYPE> Event<OTHER_DOMAIN_EV
 	 * @param timestamp the timestamp when this event was persisted
 	 * @return a new Event instance
 	 */
-	public static final <DOMAIN_EVENT_TYPE> Event<DOMAIN_EVENT_TYPE> of ( EventStreamId stream, EventReference reference, EventType type, EventType storedType, DOMAIN_EVENT_TYPE data, Tags tags, LocalDateTime timestamp ) {
+	public static final <DOMAIN_EVENT_TYPE> Event<DOMAIN_EVENT_TYPE> of ( EventStreamId stream, EventReference reference, EventType type, EventType storedType, DOMAIN_EVENT_TYPE data, Tags tags, LocalDateTime timestamp) {
 		return new Event<DOMAIN_EVENT_TYPE>(stream, type, storedType, reference, data, tags, timestamp);
 	}
 

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/projection/Projector.java

@@ -516,6 +516,7 @@ public static class Builder<EVENT_TYPE> {
 		private EventSource<EVENT_TYPE> eventSource;
 		private Projection<EVENT_TYPE> projection;
 		private EventReference after;
+		private boolean subscribe;
 		private int maxEventsPerQuery = DEFAULT_MAX_EVENTS_PER_QUERY;
 
 		private BookmarkBuilder bookmarkBuilder = new BookmarkBuilder(this);
@@ -531,6 +532,31 @@ public Builder<EVENT_TYPE> from ( EventSource<EVENT_TYPE> eventSource ) {
 			return this;
 		}
 
+		/**
+		 * Configures the projector to automatically subscribe to the event source for eventually consistent updates.
+		 * <p>
+		 * When enabled, the projector will automatically re-run whenever new events are appended to the event source,
+		 * enabling near-real-time projection updates without manual polling. This is useful for:
+		 * <ul>
+		 *   <li>Live dashboards and read models that need to reflect recent changes quickly</li>
+		 *   <li>Reactive projections that respond to events as they arrive</li>
+		 *   <li>Systems where projection staleness needs to be minimized</li>
+		 * </ul>
+		 * <p>
+		 * Note that the updates are <em>eventually consistent</em> - there may be a small delay between
+		 * event append and projection update. The subscription mechanism is optimized for efficiency
+		 * and may batch multiple events before triggering a projection run.
+		 * <p>
+		 * Without this setting, projections only update when explicitly triggered via {@link Projector#run()}.
+		 *
+		 * @return this builder for method chaining
+		 * @see EventStreamEventuallyConsistentAppendListener
+		 */
+		public Builder<EVENT_TYPE> subscribe ( ) {
+			this.subscribe = true;
+			return this;
+		}
+
 		/**
 		 * Provides access to the bookmark configuration builder.
 		 * <p>
@@ -603,7 +629,12 @@ public Builder<EVENT_TYPE> inBatchesOf ( int maxEventsPerQuery ) {
 		 * @return a new Projector configured with the builder's settings
 		 */
 		public Projector<EVENT_TYPE> build ( ) {
-			return new Projector<>(eventSource, projection, after, maxEventsPerQuery, bookmarkBuilder.readerName, bookmarkBuilder.tags, bookmarkBuilder.bookmarkReadFrequency);
+			Projector<EVENT_TYPE> projector = new Projector<>(eventSource, projection, after, maxEventsPerQuery, bookmarkBuilder.readerName, bookmarkBuilder.tags, bookmarkBuilder.bookmarkReadFrequency);
+			if ( subscribe ) {
+				// subscribe for eventually consistent updates about event appends, so the projector will automatically trigger projection updates
+				eventSource.subscribe(projector);
+			}
+			return projector;
 		}
 
 		/**
@@ -973,9 +1004,25 @@ void stopBatchIfNeeded ( Optional<EventReference> lastEventReference ) {
 
 	}
 
+	/**
+	 * Handles notifications of newly appended events when the projector is subscribed to an event source.
+	 * <p>
+	 * This method is called by the event source when new events are appended, triggering an automatic
+	 * projection run to process the new events. It implements the {@link EventStreamEventuallyConsistentAppendListener}
+	 * interface to enable reactive, near-real-time projection updates.
+	 * <p>
+	 * The method runs the projection and returns the reference of the last processed event. This allows
+	 * the event source to track which events have been successfully processed by this projector.
+	 * <p>
+	 * This method is only invoked if the projector was configured with {@link Builder#subscribe()}.
+	 *
+	 * @param atLeastUntil the reference indicating events have been appended at least up to this point
+	 * @return the reference of the last event processed by this projection run, or null if no events were processed
+	 * @see Builder#subscribe()
+	 */
 	@Override
 	public EventReference eventsAppended(EventReference atLeastUntil) {
 		return run().lastEventReference();
 	}
-	
+
 }

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/spi/EventStorage.java

@@ -502,7 +502,7 @@ interface EventStoreListener {
 	 * @see StoredEvent
 	 * @see #append(AppendCriteria, Optional, List)
 	 */
-	public record EventToStore ( EventStreamId stream, EventType type, String immutableData, String erasableData, Tags tags ) {
+	public record EventToStore ( EventStreamId stream, EventType type, String immutableData, String erasableData, Tags tags, String idempotencyKey ) {
 
 		/**
 		 * Converts this event to a stored event by assigning a reference and timestamp.

sliceworkz-eventstore-benchmark/src/main/java/org/sliceworkz/eventstore/benchmark/BenchmarkApplication.java

@@ -120,26 +120,24 @@ public EventReference eventsAppended(EventReference atLeastUntil) {
 		Projector<CustomerEvent> customerProjector = Projector.<CustomerEvent>newBuilder()
 			.from(customerStream)
 			.towards(customerProjection)
+			.subscribe()
 			.bookmarkProgress()
 				.withReader("customer-projector")
 				.readBeforeFirstExecution()
 				.done()
 			.build();
-		customerStream.subscribe(customerProjector);
-
 
 
 		SupplierEventProjection supplierProjection = new SupplierEventProjection(dataSource);
 		Projector<SupplierEvent> supplierProjector = Projector.<SupplierEvent>newBuilder()
 			.from(supplierStream)
 			.towards(supplierProjection)
+			.subscribe()
 			.bookmarkProgress()
 				.withReader("supplier-projector")
 				.readBeforeFirstExecution()
 				.done()
 			.build();
-		customerStream.subscribe(supplierProjector);
-		
 
 		CustomerEventProducer cep = new CustomerEventProducer(customerStream, EVENTS_PER_PRODUCER_INSTANCE, MS_WAIT_BETWEEN_EVENTS);
 		SupplierEventProducer sep = new SupplierEventProducer(supplierStream, EVENTS_PER_PRODUCER_INSTANCE, MS_WAIT_BETWEEN_EVENTS);

sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/EventStoreImpl.java

@@ -115,14 +115,15 @@ public class EventStoreImpl implements EventStore {
 	private final EventStorage eventStorage;
 
 	/**
-	 * Executor service using virtual threads for asynchronously notifying eventually consistent subscribers.
-	 * Named threads help with debugging and monitoring.
+	 * Executor service using virtual threads for asynchronously notifying eventually consistent subscribers
+	 * about new event appends. Named threads help with debugging and monitoring.
 	 */
 	private final ExecutorService executorServiceForEventAppends;
 
 	/**
-	 * Executor service using virtual threads for asynchronously notifying eventually consistent subscribers.
-	 * Named threads help with debugging and monitoring.
+	 * Executor service using virtual threads for asynchronously notifying eventually consistent subscribers
+	 * about bookmark updates. This uses a single-threaded executor to ensure bookmark notifications are
+	 * processed sequentially. Named threads help with debugging and monitoring.
 	 */
 	private final ExecutorService executorServiceForBookmarkUpdates;
 
@@ -295,7 +296,7 @@ private EventToStore reduce ( EphemeralEvent<? extends EVENT_TYPE> event, EventS
 			meterAppendEvent.increment(); // one event more sent to storage for appending
 			Tags tags = event.tags(); 
 			TypeAndSerializedPayload data = serde.serialize(event.data());
-			return new EventToStore(streamToAppendTo, data.type(), data.immutablePayload(), data.erasablePayload(), tags);
+			return new EventToStore(streamToAppendTo, data.type(), data.immutablePayload(), data.erasablePayload(), tags, event.idempotencyKey());
 		}
 
 		private List<EventToStore> reduce ( List<? extends EphemeralEvent<? extends EVENT_TYPE>> events, EventStreamId streamToAppendTo ) {
@@ -322,6 +323,12 @@ public List<Event<EVENT_TYPE>> append(AppendCriteria appendCriteria, List<Epheme
 			if ( !unAppendable.isEmpty() ) {
 				throw new IllegalArgumentException("cannot append event type '%s' via this stream".formatted(unAppendable.getFirst()));
 			}
+			
+			if ( events.size() > 1 ) {
+				if ( events.stream().filter(e->e.idempotencyKey()!=null).findAny().isPresent()) {
+					throw new IllegalArgumentException("cannot append multiple events in combination with an idempotency key");
+				}
+			}
 
 			// append events to the eventstore (with optimistic locking)
 			List<Event<EVENT_TYPE>> appendedEvents;

sliceworkz-eventstore-infra-inmem/src/main/java/org/sliceworkz/eventstore/infra/inmem/InMemoryEventStorage.java

@@ -213,7 +213,7 @@ public static Builder newBuilder ( ) {
 		 * @see InMemoryEventStorageImpl
 		 */
 		public EventStorage build ( ) {
-			return new InMemoryEventStorageImpl(limit, meterRegistry);
+			return new InMemoryEventStorageImpl(limit);
 		}
 
 		/**

sliceworkz-eventstore-infra-inmem/src/main/java/org/sliceworkz/eventstore/infra/inmem/InMemoryEventStorageImpl.java

@@ -22,10 +22,12 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -45,8 +47,6 @@
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.json.JsonMapper;
 
-import io.micrometer.core.instrument.MeterRegistry;
-
 /**
  * Thread-safe in-memory implementation of the {@link EventStorage} interface.
  * <p>
@@ -98,11 +98,11 @@ public class InMemoryEventStorageImpl implements EventStorage {
 
 	private String name;
 	private List<StoredEvent> eventlog = new CopyOnWriteArrayList<>();
+	private Set<String> idempotencyKeys = new HashSet<>();
 	private List<WeakReference<EventStoreListener>> listeners = new CopyOnWriteArrayList<>();
 	private Map<String,EventReference> bookmarks = new HashMap<>();
 	private JsonMapper jsonMapper;
 	private Limit absoluteLimit;
-	private MeterRegistry meterRegistry;
 
 	/**
 	 * Constructs a new in-memory event storage instance with the specified absolute query limit and observability support.
@@ -117,19 +117,16 @@ public class InMemoryEventStorageImpl implements EventStorage {
 	 *   <li>An empty list of event listeners</li>
 	 *   <li>An empty bookmark map</li>
 	 *   <li>A Jackson {@link JsonMapper} with auto-discovered modules for event serialization validation</li>
-	 *   <li>A Micrometer meter registry for collecting metrics about storage operations</li>
 	 * </ul>
 	 *
 	 * @param absoluteLimit the absolute limit on query results, or {@link Limit#none()} for no limit
-	 * @param meterRegistry the Micrometer meter registry for collecting observability metrics
 	 * @see InMemoryEventStorage.Builder#build()
 	 */
-	public InMemoryEventStorageImpl ( Limit absoluteLimit, MeterRegistry meterRegistry ) {
+	public InMemoryEventStorageImpl ( Limit absoluteLimit ) {
 		this.name = "inmem-%s".formatted(System.identityHashCode(this)); // unique name in case different objects are used
 		this.jsonMapper = new JsonMapper();
 		this.jsonMapper.findAndRegisterModules();
 		this.absoluteLimit = absoluteLimit;
-		this.meterRegistry = meterRegistry;
 	}
 
 	/**
@@ -333,7 +330,7 @@ private void verifyPersistableJson ( List<EventToStore> newEvents ) {
 	}
 
 	private List<StoredEvent> addAndNotifyListeners ( List<EventToStore> events ) {
-		var addedEvents = events.stream().map(this::addEventToEventLog).toList();
+		var addedEvents = events.stream().map(this::addEventToEventLog).filter(e->e!=null).toList();
 
 		// notify each Listener about the appends, but if multiple Events were appended, only notify about the last one
 		addedEvents.stream()
@@ -353,6 +350,14 @@ private List<StoredEvent> addAndNotifyListeners ( List<EventToStore> events ) {
 	}
 
 	private StoredEvent addEventToEventLog ( EventToStore event ) {
+		
+		if ( event.idempotencyKey() != null ) {
+			if ( idempotencyKeys.contains(event.idempotencyKey())) {
+				return null;
+			}
+			idempotencyKeys.add(event.idempotencyKey());
+		}
+		
 		int posAndTxAsWell = eventlog.size()+1;
 		EventReference reference = EventReference.create(posAndTxAsWell, posAndTxAsWell);
 		StoredEvent storedEvent = event.positionAt(reference, LocalDateTime.now());

sliceworkz-eventstore-infra-postgres/src/main/java/org/sliceworkz/eventstore/infra/postgres/PostgresEventStorageImpl.java

@@ -28,6 +28,7 @@
 import java.sql.Statement;
 import java.sql.Timestamp;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
@@ -648,12 +649,12 @@ private void addEventQueryFiltering(StringBuilder sqlBuilder, List<Object> param
 	public List<StoredEvent> append(AppendCriteria appendCriteria, Optional<EventStreamId> streamId, List<EventToStore> events) {
 		// Build conditional insert with optimistic locking check
 		StringBuilder sqlBuilder = new StringBuilder();
-		sqlBuilder.append("INSERT INTO %sevents (event_id, stream_context, stream_purpose, event_type, event_data, event_erasable_data, event_tags) SELECT * FROM ( VALUES ".formatted(prefix));
+		sqlBuilder.append("INSERT INTO %sevents (event_id, idempotency_key, stream_context, stream_purpose, event_type, event_data, event_erasable_data, event_tags) SELECT * FROM ( VALUES ".formatted(prefix));
 		for ( int i = 0; i < events.size(); i++ ) {
 			if ( i > 0 ) {
 				sqlBuilder.append(", ");
 			}
-			sqlBuilder.append("(?::uuid, ?, ?, ?, ?::jsonb, ?::jsonb, ?)");
+			sqlBuilder.append("(?::uuid, ?, ?, ?, ?, ?::jsonb, ?::jsonb, ?)");
 		}
 		sqlBuilder.append(")");
 
@@ -668,6 +669,7 @@ public List<StoredEvent> append(AppendCriteria appendCriteria, Optional<EventStr
 
 			// Add to-be-appended-event parameters first
 			parameters.add(id.value());
+			parameters.add(event.idempotencyKey());
 			parameters.add(event.stream().context());
 			parameters.add(event.stream().purpose());
 			parameters.add(event.type().name());
@@ -764,7 +766,13 @@ WHERE NOT EXISTS (
 				return storedEvents;
 			} catch (SQLException e) {
 				writeConnection.rollback();
-				throw new EventStorageException("SQLException during append", e);
+				
+				// idempotency issue
+				if ( e.getMessage().contains("idempotency_key")) {
+					return Collections.emptyList();
+				} else {
+					throw new EventStorageException("SQLException during append", e);
+				}
 			}
 
 		} catch (SQLException e) {

sliceworkz-eventstore-infra-postgres/src/main/quickstart/quickstart.ddl.sql

@@ -40,6 +40,9 @@ CREATE TABLE events (
       -- Event identification
       event_id UUID NOT NULL UNIQUE,
 
+      -- Idempotency key
+      idempotency_key TEXT UNIQUE,
+
       -- Stream identification  
       stream_context TEXT NOT NULL,
       stream_purpose TEXT NOT NULL DEFAULT '',