refactor(db/startTransaction): use internal query functions

Some changes to function signature and callback.
Make null-returns commit changes rather than rollback.

Author: thelindat

src/database/rawExecute.ts

@@ -12,7 +12,8 @@ export const rawExecute = async (
   parameters: CFXParameters,
   cb?: CFXCallback,
   isPromise?: boolean,
-  unpack?: boolean
+  unpack?: boolean,
+  connectionId?: number
 ) => {
   cb = setCallback(parameters, cb);
 
@@ -27,7 +28,7 @@ export const rawExecute = async (
     return logError(invokingResource, cb, isPromise, err, query, parameters);
   }
 
-  const connection = await getPoolConnection();
+  const connection = await getPoolConnection(connectionId);
 
   if (!connection) return;
 
@@ -48,13 +49,11 @@ export const rawExecute = async (
       const startTime = !hasProfiler && performance.now();
       const [result] = await connection.execute(query, values);
 
-      if (cb) {
-        if (Array.isArray(result) && result.length > 1) {
-          for (const value of result) {
-            response.push(unpack ? parseResponse(type, value) : value);
-          }
-        } else response.push(unpack ? parseResponse(type, result) : result);
-      }
+      if (Array.isArray(result) && result.length > 1) {
+        for (const value of result) {
+          response.push(unpack ? parseResponse(type, value) : value);
+        }
+      } else response.push(unpack ? parseResponse(type, result) : result);
 
       if (hasProfiler && ((index > 0 && index % 100 === 0) || index === parametersLength - 1)) {
         await profileBatchStatements(connection, invokingResource, query, parameters, index < 100 ? 0 : index);
@@ -63,7 +62,7 @@ export const rawExecute = async (
       }
     }
 
-    if (!cb) return;
+    if (!cb) return response.length === 1 ? response[0] : response;
 
     try {
       if (response.length === 1) {
@@ -84,6 +83,8 @@ export const rawExecute = async (
       }
     }
   } catch (err: any) {
+    if (!cb) throw new Error(err.message || err);
+
     logError(invokingResource, cb, isPromise, err, query, parameters);
   } finally {
     connection.release();

src/database/rawQuery.ts

@@ -14,7 +14,8 @@ export const rawQuery = async (
   query: string,
   parameters: CFXParameters,
   cb?: CFXCallback,
-  isPromise?: boolean
+  isPromise?: boolean,
+  connectionId?: number
 ) => {
   cb = setCallback(parameters, cb);
   try {
@@ -23,7 +24,7 @@ export const rawQuery = async (
     return logError(invokingResource, cb, err, isPromise, query, parameters);
   }
 
-  const connection = await getPoolConnection();
+  const connection = await getPoolConnection(connectionId);
 
   if (!connection) return;
 
@@ -42,16 +43,19 @@ export const rawQuery = async (
       logQuery(invokingResource, query, performance.now() - startTime, parameters);
     }
 
-    if (cb)
-      try {
-        cb(parseResponse(type, result));
-      } catch (err) {
-        if (typeof err === 'string') {
-          if (err.includes('SCRIPT ERROR:')) return console.log(err);
-          console.log(`^1SCRIPT ERROR in invoking resource ${invokingResource}: ${err}^0`);
-        }
+    if (!cb) return parseResponse(type, result);
+
+    try {
+      cb(parseResponse(type, result));
+    } catch (err) {
+      if (typeof err === 'string') {
+        if (err.includes('SCRIPT ERROR:')) return console.log(err);
+        console.log(`^1SCRIPT ERROR in invoking resource ${invokingResource}: ${err}^0`);
       }
+    }
   } catch (err: any) {
+    if (!cb) throw new Error(err.message || err);
+
     logError(invokingResource, cb, isPromise, err, query, parameters, true);
   } finally {
     connection.release();

src/database/startTransaction.ts

@@ -1,35 +1,35 @@
 import { getPoolConnection } from './connection';
 import { logError } from '../logger';
 import { CFXCallback, CFXParameters } from '../types';
-import { setCallback } from '../utils/setCallback';
+import { rawQuery } from './rawQuery';
+import { rawExecute } from './rawExecute';
 
 export const startTransaction = async (
   invokingResource: string,
   queries: (...args: any[]) => Promise<boolean>,
-  parameters: CFXParameters,
   cb?: CFXCallback,
   isPromise?: boolean
 ) => {
-  cb = setCallback(parameters, cb);
-
   const conn = await getPoolConnection();
 
   if (!conn) return;
 
   let response = false;
 
   try {
+    const connectionId = (conn as any).connection.connectionId;
     await conn.beginTransaction();
 
     const commit = await queries({
-      query: async (sql: string, values?: CFXParameters) => {
-        const [rows] = await conn.query(sql, values);
-        return rows;
+      query: (sql: string, values: CFXParameters) => {
+        return rawQuery(null, invokingResource, sql, values, undefined, isPromise, connectionId);
+      },
+      execute: (sql: string, values: CFXParameters) => {
+        return rawExecute(invokingResource, sql, values, undefined, isPromise, connectionId);
       },
     });
 
-    response = !!commit;
-
+    response = commit === false ? false : true;
     response ? conn.commit() : conn.rollback();
   } catch (err: any) {
     conn.rollback();

src/index.ts

@@ -77,12 +77,11 @@ global.exports(
   'experimentalTransaction',
   async (
     transactions: () => Promise<boolean>,
-    parameters: CFXParameters,
     cb: CFXCallback,
     invokingResource = GetInvokingResource(),
     isPromise?: boolean
   ) => {
-    return await startTransaction(invokingResource, transactions, parameters, cb, isPromise);
+    return await startTransaction(invokingResource, transactions, cb, isPromise);
   }
 );