fix(store-inmemory): deliver events appended before a new subscription runs

.changeset/inmemory-subscribe-gap.md

@@ -0,0 +1,7 @@
+---
+'@codeforbreakfast/eventsourcing-store-inmemory': patch
+---
+
+Fix lost events on new subscriptions. An event appended after `subscribe` or `subscribeAll` returned, but before the returned stream started running, never reached the subscriber. A subscription now delivers every event appended after the call returns, whenever you start running its stream.
+
+`InMemoryStore.get`, `getAll` and `getAllLiveOnly` get the same fix.

packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts

@@ -1,4 +1,14 @@
-import { Chunk, Effect, HashMap, Option, PubSub, Stream, SynchronizedRef, pipe } from 'effect';
+import {
+  Chunk,
+  Effect,
+  HashMap,
+  Option,
+  PubSub,
+  Queue,
+  Stream,
+  SynchronizedRef,
+  pipe,
+} from 'effect';
 import {
   EventStreamId,
   EventStreamPosition,
@@ -214,8 +224,40 @@ export interface InMemoryStore<V = never> {
   >;
 }
 
-const createLiveEventStream = <V>(eventStream: EventStream<V>): Stream.Stream<V, never, never> =>
-  pipe(eventStream.events, Stream.fromChunk, Stream.concat(Stream.fromPubSub(eventStream.pubsub)));
+const snapshotThenLive =
+  <V>(eventStream: EventStream<V>, offset: number) =>
+  (liveEvents: Queue.Dequeue<V>): Stream.Stream<V, never, never> =>
+    pipe(
+      eventStream.events,
+      Chunk.drop(offset),
+      Stream.fromChunk,
+      Stream.concat(Stream.fromQueue(liveEvents))
+    );
+
+const subscribeFromOffset =
+  (offset: number) =>
+  <V>(eventStream: EventStream<V>) =>
+    pipe(eventStream.pubsub, PubSub.subscribe, Effect.map(snapshotThenLive(eventStream, offset)));
+
+// Append publishes while it holds this lock, so taking the snapshot and subscribing under the
+// same lock leaves no gap between them. Doing both when the stream runs, not when subscribe
+// returns, catches events appended in the meantime without holding a subscription for a stream
+// that is never run.
+const liveStreamFromOffset = <V, A>(
+  value: SynchronizedRef.SynchronizedRef<Value<V>>,
+  selectEventStream: (current: Value<V>) => Effect.Effect<EventStream<A>, never, never>,
+  offset: number
+): Stream.Stream<A, never, never> =>
+  Stream.unwrapScoped(
+    SynchronizedRef.modifyEffect(value, (current) =>
+      pipe(
+        current,
+        selectEventStream,
+        Effect.flatMap(subscribeFromOffset(offset)),
+        Effect.map((stream) => [stream, current] as const)
+      )
+    )
+  );
 
 const createHistoricalEventStream = <V>(
   eventStream: EventStream<V>
@@ -237,12 +279,6 @@ const findEventStreamOrDie = <V>(
     })
   );
 
-const getEventStreamAndCreateLive = <V>(
-  eventStreamsById: HashMap.HashMap<EventStreamId, EventStream<V>>,
-  streamId: EventStreamId
-): Effect.Effect<Stream.Stream<V, never, never>, never, never> =>
-  pipe(findEventStreamOrDie(eventStreamsById, streamId), Effect.map(createLiveEventStream));
-
 const getEventStreamAndCreateHistorical = <V>(
   eventStreamsById: HashMap.HashMap<EventStreamId, EventStream<V>>,
   streamId: EventStreamId
@@ -266,9 +302,13 @@ const getLiveStreamForId =
   (value: SynchronizedRef.SynchronizedRef<Value<V>>) =>
     pipe(
       value,
-      SynchronizedRef.updateAndGetEffect(ensureEventStream(streamId)),
-      Effect.flatMap(({ eventStreamsById }) =>
-        getEventStreamAndCreateLive(eventStreamsById, streamId)
+      SynchronizedRef.updateEffect(ensureEventStream(streamId)),
+      Effect.as(
+        liveStreamFromOffset(
+          value,
+          ({ eventStreamsById }) => findEventStreamOrDie(eventStreamsById, streamId),
+          0
+        )
       )
     );
 
@@ -283,22 +323,22 @@ const getHistoricalStreamForId =
       )
     );
 
+const selectAllEventsStream = <V>({ allEventsStream }: Value<V>) => Effect.succeed(allEventsStream);
+
 const getAllEventsStream = <V>(
   value: SynchronizedRef.SynchronizedRef<Value<V>>
 ): Effect.Effect<Stream.Stream<StreamEvent<V>, never, never>, never, never> =>
