Support pool connection for mysql and postgresql #202

Activity

1. ITsPro-CN

   added

   enhancement

   on Dec 22, 2025
2. beeequeue commented on Jan 23, 2026

   

   beeequeue

   on Jan 23, 2026

   Last edited by beeequeue

   Contributor

   More actions

   connection pooling would be required to implement transactions support (#49).
   a db connection should never accidentally run an unrelated query between opening and closing a transaction, meaning that if the client only has one connection it has to block all other queries until the transaction is finished.

   ive been trying my hand at an implementation based on what ive seen from other db clients that looks like this:

   Connector gets a new acquireConnection function that returns a Connection instance. A Connection is pretty much a Database except it calls the db client via whatever connection pooling interface it uses, or blocks other queries until the connection is released if pooling doesnt exist (e.g. sqlite)

   then Database also gets a acquireConnection function that users can use to execute queries without worrying about other queries interfering.

   const db = createDatabase(x)
   
   await db.acquireConnection(async (conn) => {
     await conn.sql`DELETE FROM users;`
   })

   internally, the database automatically disposes of the connection when the function is done

   export type Connection = Pick<
     Database,
     | "dialect"
     | "exec"
     | "prepare"
     | "sql"
     | "dispose"
     | typeof Symbol.asyncDispose
   >;
   
   type Connector = {
     acquireConnection: () => Promise<Omit<Connection, "sql">>;
   }
   
   type Database = {
     acquireConnection: (fn: (conn: Connection) => void | Promise<void>) => Promise<void>
   }

   as for things im unsure about is the naming of various things, as well as whether it would be good to also expose a function on Database that just returns a raw connection
3. ITsPro-CN commented on Jan 27, 2026

   

   ITsPro-CN

   on Jan 27, 2026

   Last edited by ITsPro-CN

   Author

   More actions

   Yes, this has been taken into consideration.

   For the non-pooling Connector, the getInstance method returns the database client directly.
   For the pooling Connector, the getInstance method returns the connection pool.

   You can call useDatabase().getInstance().connect() to retrieve a PostgreSQL pool client for transactions.
   However, a key distinction is that while you can execute queries directly on the pool using syntax like

   db.sql`SOME SQL STATEMENT`

   And that is not supported on the pool client itself.
   Therefore, to enable a consistent query interface for the pool client, we need to implement a wrapper around it.

   export type ConnectorOptions = pg.PoolConfig;
   type Pool = pg.Pool & {
     getClient?: () => Promise<Database<Connector<pg.PoolClient>>>;
   };
   
   export default function postgresqlPoolConnector(
     opts: ConnectorOptions,
   ): Connector<Pool> {
     let _pool: Pool | undefined;
     const getPool = () => {
       if (!_pool) _pool = new pg.Pool(opts);
       _pool.getClient = getClient;
       return _pool;
     };
     const getClient = async () => {
       let _client: pg.PoolClient | undefined;
       let _clientPromise: Promise<pg.PoolClient> | undefined;
       const _getClient = async () => {
         if (!_client) {
           if (!_clientPromise) {
             _clientPromise = getPool()
               .connect()
               .then((client) => {
                 _client = client;
                 _clientPromise = undefined;
                 return client;
               });
           }
           return _clientPromise;
         }
         return _client;
       };
       const _query: InternalQuery = (sql, params) =>
         _getClient().then((c) => c.query(normalizeParams(sql), params));
       return createDatabase({
         name: "postgresql-pool-client",
         dialect: "postgresql",
         getInstance: () => _getClient(),
         exec: (sql) => _query(sql),
         prepare: (sql) => new StatementWrapper(sql, _query),
         dispose: async () => {
           _client?.release?.();
           _client = undefined;
         },
       });
     };
   
     const query: InternalQuery = async (sql, params) => {
       const pool = getPool();
       return pool.query(normalizeParams(sql), params);
     };
   
     return {
       name: "postgresql-pool",
       dialect: "postgresql",
       getInstance: () => getPool(),
       exec: (sql) => query(sql),
       prepare: (sql) => new StatementWrapper(sql, query),
       dispose: async () => {
         await _pool?.end?.();
         _pool = undefined;
       },
     };
   }
   
   // https://www.postgresql.org/docs/9.3/sql-prepare.html
   function normalizeParams(sql: string) {
     let i = 0;
     return sql.replace(/\?/g, () => `$${++i}`);
   }
   
   class StatementWrapper extends BoundableStatement<void> {
     #query: InternalQuery;
     #sql: string;
   
     constructor(sql: string, query: InternalQuery) {
       super();
       this.#sql = sql;
       this.#query = query;
     }
   
     async all(...params: Primitive[]) {
       const res = await this.#query(this.#sql, params);
       return res.rows;
     }
   
     async run(...params: Primitive[]) {
       const res = await this.#query(this.#sql, params);
       return {
         success: true,
         ...res,
       };
     }
   
     async get(...params: Primitive[]) {
       const res = await this.#query(this.#sql, params);
       return res.rows[0];
     }
   }

   It looks like this PR hasn’t been reviewed yet.
4. ITsPro-CN

   closed this as completedon Jan 27, 2026