Moved deferred pool, connection and executor to respectful files

Author: oskardudycz

src/packages/dumbo/src/benchmarks/index.ts

@@ -2,10 +2,14 @@ import 'dotenv/config';
 
 import Benchmark from 'benchmark';
 import pg from 'pg';
-import { postgreSQLConnectionString, rawSql, single } from '..';
-import { defaultPostgreSQLConenctionString, dumbo } from '../pg';
+import { rawSql, single } from '..';
+import {
+  defaultPostgreSQLConenctionString,
+  dumbo,
+  PostgreSQLConnectionString,
+} from '../pg';
 
-const connectionString = postgreSQLConnectionString(
+const connectionString = PostgreSQLConnectionString(
   process.env.BENCHMARK_POSTGRESQL_CONNECTION_STRING ??
     defaultPostgreSQLConenctionString,
 );

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

@@ -1,5 +1,6 @@
 import type { ConnectorType } from '../connectors';
 import {
+  createLazyExecutor,
   sqlExecutor,
   type DbSQLExecutor,
   type WithSQLExecutor,
@@ -85,3 +86,51 @@ export const createConnection = <
 
   return typedConnection;
 };
+
+export const createDeferredConnection = <Connector extends ConnectorType>(
+  connector: Connector,
+  importConnection: () => Promise<Connection<Connector>>,
+): Connection<Connector> => {
+  const getConnection = importConnection();
+
+  const execute = createLazyExecutor(async () => {
+    const conn = await getConnection;
+    return conn.execute;
+  });
+
+  const connection: Connection<Connector> = {
+    connector,
+    execute,
+
+    open: async (): Promise<unknown> => {
+      const conn = await getConnection;
+      return conn.open();
+    },
+
+    close: async (): Promise<void> => {
+      if (getConnection) {
+        const conn = await getConnection;
+        await conn.close();
+      }
+    },
+
+    transaction: () => {
+      const transaction = getConnection.then((c) => c.transaction());
+
+      return {
+        connector,
+        connection,
+        execute: createLazyExecutor(async () => (await transaction).execute),
+        begin: async () => (await transaction).begin(),
+        commit: async () => (await transaction).commit(),
+        rollback: async () => (await transaction).rollback(),
+      };
+    },
+    withTransaction: async (handle) => {
+      const connection = await getConnection;
+      return connection.withTransaction(handle);
+    },
+  };
+
+  return connection;
+};

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

@@ -1,3 +1,5 @@
+import { createDeferredConnection } from '../connections';
+import type { ConnectorType } from '../connectors';
 import {
   executeInNewConnection,
   sqlExecutorInNewConnection,
@@ -72,3 +74,37 @@ export const createConnectionPool = <
 
   return result as ConnectionPoolType;
 };
+
+export const createDeferredConnectionPool = <Connector extends ConnectorType>(
+  connector: Connector,
+  importPool: () => Promise<ConnectionPool<Connection<Connector>>>,
+): ConnectionPool<Connection<Connector>> => {
+  let poolPromise: Promise<ConnectionPool<Connection<Connector>>> | null = null;
+
+  const getPool = async (): Promise<ConnectionPool<Connection<Connector>>> => {
+    if (!poolPromise) {
+      try {
+        poolPromise = importPool();
+      } catch (error) {
+        throw new Error(
+          `Failed to import connection pool: ${error instanceof Error ? error.message : String(error)}`,
+        );
+      }
+    }
+    return poolPromise;
+  };
+
+  return createConnectionPool({
+    connector,
+    close: async () => {
+      if (!poolPromise) return;
+      const pool = await poolPromise;
+      await pool.close();
+      poolPromise = null;
+    },
+    getConnection: () =>
+      createDeferredConnection(connector, async () =>
+        (await getPool()).connection(),
+      ),
+  });
+};

src/packages/dumbo/src/core/execute/execute.ts

@@ -154,3 +154,56 @@ export const executeInNewConnection = async <
     await connection.close();
   }
 };
+
+export const createLazyExecutor = (
+  importExecutor: () => Promise<SQLExecutor>,
+): SQLExecutor => {
+  let executor: SQLExecutor | null = null;
+
+  const getExecutor = async (): Promise<SQLExecutor> => {
+    if (!executor) {
+      try {
+        executor = await importExecutor();
+      } catch (error) {
+        throw new Error(
+          `Failed to import SQL executor: ${error instanceof Error ? error.message : String(error)}`,
+        );
+      }
+    }
+    return executor;
+  };
+
+  return {
+    query: async <Result extends QueryResultRow = QueryResultRow>(
+      sql: SQL,
+      options?: SQLQueryOptions,
+    ): Promise<QueryResult<Result>> => {
+      const exec = await getExecutor();
+      return exec.query<Result>(sql, options);
+    },
+
+    batchQuery: async <Result extends QueryResultRow = QueryResultRow>(
+      sqls: SQL[],
+      options?: SQLQueryOptions,
+    ): Promise<QueryResult<Result>[]> => {
+      const exec = await getExecutor();
+      return exec.batchQuery<Result>(sqls, options);
+    },
+
+    command: async <Result extends QueryResultRow = QueryResultRow>(
+      sql: SQL,
+      options?: SQLCommandOptions,
+    ): Promise<QueryResult<Result>> => {
+      const exec = await getExecutor();
+      return exec.command<Result>(sql, options);
+    },
+
+    batchCommand: async <Result extends QueryResultRow = QueryResultRow>(
+      sqls: SQL[],
+      options?: SQLCommandOptions,
+    ): Promise<QueryResult<Result>[]> => {
+      const exec = await getExecutor();
+      return exec.batchCommand<Result>(sqls, options);
+    },
+  };
+};