-  pipe(
-    value,
-    SynchronizedRef.get,
-    Effect.map(({ allEventsStream }) => createLiveEventStream(allEventsStream))
-  );
+  Effect.succeed(liveStreamFromOffset(value, selectAllEventsStream, 0));
 
 const getAllEventsLiveOnlyStream = <V>(
   value: SynchronizedRef.SynchronizedRef<Value<V>>
 ): Effect.Effect<Stream.Stream<StreamEvent<V>, never, never>, never, never> =>
   pipe(
     value,
     SynchronizedRef.get,
-    Effect.map(({ allEventsStream }) => Stream.fromPubSub(allEventsStream.pubsub))
+    Effect.map(({ allEventsStream }) =>
+      liveStreamFromOffset(value, selectAllEventsStream, allEventsStream.events.length)
+    )
   );
 
 const appendForStore =
@@ -326,13 +366,11 @@ export const make = <V>() =>
         allEventsStream,
       })
     ),
-    Effect.map(
-      (value: SynchronizedRef.SynchronizedRef<Value<V>>): InMemoryStore<V> => ({
-        append: appendForStore(value),
-        get: getForStore(value),
-        getHistorical: getHistoricalForStore(value),
-        getAll: () => getAllEventsStream(value),
-        getAllLiveOnly: () => getAllEventsLiveOnlyStream(value),
-      })
-    )
+    Effect.map((value: SynchronizedRef.SynchronizedRef<Value<V>>): InMemoryStore<V> => ({
+      append: appendForStore(value),
+      get: getForStore(value),
+      getHistorical: getHistoricalForStore(value),
+      getAll: () => getAllEventsStream(value),
+      getAllLiveOnly: () => getAllEventsLiveOnlyStream(value),
+    }))
   );

packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts

@@ -1,10 +1,12 @@
-import { Effect, Layer, Schema, pipe } from 'effect';
+import { Chunk, Effect, Layer, Schema, Stream, pipe } from 'effect';
+import { describe, expect, it } from 'bun:test';
 import { silentLogger } from '@codeforbreakfast/bun-test-effect';
 import {
   runEventStoreTestSuite,
   FooEventStore,
   encodedEventStore,
   EventStore,
+  EventStreamId,
 } from '@codeforbreakfast/eventsourcing-store';
 import { subscribeAllContract } from '@codeforbreakfast/eventsourcing-testing-contracts';
 import { makeInMemoryEventStore } from './index';
@@ -35,10 +37,64 @@
   EventStore<string>
 >() {}
 
-subscribeAllContract(
-  'InMemoryEventStore',
-  Layer.provide(
-    Layer.effect(StringEventStore, Effect.flatMap(make<string>(), makeInMemoryEventStore)),
-    silentLogger
-  )
+const StringEventStoreLayer = Layer.provide(
+  Layer.effect(StringEventStore, Effect.flatMap(make<string>(), makeInMemoryEventStore)),
+  silentLogger
 );
+
+subscribeAllContract('InMemoryEventStore', StringEventStoreLayer);
+
+const decodeStreamId = Schema.decodeSync(EventStreamId);
+const streamId = decodeStreamId('immediate-stream');
+
+const appendEvent = (store: EventStore<string>) =>
+  pipe(
+    ['event-after-subscribe'],
+    Stream.fromIterable,
+    Stream.run(store.append({ streamId, eventNumber: 0 }))
+  );
+
+const collectFirst = <A>(stream: Stream.Stream<A, unknown, never>) =>
+  pipe(stream, Stream.take(1), Stream.runCollect);
+
+const appendThenCollectFirst =
+  (store: EventStore<string>) =>
+  <A>(stream: Stream.Stream<A, unknown, never>) =>
+    pipe(store, appendEvent, Effect.andThen(collectFirst(stream)), Effect.timeout('1 second'));
+
+const runPerStreamTest = (store: EventStore<string>) =>
+  pipe(
+    { streamId, eventNumber: 0 },
+    store.subscribe,
+    Effect.flatMap(appendThenCollectFirst(store)),
+    Effect.map((events) => {
+      expect(Array.from(events)).toEqual(['event-after-subscribe']);
+    })
+  );
+
+const runAllEventsTest = (store: EventStore<string>) =>
+  pipe(
+    store.subscribeAll(),
+    Effect.flatMap(appendThenCollectFirst(store)),
+    Effect.map((events) => {
+      expect(Array.from(Chunk.map(events, (e) => e.event))).toEqual(['event-after-subscribe']);
+    })
+  );
+
+describe('Subscriptions are live as soon as they return', () => {
+  it('should deliver an event appended after subscribe returns but before the stream runs', () =>
+    pipe(
+      StringEventStore,
+      Effect.flatMap(runPerStreamTest),
+      Effect.provide(StringEventStoreLayer),
+      Effect.runPromise
+    ));
+
+  it('should deliver an event appended after subscribeAll returns but before the stream runs', () =>
+    pipe(
+      StringEventStore,
+      Effect.flatMap(runAllEventsTest),
+      Effect.provide(StringEventStoreLayer),
+      Effect.runPromise
+    ));
+});