Added support for Mongo-like Sessions and Transactions

src/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@event-driven-io/pongo-core",
-  "version": "0.7.0",
+  "version": "0.8.0",
   "description": "Pongo - Mongo with strong consistency on top of Postgres",
   "type": "module",
   "engines": {

src/packages/dumbo/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@event-driven-io/dumbo",
-  "version": "0.5.0",
+  "version": "0.6.0",
   "description": "Dumbo - tools for dealing with PostgreSQL",
   "type": "module",
   "scripts": {

src/packages/dumbo/src/core/connections/connection.ts

@@ -0,0 +1,12 @@
+import type { WithSQLExecutor } from '../execute';
+import type { DatabaseTransactionFactory } from './transaction';
+
+export interface Connection<
+  ConnectorType extends string = string,
+  DbClient = unknown,
+> extends WithSQLExecutor,
+    DatabaseTransactionFactory<ConnectorType> {
+  type: ConnectorType;
+  connect: () => Promise<DbClient>;
+  close: () => Promise<void>;
+}

src/packages/dumbo/src/core/connections/index.ts

@@ -0,0 +1,3 @@
+export * from './connection';
+export * from './pool';
+export * from './transaction';

src/packages/dumbo/src/core/connections/pool.ts

@@ -0,0 +1,16 @@
+import type { WithSQLExecutor } from '../execute';
+import { type Connection } from './connection';
+import type { DatabaseTransactionFactory } from './transaction';
+
+export interface ConnectionPool<ConnectionType extends Connection = Connection>
+  extends WithSQLExecutor,
+    DatabaseTransactionFactory<ConnectionType['type']> {
+  type: ConnectionType['type'];
+  open: () => Promise<ConnectionType>;
+  close: () => Promise<void>;
+}
+
+export type ConnectionPoolProvider<
+  ConnectionPoolType extends ConnectionPool = ConnectionPool,
+  ConnectionPoolOptions = unknown,
+> = (options: ConnectionPoolOptions) => ConnectionPoolType;

src/packages/dumbo/src/core/connections/transaction.ts

@@ -0,0 +1,99 @@
+import type { WithSQLExecutor } from '../execute';
+import { type Connection } from './connection';
+
+export interface DatabaseTransaction<ConnectorType extends string = string>
+  extends WithSQLExecutor {
+  type: ConnectorType;
+  begin: () => Promise<void>;
+  commit: () => Promise<void>;
+  rollback: (error?: unknown) => Promise<void>;
+}
+
+export interface DatabaseTransactionFactory<
+  ConnectorType extends string = string,
+> {
+  transaction: () => DatabaseTransaction<ConnectorType>;
+
+  withTransaction: <Result = unknown>(
+    handle: (
+      transaction: DatabaseTransaction<ConnectorType>,
+    ) => Promise<{ success: boolean; result: Result }>,
+  ) => Promise<Result>;
+}
+
+export const executeInTransaction = async <
+  ConnectorType extends string = string,
+  Result = unknown,
+>(
+  transaction: DatabaseTransaction<ConnectorType>,
+  handle: (
+    transaction: DatabaseTransaction<ConnectorType>,
+  ) => Promise<{ success: boolean; result: Result }>,
+): Promise<Result> => {
+  await transaction.begin();
+
+  try {
+    const { success, result } = await handle(transaction);
+
+    if (success) await transaction.commit();
+    else await transaction.rollback();
+
+    return result;
+  } catch (e) {
+    await transaction.rollback();
+    throw e;
+  }
+};
+
+export const transactionFactoryWithDbClient = <
+  ConnectorType extends string = string,
+  DbClient = unknown,
+>(
+  connect: () => Promise<DbClient>,
+  initTransaction: (
+    client: Promise<DbClient>,
+  ) => DatabaseTransaction<ConnectorType>,
+): DatabaseTransactionFactory<ConnectorType> => ({
+  transaction: () => initTransaction(connect()),
+  withTransaction: (handle) =>
+    executeInTransaction(initTransaction(connect()), handle),
+});
+
+const wrapInConnectionClosure = async <
+  ConnectionType extends Connection = Connection,
+  Result = unknown,
+>(
+  connection: ConnectionType,
+  handle: () => Promise<Result>,
+) => {
+  try {
+    return await handle();
+  } finally {
+    await connection.close();
+  }
+};
+
+export const transactionFactoryWithNewConnection = <
+  ConnectionType extends Connection = Connection,
+>(
+  connect: () => ConnectionType,
+): DatabaseTransactionFactory<ConnectionType['type']> => ({
+  transaction: () => {
+    const connection = connect();
+    const transaction = connection.transaction();
+
+    return {
+      ...transaction,
+      commit: () =>
+        wrapInConnectionClosure(connection, () => transaction.commit()),
+      rollback: () =>
+        wrapInConnectionClosure(connection, () => transaction.rollback()),
+    };
+  },
+  withTransaction: (handle) => {
+    const connection = connect();
+    return wrapInConnectionClosure(connection, () =>
+      connection.withTransaction(handle),
+    );
+  },
+});