Fix null reader validation and typo in OptimizingAppendListenerDecorator

CLAUDE.md

@@ -1009,7 +1009,7 @@ can pass all of it and still violate the boundary in production.
   function emits one `pg_notify` per distinct `(stream_context, stream_purpose)` in the transition table,
   carrying that stream's maximum over the total `(event_tx, event_position)` order. It was `FOR EACH ROW`,
   which meant a 1000-event append queued 1000 notifications and an import chunk queued 5000 — all but one
-  per stream discarded by `OptimizingApendListenerDecorator` after being built as JSON, written to the
+  per stream discarded by `OptimizingAppendListenerDecorator` after being built as JSON, written to the
   cluster-wide async queue, sent over the wire, parsed by Jackson and fanned out to every listener.
   Measured on the PG16 floor, a 100k-row insert: the notification count falls from 100.000 to exactly 1,
   and trigger time roughly halves — 1230ms to 460ms on one run, 808ms to 369ms on another (the absolute
@@ -1050,7 +1050,9 @@ can pass all of it and still violate the boundary in production.
   every listener has consumed, so a stalled listener does make usage accumulate monotonically across
   transactions — but from a base low enough that the amplification was a throughput and latency problem,
   not a correctness-of-operation one.
-- `stream_purpose` defaults to `'default'` in the DDL, matching `EventStreamId.DEFAULT_PURPOSE`. On a database created before this alignment (default was `''`), operators doing raw SQL inserts should run `ALTER TABLE <prefix>events ALTER COLUMN stream_purpose SET DEFAULT 'default';` — no data migration is needed since all events written through the library bind the purpose explicitly
+- `stream_purpose` defaults to `'default'` in the DDL, matching `EventStreamId.DEFAULT_PURPOSE` — a public
+  constant, so an interop layer can bind the same value the library does rather than copy the literal out
+  of this file. On a database created before this alignment (default was `''`), operators doing raw SQL inserts should run `ALTER TABLE <prefix>events ALTER COLUMN stream_purpose SET DEFAULT 'default';` — no data migration is needed since all events written through the library bind the purpose explicitly
 - **Idempotency keys are scoped per event stream (context + purpose), not per storage/table.** Uniqueness is enforced by the partial unique index `idx_events_stream_idempotency` on `(stream_context, stream_purpose, idempotency_key) WHERE idempotency_key IS NOT NULL` (schema validation requires it), so the same key used on two unrelated streams does not collide and dedup behaviour does not depend on how storage instances / prefixes are wired at runtime. The `idempotency_key` is persisted and surfaced on `StoredEvent` when reading (it is not exposed on the public `Event` record). A duplicate append is still silently ignored (returns an empty result). On a database created before this change (when `idempotency_key` had a table-wide `UNIQUE`), migrate with: `ALTER TABLE <prefix>events DROP CONSTRAINT <prefix>events_idempotency_key_key; CREATE UNIQUE INDEX <prefix>idx_events_stream_idempotency ON <prefix>events (stream_context, stream_purpose, idempotency_key) WHERE idempotency_key IS NOT NULL;` — no data migration is needed
   - **The duplicate is recognised by the index the server names, never by the message text.** Both the
     append and the import path go through `isIdempotencyKeyViolation`, which pairs SQLSTATE 23505 with

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventSource.java

@@ -347,9 +347,10 @@ default Stream<Event<DOMAIN_EVENT_TYPE>> query ( EventQuery query ) {
 	 * maintain only one bookmark per stream. Tags can be attached to bookmarks for
 	 * additional metadata (e.g., processing status, reader state).
 	 *
-	 * @param reader the unique name/identifier of the reader placing the bookmark
+	 * @param reader the unique name/identifier of the reader placing the bookmark; must not be null
 	 * @param reference the event reference to bookmark (the last processed event)
 	 * @param tags optional tags to attach to the bookmark for metadata
+	 * @throws NullPointerException if {@code reader} is null
 	 */
 	void placeBookmark ( String reader, EventReference reference, Tags tags );
 
@@ -359,8 +360,9 @@ default Stream<Event<DOMAIN_EVENT_TYPE>> query ( EventQuery query ) {
 	 * Returns the last bookmarked position for the specified reader, allowing
 	 * the reader to resume processing from where it left off.
 	 *
-	 * @param reader the unique name/identifier of the reader
+	 * @param reader the unique name/identifier of the reader; must not be null
 	 * @return an Optional containing the bookmarked EventReference if found, empty if no bookmark exists
+	 * @throws NullPointerException if {@code reader} is null
 	 */
 	Optional<EventReference> getBookmark ( String reader );
 
@@ -394,8 +396,9 @@ default Stream<Event<DOMAIN_EVENT_TYPE>> query ( EventQuery query ) {
 	 * stream.query(EventQuery.matchAll()).forEach(this::processEvent);
 	 * }</pre>
 	 *
-	 * @param reader the unique name/identifier of the reader whose bookmark should be removed
+	 * @param reader the unique name/identifier of the reader whose bookmark should be removed; must not be null
 	 * @return an Optional containing the previous bookmarked EventReference if one existed, empty otherwise
+	 * @throws NullPointerException if {@code reader} is null
 	 */
 	Optional<EventReference> removeBookmark ( String reader );
 

sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventStreamId.java

@@ -72,7 +72,21 @@
  */
 public record EventStreamId ( String context, String purpose ) {
 
-	private static final String DEFAULT_PURPOSE = "default";
+	/**
+	 * The purpose given to a stream created with {@link #forContext(String)} or {@link #defaultPurpose()}:
+	 * {@code "default"}.
+	 * <p>
+	 * Public because it is a storage-level value, not only a Java one. It is what the library binds into
+	 * the {@code stream_purpose} column for a context that never sets a purpose, and what the PostgreSQL
+	 * DDL carries as that column's default — so anyone writing rows by hand, building an interop layer,
+	 * or querying the events table directly needs the exact string this library agrees on rather than a
+	 * literal copied from documentation.
+	 * <p>
+	 * Note that this is a compile-time constant, so a reference to it is inlined into the calling class.
+	 * Changing it would therefore not be a drop-in replacement — but it is stored data (see the
+	 * {@code stream_purpose} notes in the project documentation), so it is not going to change.
+	 */
+	public static final String DEFAULT_PURPOSE = "default";
 
 	/**
 	 * Creates an EventStreamId for a specific context with the default purpose.
@@ -242,14 +256,21 @@ public boolean canRead ( EventStreamId actualStreamId ) {
 	 * <p>
 	 * This is useful for scenarios where a general stream (e.g., "customer#anyPurpose") needs to
 	 * append events to specific instances (e.g., "customer#123").
+	 * <p>
+	 * A wildcard concretizes nothing: {@code forContext("customer").anyPurpose()} does <em>not</em>
+	 * concretize itself, or any other wildcard-purpose stream, because it supplies no purpose to fill
+	 * the other's wildcard in with.
 	 *
 	 * @param otherStreamId the stream ID to check if this stream concretizes it
 	 * @return true if this stream ID concretizes the other stream ID, false otherwise
 	 * @see #canAppendTo(EventStreamId)
 	 */
 	public boolean concretizes ( EventStreamId otherStreamId ) {
 		// if the other stream is of the type "<businessObject>#<anyPurpose>" and this is "<businessObject>#<id>"
-		return otherStreamId.isAnyPurpose() && !otherStreamId.isAnyContext() && otherStreamId.context().equals(this.context);
+		return otherStreamId.isAnyPurpose()
+				&& !otherStreamId.isAnyContext()
+				&& !this.isAnyPurpose()                             // a wildcard fills in nothing
+				&& otherStreamId.context().equals(this.context);
 	}
 
 	/**

sliceworkz-eventstore-api/src/test/java/org/sliceworkz/eventstore/stream/EventStreamIdTest.java

@@ -72,6 +72,29 @@ void testConcretizes ( ) {
 		assertFalse(EventStreamId.forContext("customer").withPurpose("123").concretizes(EventStreamId.anyContext().withPurpose("123")));
 	}
 
+	/**
+	 * A wildcard supplies no purpose, so it concretizes nothing — not another wildcard-purpose stream
+	 * in the same context, and not itself. The javadoc has always said so ("This stream has a specific
+	 * purpose (not a wildcard)"); the implementation used to check only the other three conditions.
+	 */
+	@Test
+	void testWildcardPurposeConcretizesNothing ( ) {
+		EventStreamId customerAnyPurpose = EventStreamId.forContext("customer").anyPurpose();
+		EventStreamId otherCustomerAnyPurpose = EventStreamId.forContext("customer").anyPurpose();
+		EventStreamId supplierAnyPurpose = EventStreamId.forContext("supplier").anyPurpose();
+
+		assertFalse(customerAnyPurpose.concretizes(otherCustomerAnyPurpose));
+		assertFalse(customerAnyPurpose.concretizes(customerAnyPurpose));
+		assertFalse(customerAnyPurpose.concretizes(supplierAnyPurpose));
+		assertFalse(EventStreamId.anyContext().concretizes(customerAnyPurpose));
+
+		// canAppendTo is unaffected: two equal wildcard streams already matched on equals(), and the
+		// append path rejects a read-only target regardless -- which is why this was never reachable
+		// through append().
+		assertTrue(customerAnyPurpose.canAppendTo(otherCustomerAnyPurpose));
+		assertTrue(customerAnyPurpose.isReadOnly());
+	}
+
 	@Test
 	void testToString ( ) {
 		EventStreamId i = EventStreamId.anyContext();

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

@@ -19,6 +19,7 @@
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -607,7 +608,7 @@ public void close ( ) {
 		@Override
 		public void subscribe(EventStreamEventuallyConsistentAppendListener eventuallyConsistentSubscriber) {
 			checkStoreNotClosed();
-			this.eventuallyConsistentSubscribers.add(new OptimizingApendListenerDecorator(eventuallyConsistentSubscriber));
+			this.eventuallyConsistentSubscribers.add(new OptimizingAppendListenerDecorator(eventuallyConsistentSubscriber));
 			subscribeToStorage();
 		}
 
@@ -797,11 +798,11 @@ public void notify(BookmarkPlacedNotification bookmarkPlaced) {
 			if ( closed.get() ) {
 				return; // see notify(AppendsToEventStoreNotification)
 			}
-			LOGGER.debug("Must asynchronously notify {} eventually consistent bookmark listeners on {} of update for {} to {}", eventuallyConsistentSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark());
+			LOGGER.debug("Must asynchronously notify {} eventually consistent bookmark listeners on {} of update for {} to {}", bookmarkSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark());
 
 			// schedule for execution on different thread to notify/interrupt any waiting eventual consistent processors
 			submitOrDropIfClosed(executorServiceForBookmarkUpdates, ( ) -> {
-					LOGGER.debug("Notifying {} eventually consistent bookmark listeners on {} of update for {} to {}", eventuallyConsistentSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark());
+					LOGGER.debug("Notifying {} eventually consistent bookmark listeners on {} of update for {} to {}", bookmarkSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark());
 					bookmarkSubscribers.stream().forEach(s->s.bookmarkUpdated(bookmarkPlaced.reader(), bookmarkPlaced.bookmark()));
 			});
 		}
@@ -826,6 +827,7 @@ private void submitOrDropIfClosed ( ExecutorService executorService, Runnable no
 		@Override
 		public void placeBookmark(String reader, EventReference reference, Tags tags) {
 			checkStoreNotClosed();
+			requireReader(reader);
 			meterBookmarkPlace.increment();
 			eventStorage.bookmark(reader, reference, tags);
 		}
@@ -843,8 +845,29 @@ public Optional<EventReference> removeBookmark(String reader) {
 		@Override
 		public Optional<EventReference> getBookmark(String reader) {
 			checkStoreNotClosed();
+			requireReader(reader);
 			meterBookmarkGet.increment();
-			return eventStorage.getBookmark(reader.toString());
+			return eventStorage.getBookmark(reader);
+		}
+
+		/**
+		 * Rejects a null reader name here, rather than letting one reach a backend.
+		 * <p>
+		 * A reader name is the whole identity of a bookmark, so a null one is a programming error in
+		 * every case. What it is not is a defined one: the backends disagree about it. The in-memory
+		 * stores keep bookmarks in a {@link java.util.HashMap}, which accepts a null key happily and
+		 * stores a bookmark nobody can name; PostgreSQL has {@code reader TEXT PRIMARY KEY}, so a null
+		 * reaches the database and comes back as a not-null violation on a place, and as a silent
+		 * no-op on a remove and an empty {@code Optional} on a get, because {@code WHERE reader = NULL}
+		 * matches nothing.
+		 * <p>
+		 * Two of those paths used to be guarded accidentally, by a {@code reader.toString()} on a value
+		 * already typed {@code String}. Removing those calls (they convert nothing) would have taken the
+		 * guard with them, so the check is made explicit and uniform instead — one place, same answer on
+		 * every backend, and it names the parameter.
+		 */
+		private void requireReader ( String reader ) {
+			Objects.requireNonNull(reader, "reader must not be null");
 		}
 
 		@Override

sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecorator.java → sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecorator.java

@@ -43,10 +43,15 @@
  * The optimization leverages the return value of {@link EventStreamEventuallyConsistentAppendListener#eventsAppended(EventReference)}
  * to track what the delegate listener has actually processed, allowing it to skip notifications
  * for event references already handled.
+ * <p>
+ * <strong>Internal.</strong> Every listener passed to {@code subscribe(...)} is wrapped in one of these
+ * by the store itself, so there is no reason for a caller to name this class — wrapping a listener
+ * before subscribing it only gets it wrapped twice. It lives in the implementation package the
+ * ServiceLoader exists to hide, and carries no compatibility promise.
  *
  * @see EventStreamEventuallyConsistentAppendListener
  */
-public class OptimizingApendListenerDecorator implements EventStreamEventuallyConsistentAppendListener {
+public class OptimizingAppendListenerDecorator implements EventStreamEventuallyConsistentAppendListener {
     private final EventStreamEventuallyConsistentAppendListener delegate;
     private final ReentrantLock lock;
     private final AtomicReference<EventReference> lastNotifiedReference;
@@ -58,7 +63,7 @@ public class OptimizingApendListenerDecorator implements EventStreamEventuallyCo
      *
      * @param delegate the listener to decorate with optimization logic; must not be null
      */
-    public OptimizingApendListenerDecorator(EventStreamEventuallyConsistentAppendListener delegate) {
+    public OptimizingAppendListenerDecorator(EventStreamEventuallyConsistentAppendListener delegate) {
         this.delegate = delegate;
         this.lock = new ReentrantLock();
         this.lastNotifiedReference = new AtomicReference<>();

sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecoratorTest.java → sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecoratorTest.java

@@ -32,7 +32,7 @@
 import org.sliceworkz.eventstore.events.EventReference;
 import org.sliceworkz.eventstore.stream.EventStreamEventuallyConsistentAppendListener;
 
-class OptimizingApendListenerDecoratorTest {
+class OptimizingAppendListenerDecoratorTest {
 
 	/**
 	 * Enough pairs to catch a lost notification; the window is narrow, so the count is what makes this
@@ -59,7 +59,7 @@ void testLaterOfTwoSimultaneousNotificationsIsNeverLost ( ) throws InterruptedEx
 		try ( ExecutorService notifiers = Executors.newVirtualThreadPerTaskExecutor() ) {
 			for ( int pair = 0; pair < RACING_PAIRS; pair++ ) {
 				AtomicReference<EventReference> seenByDelegate = new AtomicReference<>();
-				OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator(
+				OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator(
 						reference -> {
 							seenByDelegate.set(reference);
 							return reference;
@@ -105,7 +105,7 @@ void testNotificationAlreadySeenByTheListenerIsSkipped ( ) {
 			deliveries.incrementAndGet();
 			return reference;
 		};
-		OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator(counting);
+		OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator(counting);
 
 		EventReference reference = reference(1);
 		decorator.eventsAppended(reference);
@@ -125,7 +125,7 @@ void testSlowListenerDoesNotBlockTheNotifyingThread ( ) throws Exception {
 		CountDownLatch releaseListener = new CountDownLatch(1);
 		AtomicReference<EventReference> seenByDelegate = new AtomicReference<>();
 
-		OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator(reference -> {
+		OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator(reference -> {
 			listenerEntered.countDown();
 			try {
 				releaseListener.await(5, TimeUnit.SECONDS);

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

@@ -2215,7 +2215,7 @@ ON CONFLICT (reader)
 					}
 
 					try ( PreparedStatement stmt = writeConnection.prepareStatement(sql) ) {
-						stmt.setString(1, reader.toString());
+						stmt.setString(1, reader);
 						stmt.setLong(2, eventReference == null?0:eventReference.position());
 						stmt.setString(3, eventReference == null?"0":Long.toUnsignedString(eventReference.tx()));
 						stmt.setString(4, eventReference==null?null:eventReference.id().value());
@@ -2258,8 +2258,8 @@ public void removeBookmark(String reader ) {
 				writeConnection.setAutoCommit(false);
 
 				try ( PreparedStatement stmt = writeConnection.prepareStatement(sql) ) {
-					stmt.setString(1, reader.toString());
-					
+					stmt.setString(1, reader);
+
 					stmt.executeUpdate();
 					writeConnection.commit();
 				}