Merge remote-tracking branch 'upstream/main' into feat/subscriptions

Author: arturwojnar

src/packages/emmett-postgresql/src/eventStore/postgreSQLEventStore.e2e.spec.ts

@@ -11,7 +11,7 @@ import {
   PostgreSqlContainer,
   StartedPostgreSqlContainer,
 } from '@testcontainers/postgresql';
-import { after, before, describe, it } from 'node:test';
+import { after, afterEach, before, beforeEach, describe, it } from 'node:test';
 import { v4 as uuid } from 'uuid';
 import {
   getPostgreSQLEventStore,
@@ -26,38 +26,97 @@ void describe('EventStoreDBEventStore', () => {
   let connectionString: string;
   let pongo: PongoClient;
 
+  const productItem: PricedProductItem = {
+    productId: '123',
+    quantity: 10,
+    price: 3,
+  };
+  let clientId: string;
+  let shoppingCartId: string;
+  let schemaHookCreationHookCalls = 0;
+
   before(async () => {
     postgres = await new PostgreSqlContainer().start();
     connectionString = postgres.getConnectionUri();
+    pongo = pongoClient(connectionString);
+  });
+
+  beforeEach(() => {
     eventStore = getPostgreSQLEventStore(connectionString, {
       projections: projections.inline([
         shoppingCartShortInfoProjection,
         customProjection,
       ]),
+      hooks: {
+        onAfterSchemaCreated: () => {
+          schemaHookCreationHookCalls++;
+        },
+      },
     });
-    pongo = pongoClient(connectionString);
-    return eventStore;
+    clientId = uuid();
+    shoppingCartId = `shopping_cart-${clientId}`;
+    schemaHookCreationHookCalls = 0;
   });
 
-  after(async () => {
+  afterEach(async () => {
     try {
       await eventStore.close();
+    } catch (error) {
+      console.log(error);
+    }
+  });
+
+  after(async () => {
+    try {
       await pongo.close();
       await postgres.stop();
     } catch (error) {
       console.log(error);
     }
   });
 
+  void it('should create schema only once ', async () => {
+    await eventStore.appendToStream<ShoppingCartEvent>(shoppingCartId, [
+      {
+        type: 'ProductItemAdded',
+        data: { productItem },
+        metadata: { clientId },
+      },
+    ]);
+    await eventStore.appendToStream<ShoppingCartEvent>(shoppingCartId, [
+      {
+        type: 'ProductItemAdded',
+        data: { productItem },
+        metadata: { clientId },
+      },
+    ]);
+
+    assertEqual(1, schemaHookCreationHookCalls);
+  });
+
+  void it('should create schema only once with session', async () => {
+    await eventStore.appendToStream<ShoppingCartEvent>(shoppingCartId, [
+      {
+        type: 'ProductItemAdded',
+        data: { productItem },
+        metadata: { clientId },
+      },
+    ]);
+    await eventStore.withSession(({ eventStore: session }) =>
+      session.appendToStream<ShoppingCartEvent>(shoppingCartId, [
+        {
+          type: 'ProductItemAdded',
+          data: { productItem },
+          metadata: { clientId },
+        },
+      ]),
+    );
+
+    assertEqual(1, schemaHookCreationHookCalls);
+  });
+
   void it('should append events correctly using appendEvent function', async () => {
-    const productItem: PricedProductItem = {
-      productId: '123',
-      quantity: 10,
-      price: 3,
-    };
     const discount = 10;
-    const clientId = uuid();
-    const shoppingCartId = `shopping_cart-${clientId}`;
     handledEventsInCustomProjection = [];
 
     await eventStore.appendToStream<ShoppingCartEvent>(shoppingCartId, [

src/packages/emmett-postgresql/src/eventStore/postgreSQLEventStore.ts

@@ -143,6 +143,12 @@ export type PostgresEventStoreOptions = {
   >[];
   schema?: { autoMigration?: MigrationStyle };
   connectionOptions?: PostgresEventStoreConnectionOptions;
+  hooks?: {
+    /**
+     * This hook will be called **AFTER** event store schema was created
+     */
+    onAfterSchemaCreated?: () => Promise<void> | void;
+  };
 };
 
 export const defaultPostgreSQLOptions: PostgresEventStoreOptions = {
@@ -161,7 +167,7 @@ export const getPostgreSQLEventStore = (
     ...(options.connectionOptions ? options.connectionOptions : {}),
   };
   const pool = 'dumbo' in poolOptions ? poolOptions.dumbo : dumbo(poolOptions);
-  let migrateSchema: Promise<void>;
+  let migrateSchema: Promise<void> | undefined = undefined;
 
   const autoGenerateSchema =
     options.schema?.autoMigration === undefined ||
@@ -171,7 +177,11 @@ export const getPostgreSQLEventStore = (
     if (!autoGenerateSchema) return Promise.resolve();
 
     if (!migrateSchema) {
-      migrateSchema = createEventStoreSchema(pool);
+      migrateSchema = createEventStoreSchema(pool).then(async () => {
+        if (options.hooks?.onAfterSchemaCreated) {
+          await options.hooks.onAfterSchemaCreated();
+        }
+      });
     }
     return migrateSchema;
   };
@@ -301,11 +311,28 @@ export const getPostgreSQLEventStore = (
       callback: (session: EventStoreSession<PostgresEventStore>) => Promise<T>,
     ): Promise<T> {
       return await pool.withConnection(async (connection) => {
+        const autoMigration: MigrationStyle = migrateSchema
+          ? 'None'
+          : (options.schema?.autoMigration ?? 'CreateOrUpdate');
+
         const storeOptions: PostgresEventStoreOptions = {
           ...options,
           connectionOptions: {
             connection,
           },
+          schema: {
+            autoMigration,
+          },
+          hooks: {
+            ...(options.hooks ?? {}),
+            onAfterSchemaCreated: async () => {
+              migrateSchema = Promise.resolve();
+
+              if (options.hooks?.onAfterSchemaCreated) {
+                await options.hooks.onAfterSchemaCreated();
+              }
+            },
+          },
         };
 
         const eventStore = getPostgreSQLEventStore(

src/packages/emmett-postgresql/src/eventStore/schema/appendToStream.ts

@@ -48,9 +48,15 @@ export const appendToStreamSQL = rawSql(
       v_transaction_id := pg_current_xact_id();
 
       IF v_expected_stream_position IS NULL THEN
-          SELECT COALESCE(max(stream_position), 0) INTO v_expected_stream_position
-          FROM ${streamsTable.name}
-          WHERE stream_id = v_stream_id AND partition = v_partition;
+          SELECT COALESCE(
+            (SELECT stream_position 
+            FROM ${streamsTable.name}
+            WHERE stream_id = v_stream_id 
+              AND partition = v_partition 
+              AND is_archived = FALSE
+            LIMIT 1), 
+            0
+        ) INTO v_expected_stream_position;
       END IF;
 
       v_next_stream_position := v_expected_stream_position + array_upper(v_messages_data, 1);

src/packages/emmett-postgresql/src/eventStore/schema/tables.ts

@@ -15,9 +15,12 @@ export const streamsTableSQL = rawSql(
       stream_type       TEXT                      NOT NULL,
       stream_metadata   JSONB                     NOT NULL,
       is_archived       BOOLEAN                   NOT NULL DEFAULT FALSE,
-      PRIMARY KEY (stream_id, stream_position, partition, is_archived),
-      UNIQUE (stream_id, partition, is_archived)
-  ) PARTITION BY LIST (partition);`,
+      PRIMARY KEY (stream_id, partition, is_archived)
+  ) PARTITION BY LIST (partition);
+   
+  CREATE UNIQUE INDEX IF NOT EXISTS idx_streams_unique 
+  ON ${streamsTable.name}(stream_id, partition, is_archived) 
+  INCLUDE (stream_position);`,
 );
 
 export const messagesTableSQL = rawSql(

src/packages/emmett-sqlite/src/eventStore/schema/tables.ts

@@ -16,7 +16,7 @@ export const streamsTableSQL = sql(
       stream_type       TEXT                      NOT NULL,
       stream_metadata   JSONB                     NOT NULL,
       is_archived       BOOLEAN                   NOT NULL DEFAULT FALSE,
-      PRIMARY KEY (stream_id, stream_position, partition, is_archived),
+      PRIMARY KEY (stream_id, partition, is_archived),
       UNIQUE (stream_id, partition, is_archived)
   );`,
 );