feat: improve concurrency features

Author: chrispader

package/src/DatabaseQueue.ts

@@ -0,0 +1,128 @@
+import NitroSQLiteError from './NitroSQLiteError'
+
+export interface QueuedOperation {
+  /**
+   * Starts the operation
+   */
+  start: () => void
+}
+
+export type DatabaseQueue = {
+  queue: QueuedOperation[]
+  inProgress: boolean
+}
+
+const databaseQueues = new Map<string, DatabaseQueue>()
+
+export function openDatabaseQueue(dbName: string) {
+  if (isDatabaseOpen(dbName)) {
+    throw new NitroSQLiteError(
+      `Database ${dbName} is already open. There is already a connection to the database.`
+    )
+  }
+
+  databaseQueues.set(dbName, { queue: [], inProgress: false })
+}
+
+export function closeDatabaseQueue(dbName: string) {
+  const queue = getDatabaseQueue(dbName)
+
+  if (queue.inProgress || queue.queue.length > 0) {
+    console.warn(
+      `Database queue for ${dbName} has operations in the queue. Closing anyway.`
+    )
+  }
+
+  databaseQueues.delete(dbName)
+}
+
+export function isDatabaseOpen(dbName: string) {
+  return databaseQueues.has(dbName)
+}
+
+export function throwIfDatabaseIsNotOpen(dbName: string) {
+  if (!isDatabaseOpen(dbName))
+    throw new NitroSQLiteError(
+      `Database ${dbName} is not open. There is no connection to the database.`
+    )
+}
+
+function getDatabaseQueue(dbName: string) {
+  throwIfDatabaseIsNotOpen(dbName)
+
+  const queue = databaseQueues.get(dbName)!
+  return queue
+}
+
+export function openDatabase(dbName: string) {
+  databaseQueues.set(dbName, { queue: [], inProgress: false })
+}
+
+export function closeDatabase(dbName: string) {
+  databaseQueues.delete(dbName)
+}
+
+export function queueOperationAsync<
+  OperationCallback extends () => Promise<Result>,
+  Result = void,
+>(dbName: string, callback: OperationCallback) {
+  const queue = getDatabaseQueue(dbName)
+
+  return new Promise<Result>((resolve, reject) => {
+    const operation: QueuedOperation = {
+      start: async () => {
+        try {
+          const result = await callback()
+          resolve(result)
+        } catch (error) {
+          reject(error)
+        } finally {
+          queue.inProgress = false
+          startOperationAsync(dbName)
+        }
+      },
+    }
+
+    queue.queue.push(operation)
+    startOperationAsync(dbName)
+  })
+}
+
+function startOperationAsync(dbName: string) {
+  const queue = getDatabaseQueue(dbName)
+
+  if (queue.inProgress) {
+    // Operation is already in process bail out
+    return
+  }
+
+  if (queue.queue.length > 0) {
+    queue.inProgress = true
+
+    const operation = queue.queue.shift()!
+    setImmediate(() => {
+      operation.start()
+    })
+  }
+}
+
+export function startOperationSync<
+  OperationCallback extends () => Result,
+  Result = void,
+>(dbName: string, callback: OperationCallback) {
+  const queue = getDatabaseQueue(dbName)
+
+  // Database is busy - cannot execute synchronously
+  if (queue.inProgress || queue.queue.length > 0) {
+    throw new NitroSQLiteError(
+      `Cannot run synchronous operation on database. Database ${dbName} is busy with another operation.`
+    )
+  }
+
+  // Execute synchronously
+  queue.inProgress = true
+  const result = callback()
+  queue.inProgress = false
+
+  return result
+}

package/src/concurrency.ts

@@ -3,78 +3,62 @@ import {
   replaceWithNativeNullValue,
 } from '../nullHandling'
 import { HybridNitroSQLite } from '../nitro'
-import { locks, type QueuedOperation } from '../concurrency'
+import {
+  queueOperationAsync,
+  startOperationSync,
+  throwIfDatabaseIsNotOpen,
+} from '../DatabaseQueue'
 import type {
   NativeSQLiteQueryParams,
   BatchQueryResult,
   BatchQueryCommand,
   NativeBatchQueryCommand,
 } from '../types'
