Added hook to be able to detect if schema was created already when using event store session (e.g. in command handler)

Author: oskardudycz

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/tables.ts

@@ -18,7 +18,7 @@ export const streamsTableSQL = rawSql(
       PRIMARY KEY (stream_id, partition, is_archived)
   ) PARTITION BY LIST (partition);
    
-  CREATE UNIQUE INDEX idx_streams_unique 
+  CREATE UNIQUE INDEX IF NOT EXISTS idx_streams_unique 
   ON ${streamsTable.name}(stream_id, partition, is_archived) 
   INCLUDE (stream_position);`,
 );