feat: add transaction support to postgres provider

adapters/postgres/database.ts

@@ -1,4 +1,4 @@
-import postgres, { type Sql } from "postgres";
+import postgres, { type Sql, TransactionSql } from "postgres";
 
 import { PostgresConnection } from "./connection.ts";
 
@@ -34,3 +34,7 @@ export class PostgresDatabase {
 export type DatabaseAccessor = {
   sql: Sql;
 };
+
+export type Options = {
+  tx?: TransactionSql;
+};

adapters/postgres/providers/event.ts

@@ -3,7 +3,7 @@ import type { Helper } from "postgres";
 import type { EventRecord } from "../../../libraries/event.ts";
 import type { EventsProvider } from "../../../types/adapter.ts";
 import type { EventReadOptions } from "../../../types/query.ts";
-import type { PostgresDatabase } from "../database.ts";
+import type { Options, PostgresDatabase } from "../database.ts";
 
 type PGEventRecord = Omit<EventRecord, "data" | "meta"> & { data: string; meta: string };
 
@@ -25,8 +25,8 @@ export class PostgresEventsProvider implements EventsProvider {
    *
    * @param record - Event record to insert.
    */
-  async insert(record: EventRecord): Promise<void> {
+  async insert(record: EventRecord, { tx }: Options = {}): Promise<void> {
-    await this.db.sql`INSERT INTO ${this.table} ${this.db.sql(this.#toDriver(record))}`.catch((error) => {
+    await (tx ?? this.db.sql)`INSERT INTO ${this.table} ${this.db.sql(this.#toDriver(record))}`.catch((error) => {
       throw new Error(`EventStore > 'events.insert' failed with postgres error: ${error.message}`);
     });
   }
@@ -37,16 +37,22 @@ export class PostgresEventsProvider implements EventsProvider {
    * @param records   - Event records to insert.
    * @param batchSize - Batch size for the insert loop.
    */
-  async insertMany(records: EventRecord[], batchSize: number = 1_000): Promise<void> {
+  async insertMany(records: EventRecord[], batchSize: number = 1_000, { tx }: Options = {}): Promise<void> {
-    await this.db.sql
+    if (tx !== undefined) {
-      .begin(async (sql) => {
+      for (let i = 0; i < records.length; i += batchSize) {
-        for (let i = 0; i < records.length; i += batchSize) {
+        await tx`INSERT INTO ${this.table} ${this.db.sql(records.slice(i, i + batchSize).map(this.#toDriver))}`;
-          await sql`INSERT INTO ${this.table} ${this.db.sql(records.slice(i, i + batchSize).map(this.#toDriver))}`;
+      }
-        }
+    } else {
-      })
+      await this.db.sql
-      .catch((error) => {
+        .begin(async (sql) => {
-        throw new Error(`EventStore > 'events.insertMany' failed with postgres error: ${error.message}`);
+          for (let i = 0; i < records.length; i += batchSize) {
-      });
+            await sql`INSERT INTO ${this.table} ${this.db.sql(records.slice(i, i + batchSize).map(this.#toDriver))}`;
+          }
+        })
+        .catch((error) => {
+          throw new Error(`EventStore > 'events.insertMany' failed with postgres error: ${error.message}`);
+        });
+    }
   }
 
   /**
@@ -55,10 +61,10 @@ export class PostgresEventsProvider implements EventsProvider {
    *
    * @param options - Find options.
    */
-  async get(options: EventReadOptions): Promise<EventRecord[]> {
+  async get(options: EventReadOptions, { tx }: Options = {}): Promise<EventRecord[]> {
     if (options !== undefined) {
       const { filter, cursor, direction, limit } = options;
-      return this.db.sql<PGEventRecord[]>`
+      return (tx ?? this.db.sql)<PGEventRecord[]>`
         SELECT * FROM ${this.table} 
         WHERE
           ${filter?.types ? this.#withTypes(filter.types) : this.db.sql``}
@@ -79,8 +85,9 @@ export class PostgresEventsProvider implements EventsProvider {
   async getByStream(
     stream: string,
     { filter, cursor, direction, limit }: EventReadOptions = {},
+    { tx }: Options = {},
   ): Promise<EventRecord[]> {
-    return this.db.sql<PGEventRecord[]>`
+    return (tx ?? this.db.sql)<PGEventRecord[]>`
       SELECT * FROM ${this.table} 
       WHERE 
         stream = ${stream}
@@ -100,8 +107,9 @@ export class PostgresEventsProvider implements EventsProvider {
   async getByStreams(
     streams: string[],
     { filter, cursor, direction, limit }: EventReadOptions = {},
+    { tx }: Options = {},
   ): Promise<EventRecord[]> {
-    return this.db.sql<PGEventRecord[]>`
+    return (tx ?? this.db.sql)<PGEventRecord[]>`
       SELECT * FROM ${this.table} 
       WHERE 
         stream IN ${this.db.sql(streams)}
@@ -117,8 +125,8 @@ export class PostgresEventsProvider implements EventsProvider {
    *
    * @param id - Event id.
    */
-  async getById(id: string): Promise<EventRecord | undefined> {
+  async getById(id: string, { tx }: Options = {}): Promise<EventRecord | undefined> {
-    return this.db.sql<PGEventRecord[]>`SELECT * FROM ${this.table} WHERE id = ${id}`
+    return (tx ?? this.db.sql)<PGEventRecord[]>`SELECT * FROM ${this.table} WHERE id = ${id}`
       .then(this.#fromDriver)
       .then(([record]) => record);
   }
@@ -126,8 +134,8 @@ export class PostgresEventsProvider implements EventsProvider {
   /**
    * Check if the given event is outdated in relation to the local event data.
    */
-  async checkOutdated({ stream, type, created }: EventRecord): Promise<boolean> {
+  async checkOutdated({ stream, type, created }: EventRecord, { tx }: Options = {}): Promise<boolean> {
-    const count = await await this.db.sql`
+    const count = await (tx ?? this.db.sql)`
       SELECT COUNT(*) AS count
       FROM ${this.table}
       WHERE

adapters/postgres/providers/relations.ts

@@ -1,7 +1,7 @@
 import type { Helper } from "postgres";
 
 import type { Relation, RelationPayload, RelationsProvider } from "../../../types/adapter.ts";
-import type { PostgresDatabase } from "../database.ts";
+import type { Options, PostgresDatabase } from "../database.ts";
 
 export class PostgresRelationsProvider implements RelationsProvider {
   constructor(
@@ -34,8 +34,9 @@ export class PostgresRelationsProvider implements RelationsProvider {
    * @param key    - Relational key to add stream to.
    * @param stream - Stream to add to the key.
    */
-  async insert(key: string, stream: string): Promise<void> {
+  async insert(key: string, stream: string, { tx }: Options = {}): Promise<void> {
-    await this.db.sql`INSERT INTO ${this.table} (key, stream) VALUES (${key}, ${stream}) ON CONFLICT DO NOTHING`.catch(
+    await (tx ??
+      this.db.sql)`INSERT INTO ${this.table} (key, stream) VALUES (${key}, ${stream}) ON CONFLICT DO NOTHING`.catch(
       (error) => {
         throw new Error(`EventStore > 'relations.insert' failed with postgres error: ${error.message}`);
       },
@@ -48,17 +49,24 @@ export class PostgresRelationsProvider implements RelationsProvider {
    * @param relations - Relations to insert.
    * @param batchSize - Batch size for the insert loop.
    */
-  async insertMany(relations: RelationPayload[], batchSize: number = 1_000): Promise<void> {
+  async insertMany(relations: RelationPayload[], batchSize: number = 1_000, { tx }: Options = {}): Promise<void> {
-    await this.db.sql
+    if (tx !== undefined) {
-      .begin(async (sql) => {
+      for (let i = 0; i < relations.length; i += batchSize) {
-        for (let i = 0; i < relations.length; i += batchSize) {
+        const values = relations.slice(i, i + batchSize).map(({ key, stream }) => [key, stream]);
-          const values = relations.slice(i, i + batchSize).map(({ key, stream }) => [key, stream]);
+        await tx`INSERT INTO ${this.table} (key, stream) VALUES ${this.db.sql(values)} ON CONFLICT DO NOTHING`;
-          await sql`INSERT INTO ${this.table} (key, stream) VALUES ${sql(values)} ON CONFLICT DO NOTHING`;
+      }
-        }
+    } else {
-      })
+      await this.db.sql
-      .catch((error) => {
+        .begin(async (sql) => {
-        throw new Error(`EventStore > 'relations.insertMany' failed with postgres error: ${error.message}`);
+          for (let i = 0; i < relations.length; i += batchSize) {
-      });
+            const values = relations.slice(i, i + batchSize).map(({ key, stream }) => [key, stream]);
+            await sql`INSERT INTO ${this.table} (key, stream) VALUES ${sql(values)} ON CONFLICT DO NOTHING`;
+          }
+        })
+        .catch((error) => {
+          throw new Error(`EventStore > 'relations.insertMany' failed with postgres error: ${error.message}`);
+        });
+    }
   }
 
   /**
@@ -66,8 +74,8 @@ export class PostgresRelationsProvider implements RelationsProvider {
    *
    * @param key - Relational key to get event streams for.
    */
-  async getByKey(key: string): Promise<string[]> {
+  async getByKey(key: string, { tx }: Options = {}): Promise<string[]> {
-    return this.db.sql`SELECT stream FROM ${this.table} WHERE key = ${key}`
+    return (tx ?? this.db.sql)`SELECT stream FROM ${this.table} WHERE key = ${key}`
       .then((rows) => rows.map(({ stream }) => stream))
       .catch((error) => {
         throw new Error(`EventStore > 'relations.getByKey' failed with postgres error: ${error.message}`);
@@ -79,8 +87,8 @@ export class PostgresRelationsProvider implements RelationsProvider {
    *
    * @param keys - Relational keys to get event streams for.
    */
-  async getByKeys(keys: string[]): Promise<string[]> {
+  async getByKeys(keys: string[], { tx }: Options = {}): Promise<string[]> {
-    return this.db.sql`SELECT DISTINCT stream FROM ${this.table} WHERE key IN ${this.db.sql(keys)}`
+    return (tx ?? this.db.sql)`SELECT DISTINCT stream FROM ${this.table} WHERE key IN ${this.db.sql(keys)}`
       .then((rows) => rows.map(({ stream }) => stream))
       .catch((error) => {
         throw new Error(`EventStore > 'relations.getByKeys' failed with postgres error: ${error.message}`);
@@ -93,8 +101,8 @@ export class PostgresRelationsProvider implements RelationsProvider {
    * @param key    - Relational key to remove stream from.
    * @param stream - Stream to remove from relation.
    */
-  async remove(key: string, stream: string): Promise<void> {
+  async remove(key: string, stream: string, { tx }: Options = {}): Promise<void> {
-    await this.db.sql`DELETE FROM ${this.table} WHERE key = ${key} AND stream = ${stream}`.catch((error) => {
+    await (tx ?? this.db.sql)`DELETE FROM ${this.table} WHERE key = ${key} AND stream = ${stream}`.catch((error) => {
       throw new Error(`EventStore > 'relations.remove' failed with postgres error: ${error.message}`);
     });
   }
@@ -105,19 +113,28 @@ export class PostgresRelationsProvider implements RelationsProvider {
    * @param relations - Relations to remove stream from.
    * @param batchSize - Batch size for the insert loop.
    */
-  async removeMany(relations: RelationPayload[], batchSize: number = 1_000): Promise<void> {
+  async removeMany(relations: RelationPayload[], batchSize: number = 1_000, { tx }: Options = {}): Promise<void> {
-    await this.db.sql
+    if (tx !== undefined) {
-      .begin(async (sql) => {
+      for (let i = 0; i < relations.length; i += batchSize) {
-        for (let i = 0; i < relations.length; i += batchSize) {
+        const conditions = relations
-          const conditions = relations
+          .slice(i, i + batchSize)
-            .slice(i, i + batchSize)
+          .map(({ key, stream }) => `(key = '${key}' AND stream = '${stream}')`);
-            .map(({ key, stream }) => `(key = '${key}' AND stream = '${stream}')`);
+        await tx`DELETE FROM ${this.table} WHERE ${this.db.sql.unsafe(conditions.join(" OR "))}`;
-          await sql`DELETE FROM ${this.table} WHERE ${this.db.sql.unsafe(conditions.join(" OR "))}`;
+      }
-        }
+    } else {
-      })
+      await this.db.sql
-      .catch((error) => {
+        .begin(async (sql) => {
-        throw new Error(`EventStore > 'relations.removeMany' failed with postgres error: ${error.message}`);
+          for (let i = 0; i < relations.length; i += batchSize) {
-      });
+            const conditions = relations
+              .slice(i, i + batchSize)
+              .map(({ key, stream }) => `(key = '${key}' AND stream = '${stream}')`);
+            await sql`DELETE FROM ${this.table} WHERE ${this.db.sql.unsafe(conditions.join(" OR "))}`;
+          }
+        })
+        .catch((error) => {
+          throw new Error(`EventStore > 'relations.removeMany' failed with postgres error: ${error.message}`);
+        });
+    }
   }
 
   /**
@@ -125,8 +142,8 @@ export class PostgresRelationsProvider implements RelationsProvider {
    *
    * @param keys - Relational keys to remove from the relational table.
    */
-  async removeByKeys(keys: string[]): Promise<void> {
+  async removeByKeys(keys: string[], { tx }: Options = {}): Promise<void> {
-    await this.db.sql`DELETE FROM ${this.table} WHERE key IN ${this.db.sql(keys)}`.catch((error) => {
+    await (tx ?? this.db.sql)`DELETE FROM ${this.table} WHERE key IN ${this.db.sql(keys)}`.catch((error) => {
       throw new Error(`EventStore > 'relations.removeByKeys' failed with postgres error: ${error.message}`);
     });
   }
@@ -136,8 +153,8 @@ export class PostgresRelationsProvider implements RelationsProvider {
    *
    * @param streams - Streams to remove from the relational table.
    */
-  async removeByStreams(streams: string[]): Promise<void> {
+  async removeByStreams(streams: string[], { tx }: Options = {}): Promise<void> {
-    await this.db.sql`DELETE FROM ${this.table} WHERE stream IN ${this.db.sql(streams)}`.catch((error) => {
+    await (tx ?? this.db.sql)`DELETE FROM ${this.table} WHERE stream IN ${this.db.sql(streams)}`.catch((error) => {
       throw new Error(`EventStore > 'relations.removeByStreams' failed with postgres error: ${error.message}`);
     });
   }

adapters/postgres/providers/snapshot.ts

@@ -1,7 +1,7 @@
 import type { Helper } from "postgres";
 
 import type { Snapshot, SnapshotsProvider } from "../../../types/adapter.ts";
-import type { PostgresDatabase } from "../database.ts";
+import type { Options, PostgresDatabase } from "../database.ts";
 
 type PGSnapshot = Omit<Snapshot, "state"> & { state: string };
 
@@ -26,8 +26,8 @@ export class PostgresSnapshotsProvider implements SnapshotsProvider {
    * @param cursor - Cursor timestamp for the last event used in the snapshot.
    * @param state  - State of the reduced events.
    */
-  async insert(name: string, stream: string, cursor: string, state: any): Promise<void> {
+  async insert(name: string, stream: string, cursor: string, state: any, { tx }: Options = {}): Promise<void> {
-    await this.db.sql`
+    await (tx ?? this.db.sql)`
       INSERT INTO ${this.table} ${this.db.sql(this.#toDriver({ name, stream, cursor, state }))}`.catch((error) => {
       throw new Error(`EventStore > 'snapshots.insert' failed with postgres error: ${error.message}`);
     });
@@ -39,8 +39,8 @@ export class PostgresSnapshotsProvider implements SnapshotsProvider {
    * @param name   - Name of the reducer which the state was created.
    * @param stream - Stream the state was reduced for.
    */
-  async getByStream(name: string, stream: string): Promise<Snapshot | undefined> {
+  async getByStream(name: string, stream: string, { tx }: Options = {}): Promise<Snapshot | undefined> {
-    return this.db.sql<PGSnapshot[]>`SELECT * FROM ${this.table} WHERE name = ${name} AND stream = ${stream}`
+    return (tx ?? this.db.sql)<PGSnapshot[]>`SELECT * FROM ${this.table} WHERE name = ${name} AND stream = ${stream}`
       .then(this.#fromDriver)
       .then(([snapshot]) => snapshot)
       .catch((error) => {
@@ -54,8 +54,8 @@ export class PostgresSnapshotsProvider implements SnapshotsProvider {
    * @param name   - Name of the reducer the snapshot is attached to.
    * @param stream - Stream to remove from snapshots.
    */
-  async remove(name: string, stream: string): Promise<void> {
+  async remove(name: string, stream: string, { tx }: Options = {}): Promise<void> {
-    await this.db.sql`DELETE FROM ${this.table} WHERE name = ${name} AND stream = ${stream}`.catch((error) => {
+    await (tx ?? this.db.sql)`DELETE FROM ${this.table} WHERE name = ${name} AND stream = ${stream}`.catch((error) => {
       throw new Error(`EventStore > 'snapshots.remove' failed with postgres error: ${error.message}`);
     });
   }

deno.json

@@ -1,6 +1,6 @@
 {
   "name": "@valkyr/event-store",
-  "version": "2.0.1",
+  "version": "2.0.2",
   "exports": {
     ".": "./mod.ts",
     "./browser": "./adapters/browser/adapter.ts",

libraries/event-store.ts

@@ -559,7 +559,7 @@ type EventStoreConfig<TEventFactory extends EventFactory, TEventStoreAdapter ext
   hooks?: EventStoreHooks<TEventFactory>;
 };
 
-export type EventsInsertSettings = {
+export interface EventsInsertSettings {
   /**
    * Should the event store emit events after successfull insertion.
    * This only takes false as value and by default events are always