Added streaming capabilities

samples/webApi/expressjs-with-esdb/tsup.config.ts

@@ -14,7 +14,7 @@ export default defineConfig({
   watch: env === 'development',
   target: 'esnext',
   outDir: 'dist', //env === 'production' ? 'dist' : 'lib',
-  entry: ['src/**/*.ts', '!src/**/*.spec.ts', '!src/**/*.internal.ts'], //include all files under src but not specs
+  entry: ['src/index.ts'],
   sourcemap: true,
   tsconfig: 'tsconfig.json', // workaround for https://github.com/egoist/tsup/issues/571#issuecomment-1760052931
 });

src/docs/snippets/gettingStarted/webApi/apiBDD.e2e.spec.ts

@@ -65,7 +65,7 @@ void describe('ShoppingCart E2E', () => {
         )
         .then([expectResponse(204)]));
 
-    void it('should return details', () =>
+    void it('returns details', () =>
       given(openedShoppingCartWithProduct)
         .when((request) =>
           request.get(`/clients/${clientId}/shopping-carts/current`).send(),

src/package.json

@@ -86,9 +86,11 @@
     "@types/express": "4.17.21",
     "@types/node": "^22.4.1",
     "@types/supertest": "6.0.2",
+    "web-streams-polyfill": "^4.0.0",
     "supertest": "7.0.0"
   },
   "workspaces": [
+    "packages/emmett-shims",
     "packages/emmett",
     "packages/emmett-postgresql",
     "packages/emmett-esdb",

src/packages/emmett-esdb/src/eventStore/eventstoreDBEventStore.ts

@@ -3,15 +3,17 @@ import {
   NO_CONCURRENCY_CHECK,
   STREAM_DOES_NOT_EXIST,
   STREAM_EXISTS,
+  globalStreamCaughtUp,
+  streamTransformations,
   type AggregateStreamOptions,
   type AggregateStreamResult,
   type AppendToStreamOptions,
   type AppendToStreamResult,
   type DefaultStreamVersionType,
   type Event,
-  type EventMetaDataOf,
   type EventStore,
   type ExpectedStreamVersion,
+  type GlobalStreamCaughtUp,
   type ReadEvent,
   type ReadEventMetadataWithGlobalPosition,
   type ReadStreamOptions,
@@ -25,9 +27,16 @@ import {
   StreamNotFoundError,
   WrongExpectedVersionError,
   jsonEvent,
+  type AllStreamResolvedEvent,
+  type AllStreamSubscription,
   type AppendExpectedRevision,
   type ReadStreamOptions as ESDBReadStreamOptions,
+  type JSONRecordedEvent,
 } from '@eventstore/db-client';
+import { WritableStream, type ReadableStream } from 'node:stream/web';
+import { Readable } from 'stream';
+
+const { map } = streamTransformations;
 
 const toEventStoreDBReadOptions = (
   options: ReadStreamOptions | undefined,
@@ -64,26 +73,13 @@ export const getEventStoreDBEventStore = (
         let state = initialState();
         let currentStreamVersion: bigint | undefined = undefined;
 
-        for await (const { event } of eventStore.readStream(
+        for await (const { event } of eventStore.readStream<EventType>(
           streamName,
           toEventStoreDBReadOptions(options.read),
         )) {
           if (!event) continue;
 
-          state = evolve(state, <
-            ReadEvent<EventType, ReadEventMetadataWithGlobalPosition>
-          >{
-            type: event.type,
-            data: event.data,
-            metadata: {
-              ...((event.metadata as EventMetaDataOf<EventType> | undefined) ??
-                {}),
-              eventId: event.id,
-              streamName: event.streamId,
-              streamPosition: event.revision,
-              globalPosition: event.position!.commit,
-            },
-          });
+          state = evolve(state, mapFromESDBEvent<EventType>(event));
           currentStreamVersion = event.revision;
         }
 
@@ -123,23 +119,12 @@ export const getEventStoreDBEventStore = (
       let currentStreamVersion: bigint | undefined = undefined;
 
       try {
-        for await (const { event } of eventStore.readStream(
+        for await (const { event } of eventStore.readStream<EventType>(
           streamName,
           toEventStoreDBReadOptions(options),
         )) {
           if (!event) continue;
-          events.push(<
-            ReadEvent<EventType, ReadEventMetadataWithGlobalPosition>
-          >{
-            type: event.type,
-            data: event.data,
-            metadata: {
-              eventId: event.id,
-              streamName: event.streamId,
-              streamPosition: event.revision,
-              globalPosition: event.position!.commit,
-            },
-          });
+          events.push(mapFromESDBEvent(event));
           currentStreamVersion = event.revision;
         }
         return currentStreamVersion
@@ -189,6 +174,25 @@ export const getEventStoreDBEventStore = (
         throw error;
       }
     },
+
+    //streamEvents: streamEvents(eventStore),
+  };
+};
+
+const mapFromESDBEvent = <EventType extends Event = Event>(
+  event: JSONRecordedEvent<EventType>,
+): ReadEvent<EventType, ReadEventMetadataWithGlobalPosition> => {
+  return <ReadEvent<EventType, ReadEventMetadataWithGlobalPosition>>{
+    type: event.type,
+    data: event.data,
+    metadata: {
+      ...((event.metadata as ReadEventMetadataWithGlobalPosition) ??
+        ({} as ReadEventMetadataWithGlobalPosition)),
+      eventId: event.id,
+      streamName: event.streamId,
+      streamPosition: event.revision,
+      globalPosition: event.position!.commit,
+    },
   };
 };
 
@@ -242,3 +246,62 @@ const assertExpectedVersionMatchesCurrent = (
   if (!matchesExpectedVersion(current, expected))
     throw new ExpectedVersionConflictError(current, expected);
 };
+
+// eslint-disable-next-line @typescript-eslint/no-unused-vars
+const convertToWebReadableStream = (
+  allStreamSubscription: AllStreamSubscription,
+): ReadableStream<AllStreamResolvedEvent | GlobalStreamCaughtUp> => {
+  // Validate the input type
+  if (!(allStreamSubscription instanceof Readable)) {
+    throw new Error('Provided stream is not a Node.js Readable stream.');
+  }
+
+  let globalPosition = 0n;
+
+  const stream = Readable.toWeb(
+    allStreamSubscription,
+  ) as ReadableStream<AllStreamResolvedEvent>;
+
+  const writable = new WritableStream<
+    AllStreamResolvedEvent | GlobalStreamCaughtUp
+  >();
+
+  allStreamSubscription.on('caughtUp', async () => {
+    console.log(globalPosition);
+    await writable.getWriter().write(globalStreamCaughtUp({ globalPosition }));
+  });
+
+  const transform = map<
+    AllStreamResolvedEvent,
+    AllStreamResolvedEvent | GlobalStreamCaughtUp
+  >((event) => {
+    if (event?.event?.position.commit)
+      globalPosition = event.event?.position.commit;
+
+    return event;
+  });
+
+  return stream.pipeThrough<AllStreamResolvedEvent | GlobalStreamCaughtUp>(
+    transform,
+  );
+};
+
+// const streamEvents = (eventStore: EventStoreDBClient) => () => {
+//   return restream<
+//     AllStreamResolvedEvent | GlobalSubscriptionEvent,
+//     | ReadEvent<Event, ReadEventMetadataWithGlobalPosition>
+//     | GlobalSubscriptionEvent
+//   >(
+//     (): ReadableStream<AllStreamResolvedEvent | GlobalSubscriptionEvent> =>
+//       convertToWebReadableStream(
+//         eventStore.subscribeToAll({
+//           fromPosition: START,
+//           filter: excludeSystemEvents(),
+//         }),
+//       ),
+//     (
+//       resolvedEvent: AllStreamResolvedEvent | GlobalSubscriptionEvent,
+//     ): ReadEvent<Event, ReadEventMetadataWithGlobalPosition> =>
+//       mapFromESDBEvent(resolvedEvent.event as JSONRecordedEvent<Event>),
+//   );
+// };

src/packages/emmett-esdb/tsup.config.ts

@@ -10,7 +10,6 @@ export default defineConfig({
   minify: true, //env === 'production',
   bundle: true, //env === 'production',
   skipNodeModulesBundle: true,
-  entryPoints: ['src/index.ts'],
   watch: env === 'development',
   target: 'esnext',
   outDir: 'dist', //env === 'production' ? 'dist' : 'lib',

src/packages/emmett-expressjs/tsup.config.ts

@@ -10,7 +10,6 @@ export default defineConfig({
   minify: true, //env === 'production',
   bundle: true, //env === 'production',
   skipNodeModulesBundle: true,
-  entryPoints: ['src/index.ts'],
   watch: env === 'development',
   target: 'esnext',
   outDir: 'dist', //env === 'production' ? 'dist' : 'lib',

src/packages/emmett-fastify/tsup.config.ts

@@ -10,7 +10,6 @@ export default defineConfig({
   minify: true, //env === 'production',
   bundle: true, //env === 'production',
   skipNodeModulesBundle: true,
-  entryPoints: ['src/index.ts'],
   watch: env === 'development',
   target: 'esnext',
   outDir: 'dist', //env === 'production' ? 'dist' : 'lib',

src/packages/emmett-shims/package.json

@@ -0,0 +1,55 @@
+{
+  "name": "@event-driven-io/emmett-shims",
+  "version": "0.19.1",
+  "type": "module",
+  "description": "Emmett - Event Sourcing development made simple",
+  "scripts": {
+    "build": "tsup",
+    "build:ts": "tsc",
+    "build:ts:watch": "tsc  --watch",
+    "test": "run-s test:unit test:int test:e2e",
+    "test:unit": "glob -c \"node --import tsx --test\" **/*.unit.spec.ts",
+    "test:int": "glob -c \"node --import tsx --test\" **/*.int.spec.ts",
+    "test:e2e": "glob -c \"node --import tsx --test\" **/*.e2e.spec.ts",
+    "test:watch": "node --import tsx --test --watch",
+    "test:unit:watch": "glob -c \"node --import tsx --test --watch\" **/*.unit.spec.ts",
+    "test:int:watch": "glob -c \"node --import tsx --test --watch\" **/*.int.spec.ts",
+    "test:e2e:watch": "glob -c \"node --import tsx --test --watch\" **/*.e2e.spec.ts"
+  },
+  "repository": {
+    "type": "git",
+    "url": "git+https://github.com/event-driven-io/emmett.git"
+  },
+  "keywords": [
+    "Event Sourcing"
+  ],
+  "author": "Oskar Dudycz",
+  "bugs": {
+    "url": "https://github.com/event-driven-io/emmett/issues"
+  },
+  "homepage": "https://event-driven-io.github.io/emmett/",
+  "exports": {
+    ".": {
+      "import": {
+        "types": "./dist/index.d.ts",
+        "default": "./dist/index.js"
+      },
+      "require": {
+        "types": "./dist/index.d.cts",
+        "default": "./dist/index.cjs"
+      }
+    }
+  },
+  "main": "./dist/index.cjs",
+  "module": "./dist/index.js",
+  "types": "./dist/index.d.ts",
+  "files": [
+    "dist"
+  ],
+  "peerDependencies": {
+    "web-streams-polyfill": "^4.0.0"
+  },
+  "devDependencies": {
+    "@types/node": "^20.11.30"
+  }
+}