feat: support sql transactions for multiple databases

src/helpers/values.ts

@@ -142,3 +142,19 @@ export function toEnumValue<T>(enm: { [s: string]: T }, value: string): T | unde
     ? (value as unknown as T)
     : undefined;
 }
+
+/**
+ * Unwraps a value that may be undefined or null.
+ * @param val - The value to unwrap
+ * @param onNullish - Callback to throw an error if the value is null or undefined
+ * @returns The unwrapped value
+ */
+export function unwrap<T>(val: T | null, onNullish?: () => string): Exclude<T, undefined | null> {
+  if (val === undefined) {
+    throw new Error(onNullish?.() ?? 'value is undefined');
+  }
+  if (val === null) {
+    throw new Error(onNullish?.() ?? 'value is null');
+  }
+  return val as Exclude<T, undefined | null>;
+}

src/postgres/__tests__/base-pg-store.test.ts

@@ -92,21 +92,18 @@ describe('BasePgStore', () => {
   });
 
   test('postgres transaction connection integrity', async () => {
-    const usageName = 'postgres:test;datastore-crud';
     const obj = db.sql;
+    const dbName = obj.options.database;
 
     expect(sqlTransactionContext.getStore()).toBeUndefined();
     await db.sqlTransaction(async sql => {
-      // Transaction flag is open.
-      expect(sqlTransactionContext.getStore()?.usageName).toBe(usageName);
       // New connection object.
       const newObj = sql;
       expect(obj).not.toEqual(newObj);
-      expect(sqlTransactionContext.getStore()?.sql).toEqual(newObj);
+      expect(sqlTransactionContext.getStore()?.[dbName]).toEqual(newObj);
 
       // Nested tx uses the same connection object.
       await db.sqlTransaction(sql => {
-        expect(sqlTransactionContext.getStore()?.usageName).toBe(usageName);
         expect(newObj).toEqual(sql);
       });
 

src/postgres/base-pg-store.ts

@@ -8,8 +8,7 @@ import { isProdEnv } from '../helpers/values';
  */
 export const sqlTransactionContext = new AsyncLocalStorage<SqlTransactionContext>();
 type SqlTransactionContext = {
-  usageName: string;
-  sql: PgSqlClient;
+  [dbName: string]: PgSqlClient;
 };
 type UnwrapPromiseArray<T> = T extends any[]
   ? {
@@ -26,8 +25,9 @@ export abstract class BasePgStore {
    * async context will be returned to guarantee transaction consistency.
    */
   get sql(): PgSqlClient {
+    const dbName = this._sql.options.database?.toString() ?? 'default';
     const sqlContext = sqlTransactionContext.getStore();
-    return sqlContext ? sqlContext.sql : this._sql;
+    return sqlContext ? sqlContext[dbName] : this._sql;
   }
   private readonly _sql: PgSqlClient;
 
@@ -52,15 +52,16 @@ export abstract class BasePgStore {
     callback: (sql: PgSqlClient) => T | Promise<T>,
     readOnly = true
   ): Promise<UnwrapPromiseArray<T>> {
-    // Do we have a scoped client already? Use it directly.
-    const sqlContext = sqlTransactionContext.getStore();
-    if (sqlContext) {
-      return callback(sqlContext.sql) as UnwrapPromiseArray<T>;
+    // Do we have a scoped client already? Use it directly. Key is the database name.
+    const dbName = this._sql.options.database?.toString() ?? 'default';
+    const sql = sqlTransactionContext.getStore()?.[dbName];
+    if (sql) {
+      return callback(sql) as UnwrapPromiseArray<T>;
     }
     // Otherwise, start a transaction and store the scoped connection in the current async context.
-    const usageName = this._sql.options.connection.application_name ?? '';
     return this._sql.begin(readOnly ? 'read only' : 'read write', sql => {
-      return sqlTransactionContext.run({ usageName, sql }, () => callback(sql));
+      const currentStore = sqlTransactionContext.getStore() ?? {};
+      return sqlTransactionContext.run({ ...currentStore, [dbName]: sql }, () => callback(sql));
     });
   }