-import { startNextOperation } from './transaction'
+import NitroSQLiteError from '../NitroSQLiteError'
 
 export function executeBatch(
   dbName: string,
   commands: BatchQueryCommand[]
 ): BatchQueryResult {
-  if (locks[dbName] == null)
-    throw Error(`Nitro SQLite Error: No lock found on db: ${dbName}`)
+  throwIfDatabaseIsNotOpen(dbName)
 
   const transformedCommands = isSimpleNullHandlingEnabled()
     ? toNativeBatchQueryCommands(commands)
     : (commands as NativeBatchQueryCommand[])
 
-  // If lock is immediately available, execute synchronously
-  if (!locks[dbName].inProgress && locks[dbName].queue.length === 0) {
-    locks[dbName].inProgress = true
-    try {
-      const result = HybridNitroSQLite.executeBatch(dbName, transformedCommands)
-      return result
-    } finally {
-      locks[dbName].inProgress = false
-      startNextOperation(dbName)
-    }
+  try {
+    return startOperationSync(dbName, () =>
+      HybridNitroSQLite.executeBatch(dbName, transformedCommands)
+    )
+  } catch (error) {
+    throw NitroSQLiteError.fromError(error)
   }
-
-  // Lock is busy - cannot execute synchronously
-  throw Error(
-    `Nitro SQLite Error: Database ${dbName} is busy with another operation. Use executeBatchAsync for queued execution.`
-  )
 }
 
 export async function executeBatchAsync(
   dbName: string,
   commands: BatchQueryCommand[]
 ): Promise<BatchQueryResult> {
-  if (locks[dbName] == null)
-    throw Error(`Nitro SQLite Error: No lock found on db: ${dbName}`)
+  throwIfDatabaseIsNotOpen(dbName)
 
   const transformedCommands = isSimpleNullHandlingEnabled()
     ? toNativeBatchQueryCommands(commands)
     : (commands as NativeBatchQueryCommand[])
 
-  async function run() {
-    try {
-      const result = await HybridNitroSQLite.executeBatchAsync(
-        dbName,
-        transformedCommands
-      )
-      return result
-    } finally {
-      locks[dbName]!.inProgress = false
-      startNextOperation(dbName)
-    }
+  try {
+    return queueOperationAsync(dbName, async () => {
+      try {
+        return await HybridNitroSQLite.executeBatchAsync(
+          dbName,
+          transformedCommands
+        )
+      } catch (error) {
+        throw NitroSQLiteError.fromError(error)
+      }
+    })
+  } catch (error) {
+    throw NitroSQLiteError.fromError(error)
   }
-
-  return new Promise<BatchQueryResult>((resolve, reject) => {
-    const operation: QueuedOperation = {
-      start: () => {
-        run().then(resolve).catch(reject)
-      },
-    }
-
-    locks[dbName]?.queue.push(operation)
-    startNextOperation(dbName)
-  })
 }
 
 function toNativeBatchQueryCommands(

package/src/operations/executeBatch.ts

@@ -1,4 +1,4 @@
-import { locks, HybridNitroSQLite } from '../nitro'
+import { HybridNitroSQLite } from '../nitro'
 import { transaction } from './transaction'
 import type {
   BatchQueryCommand,
@@ -11,6 +11,8 @@ import type {
 } from '../types'
 import { execute, executeAsync } from './execute'
 import { executeBatch, executeBatchAsync } from './executeBatch'
+import NitroSQLiteError from '../NitroSQLiteError'
+import { closeDatabaseQueue, openDatabaseQueue } from '../DatabaseQueue'
 
 export function open(
   options: NitroSQLiteConnectionOptions
@@ -45,15 +47,19 @@ export function open(
 }
 
 export function openDb(dbName: string, location?: string) {
-  HybridNitroSQLite.open(dbName, location)
-
-  locks[dbName] = {
-    queue: [],
-    inProgress: false,
+  try {
+    HybridNitroSQLite.open(dbName, location)
+    openDatabaseQueue(dbName)
+  } catch (error) {
+    throw NitroSQLiteError.fromError(error)
   }
 }
 
 export function close(dbName: string) {
-  HybridNitroSQLite.close(dbName)
-  delete locks[dbName]
+  try {
+    HybridNitroSQLite.close(dbName)
+    closeDatabaseQueue(dbName)
+  } catch (error) {
+    throw NitroSQLiteError.fromError(error)
+  }
 }