updated DB adapters with utils

Author: stevensJourney

packages/journeyapps-react-native-quick-sqlite

@@ -68,9 +68,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this.closed = true;
     this.options = { ...DEFAULT_POWERSYNC_DB_OPTIONS, ...options };
     this.bucketStorageAdapter = this.generateBucketStorageAdapter();
-    this.sdkVersion = this.options.database.execute('SELECT powersync_rs_version()').rows?.item(0)[
-      'powersync_rs_version()'
-    ];
+    this.sdkVersion = '';
   }
 
   get schema() {
@@ -96,7 +94,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     this.initialized = (async () => {
       await this._init();
       await this.bucketStorageAdapter.init();
-      await this.database.executeAsync('SELECT powersync_replace_schema(?)', [JSON.stringify(this.schema.toJSON())]);
+      await this.database.execute('SELECT powersync_replace_schema(?)', [JSON.stringify(this.schema.toJSON())]);
+      const version = await this.options.database.execute('SELECT powersync_rs_version()');
+      this.sdkVersion = version.rows?.item(0)['powersync_rs_version()'] ?? '';
     })();
     await this.initialized;
   }
@@ -109,7 +109,6 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     await this.disconnect();
 
     await this.initialized;
-
     this.syncStreamImplementation = this.generateSyncStreamImplementation(connector);
     this.syncStatusListenerDisposer = this.syncStreamImplementation.registerListener({
       statusChanged: (status) => {
@@ -140,20 +139,20 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     await this.disconnect();
 
     // TODO DB name, verify this is necessary with extension
-    await this.database.transaction(async (tx) => {
-      await tx.executeAsync('DELETE FROM ps_oplog WHERE 1');
-      await tx.executeAsync('DELETE FROM ps_crud WHERE 1');
-      await tx.executeAsync('DELETE FROM ps_buckets WHERE 1');
+    await this.database.writeTransaction(async (tx) => {
+      await tx.execute('DELETE FROM ps_oplog WHERE 1');
+      await tx.execute('DELETE FROM ps_crud WHERE 1');
+      await tx.execute('DELETE FROM ps_buckets WHERE 1');
 
-      const existingTableRows = await tx.executeAsync(
+      const existingTableRows = await tx.execute(
         "SELECT name FROM sqlite_master WHERE type='table' AND name GLOB 'ps_data_*'"
       );
 
       if (!existingTableRows.rows.length) {
         return;
       }
       for (const row of existingTableRows.rows._array) {
-        await tx.executeAsync(`DELETE FROM ${row.name} WHERE 1`);
+        await tx.execute(`DELETE FROM ${row.name} WHERE 1`);
       }
     });
   }
@@ -179,14 +178,12 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
   async getUploadQueueStats(includeSize?: boolean): Promise<UploadQueueStats> {
     return this.readTransaction(async (tx) => {
       if (includeSize) {
-        const result = await tx.executeAsync(
-          'SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM ps_crud'
-        );
+        const result = await tx.execute('SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM ps_crud');
 
         const row = result.rows.item(0);
         return new UploadQueueStats(row?.count ?? 0, row?.size ?? 0);
       } else {
-        const result = await tx.executeAsync('SELECT count(*) as count FROM ps_crud');
+        const result = await tx.execute('SELECT count(*) as count FROM ps_crud');
         const row = result.rows.item(0);
         return new UploadQueueStats(row?.count ?? 0);
       }
@@ -211,7 +208,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * and a single transaction may be split over multiple batches.
    */
   async getCrudBatch(limit: number): Promise<CrudBatch | null> {
-    const result = await this.database.executeAsync('SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?', [
+    const result = await this.database.execute('SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?', [
       limit + 1
     ]);
 
@@ -229,11 +226,11 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
     const last = all[all.length - 1];
     return new CrudBatch(all, haveMore, async (writeCheckpoint?: string) => {
       await this.writeTransaction(async (tx) => {
-        await tx.executeAsync('DELETE FROM ps_crud WHERE id <= ?', [last.clientId]);
-        if (writeCheckpoint != null && (await tx.executeAsync('SELECT 1 FROM ps_crud LIMIT 1')) == null) {
-          await tx.executeAsync("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [writeCheckpoint]);
+        await tx.execute('DELETE FROM ps_crud WHERE id <= ?', [last.clientId]);
+        if (writeCheckpoint != null && (await tx.execute('SELECT 1 FROM ps_crud LIMIT 1')) == null) {
+          await tx.execute("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [writeCheckpoint]);
         } else {
-          await tx.executeAsync("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [
+          await tx.execute("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [
             this.bucketStorageAdapter.getMaxOpId()
           ]);
         }
@@ -256,7 +253,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   async getNextCrudTransaction(): Promise<CrudTransaction> {
     return await this.readTransaction(async (tx) => {
-      const first = await tx.executeAsync('SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1');
+      const first = await tx.execute('SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1');
 
       if (!first.rows.length) {
         return null;
@@ -267,9 +264,7 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       if (!txId) {
         all = [CrudEntry.fromRow(first.rows.item(0))];
       } else {
-        const result = await tx.executeAsync('SELECT id, tx_id, data FROM ps_crud WHERE tx_id = ? ORDER BY id ASC', [
-          txId
-        ]);
+        const result = await tx.execute('SELECT id, tx_id, data FROM ps_crud WHERE tx_id = ? ORDER BY id ASC', [txId]);
         all = result.rows._array.map((row) => CrudEntry.fromRow(row));
       }
 
@@ -279,14 +274,14 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
         all,
         async (writeCheckpoint?: string) => {
           await this.writeTransaction(async (tx) => {
-            await tx.executeAsync('DELETE FROM ps_crud WHERE id <= ?', [last.clientId]);
+            await tx.execute('DELETE FROM ps_crud WHERE id <= ?', [last.clientId]);
             if (writeCheckpoint) {
-              const check = await tx.executeAsync('SELECT 1 FROM ps_crud LIMIT 1');
+              const check = await tx.execute('SELECT 1 FROM ps_crud LIMIT 1');
               if (!check.rows?.length) {
-                await tx.executeAsync("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [writeCheckpoint]);
+                await tx.execute("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [writeCheckpoint]);
               }
             } else {
-              await tx.executeAsync("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [
+              await tx.execute("UPDATE ps_buckets SET target_op = ? WHERE name='$local'", [
                 this.bucketStorageAdapter.getMaxOpId()
               ]);
             }
@@ -301,36 +296,32 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    * Execute a statement and optionally return results
    */
   async execute(sql: string, parameters?: any[]) {
-    const res = await this.writeLock((tx) => tx.executeAsync(sql, parameters));
-    return res;
+    await this.initialized;
+    return this.database.execute(sql, parameters);
   }
 
   /**
    *  Execute a read-only query and return results
    */
   async getAll<T>(sql: string, parameters?: any[]): Promise<T[]> {
-    const res = await this.readTransaction((tx) => tx.executeAsync(sql, parameters));
-    return res.rows?._array ?? [];
+    await this.initialized;
+    return this.database.getAll(sql, parameters);
   }
 
   /**
    * Execute a read-only query and return the first result, or null if the ResultSet is empty.
    */
   async getOptional<T>(sql: string, parameters?: any[]): Promise<T | null> {
-    const res = await this.readTransaction((tx) => tx.executeAsync(sql, parameters));
-    return res.rows?.item(0) ?? null;
+    await this.initialized;
+    return this.database.getOptional(sql, parameters);
   }
 
   /**
    * Execute a read-only query and return the first result, error if the ResultSet is empty.
    */
   async get<T>(sql: string, parameters?: any[]): Promise<T> {
-    const res = await this.readTransaction((tx) => tx.executeAsync(sql, parameters));
-    const first = res.rows?.item(0);
-    if (!first) {
-      throw new Error('Result set is empty');
-    }
-    return first;
+    await this.initialized;
+    return this.database.get(sql, parameters);
   }
 
   /**
@@ -358,28 +349,26 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
 
   async readTransaction<T>(callback: (tx: Transaction) => Promise<T>, lockTimeout?: number): Promise<T> {
     await this.initialized;
-    return this.runLockedTransaction(
-      AbstractPowerSyncDatabase.transactionMutex,
+    return this.database.readTransaction(
       async (tx) => {
-        const res = await callback(tx);
-        await tx.rollbackAsync();
+        const res = await callback({ ...tx });
+        await tx.rollback();
         return res;
       },
-      lockTimeout
+      { timeoutMs: lockTimeout }
     );
   }
 
   async writeTransaction<T>(callback: (tx: Transaction) => Promise<T>, lockTimeout?: number): Promise<T> {
     await this.initialized;
-    return this.runLockedTransaction(
-      AbstractPowerSyncDatabase.transactionMutex,
+    return this.database.writeTransaction(
       async (tx) => {
         const res = await callback(tx);
-        await tx.commitAsync();
+        await tx.commit();
         _.defer(() => this.syncStreamImplementation?.triggerCrudUpload());
         return res;
       },
-      lockTimeout
+      { timeoutMs: lockTimeout }
     );
   }
 
@@ -456,27 +445,4 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
       return () => dispose();
     });
   }
-
-  private runLockedTransaction<T>(
-    mutex: Mutex,
-    callback: (tx: Transaction) => Promise<T>,
-    lockTimeout?: number
-  ): Promise<T> {
-    return mutexRunExclusive(
-      mutex,
-      () => {
-        return new Promise<T>(async (resolve, reject) => {
-          try {
-            await this.database.transaction(async (tx) => {
-              const r = await callback(tx);
-              resolve(r);
-            });
-          } catch (ex) {
-            reject(ex);
-          }
-        });
-      },
-      { timeoutMs: lockTimeout }
-    );
-  }
 }