Finished the first draft of the new connection management, made code compiling

Author: oskardudycz

package-lock.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/package-lock.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/package.json

@@ -1,21 +1,12 @@
-export type Transaction<
-  ConnectorType extends string = string,
-  DbClient = unknown,
-> = {
-  type: ConnectorType;
-  client: Promise<DbClient>;
-  begin: () => Promise<void>;
-  commit: () => Promise<void>;
-  rollback: () => Promise<void>;
-};
+import type { WithSQLExecutor } from './execute';
+import type { TransactionFactory } from './transaction';
 
 export type Connection<
   ConnectorType extends string = string,
   DbClient = unknown,
 > = {
   type: ConnectorType;
-  open: () => Promise<DbClient>;
+  connect: () => Promise<DbClient>;
   close: () => Promise<void>;
-
-  beginTransaction: () => Promise<Transaction<ConnectorType, DbClient>>;
-};
+} & WithSQLExecutor &
+  TransactionFactory<ConnectorType>;

src/packages/dumbo/package.json

@@ -0,0 +1,78 @@
+import { type SQL } from '../sql';
+import type { Connection } from './connection';
+
+export interface QueryResultRow {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  [column: string]: any;
+}
+
+export type QueryResult<Result extends QueryResultRow = QueryResultRow> = {
+  rowCount: number | null;
+  rows: Result[];
+};
+
+export type SQLExecutor<
+  ConnectorType extends string = string,
+  DbClient = unknown,
+> = {
+  type: ConnectorType;
+  query<Result extends QueryResultRow = QueryResultRow>(
+    client: DbClient,
+    sql: SQL,
+  ): Promise<QueryResult<Result>>;
+};
+
+export type WithSQLExecutor = {
+  execute: {
+    query<Result extends QueryResultRow = QueryResultRow>(
+      sql: SQL,
+    ): Promise<QueryResult<Result>>;
+  };
+};
+
+export const withSqlExecutor = <
+  DbClient = unknown,
+  Executor extends SQLExecutor = SQLExecutor,
+>(
+  sqlExecutor: Executor,
+  options: {
+    connect: () => Promise<DbClient>;
+    close?: (client: DbClient, error?: unknown) => Promise<void>;
+  },
+): WithSQLExecutor => ({
+  execute: {
+    query: async <Result extends QueryResultRow = QueryResultRow>(sql: SQL) => {
+      const { connect, close } = options;
+      const client = await connect();
+
+      try {
+        const result = await sqlExecutor.query<Result>(client, sql);
+        if (close) await close(client);
+        return result;
+      } catch (error) {
+        if (close) await close(client, error);
+
+        throw error;
+      }
+    },
+  },
+});
+
+export const queryWithNewConnection = async <
+  ConnectionType extends Connection,
+  Result extends QueryResultRow = QueryResultRow,
+>(
+  connectionFactory: {
+    open: () => Promise<ConnectionType>;
+  },
+  sql: SQL,
+) => {
+  const { open } = connectionFactory;
+  const connection = await open();
+
+  try {
+    return await connection.execute.query<Result>(sql);
+  } finally {
+    await connection.close();
+  }
+};

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

@@ -1,4 +1,6 @@
 export * from './connection';
 export * from './connectionString';
+export * from './execute';
 export * from './pg';
 export * from './pool';
+export * from './transaction';

src/packages/dumbo/src/connections/execute.ts

@@ -4,10 +4,9 @@ import {
 } from '@testcontainers/postgresql';
 import { after, before, describe, it } from 'node:test';
 import pg from 'pg';
-import { pgConnection } from '.';
-import { executeSQL } from '../../execute';
+import { nodePostgresPool } from '.';
 import { rawSql } from '../../sql';
-import { endPool, getPool } from '../pool';
+import { endPool, getPool } from './pool';
 
 void describe('PostgreSQL connection', () => {
   let postgres: StartedPostgreSqlContainer;
@@ -24,68 +23,63 @@ void describe('PostgreSQL connection', () => {
 
   void describe('executeSQL', () => {
     void it('connects using pool', async () => {
-      const connection = pgConnection({ connectionString });
+      const pool = nodePostgresPool({ connectionString });
+      const connection = await pool.open();
 
       try {
-        await executeSQL(connection.pool, rawSql('SELECT 1'));
+        await connection.execute.query(rawSql('SELECT 1'));
       } catch (error) {
         console.log(error);
       } finally {
         await connection.close();
-      }
-    });
-
-    void it('connects using connected pool client', async () => {
-      const connection = pgConnection({ connectionString });
-      const poolClient = await connection.open();
-
-      try {
-        await executeSQL(poolClient, rawSql('SELECT 1'));
-      } finally {
-        await connection.close();
+        await pool.close();
       }
     });
 
     void it('connects using existing pool', async () => {
-      const pool = getPool(connectionString);
-      const connection = pgConnection({ connectionString, pool });
+      const nativePool = getPool(connectionString);
+      const pool = nodePostgresPool({ connectionString, pool: nativePool });
+      const connection = await pool.open();
 
       try {
-        await executeSQL(pool, rawSql('SELECT 1'));
+        await connection.execute.query(rawSql('SELECT 1'));
       } finally {
         await connection.close();
+        await pool.close();
         await endPool({ connectionString });
       }
     });
 
     void it('connects using client', async () => {
-      const connection = pgConnection({
+      const pool = nodePostgresPool({
         connectionString,
         type: 'client',
       });
-      const client = await connection.open();
+      const connection = await pool.open();
 
       try {
-        await executeSQL(client, rawSql('SELECT 1'));
+        await connection.execute.query(rawSql('SELECT 1'));
       } finally {
         await connection.close();
+        await pool.close();
       }
     });
 
     void it('connects using connected client', async () => {
       const existingClient = new pg.Client({ connectionString });
       await existingClient.connect();
 
-      const connection = pgConnection({
+      const pool = nodePostgresPool({
         connectionString,
         client: existingClient,
       });
-      const client = await connection.open();
+      const connection = await pool.open();
 
       try {
-        await executeSQL(client, rawSql('SELECT 1'));
+        await connection.execute.query(rawSql('SELECT 1'));
       } finally {
         await connection.close();
+        await pool.close();
         await existingClient.end();
       }
     });