sqlite: implement async Database.prototype.exec

Author: geeksilva97

src/node_sqlite.cc

@@ -424,6 +424,52 @@ class CustomAggregate {
   Global<Function> result_fn_;
 };
 
+template <typename T>
+class SQLiteAsyncTask : public ThreadPoolWork {
+ public:
+  explicit SQLiteAsyncTask(
+      Environment* env,
+      Database* db,
+      Local<Promise::Resolver> resolver,
+      std::function<T()> work,
+      std::function<void(T, Local<Promise::Resolver>)> after)
+      : ThreadPoolWork(env, "node_sqlite_async_work"),
+        env_(env),
+        db_(db),
+        work_(work),
+        after_(after) {
+    resolver_.Reset(env->isolate(), resolver);
+  }
+
+  void DoThreadPoolWork() override {
+    if (work_) {
+      result_ = work_();
+    }
+  }
+
+  void AfterThreadPoolWork(int status) override {
+    Isolate* isolate = env_->isolate();
+    HandleScope handle_scope(isolate);
+    Local<Promise::Resolver> resolver =
+        Local<Promise::Resolver>::New(isolate, resolver_);
+
+    if (after_) {
+      after_(result_, resolver);
+      Finalize();
+    }
+  }
+
+  void Finalize() { db_->RemoveAsyncTask(this); }
+
+ private:
+  Environment* env_;
+  Database* db_;
+  Global<Promise::Resolver> resolver_;
+  std::function<T()> work_ = nullptr;
+  std::function<void(T, Local<Promise::Resolver>)> after_ = nullptr;
+  T result_;
+};
+
 class BackupJob : public ThreadPoolWork {
  public:
   explicit BackupJob(Environment* env,
@@ -650,10 +696,10 @@ void UserDefinedFunction::xDestroy(void* self) {
 }
 
 Database::Database(Environment* env,
-                           Local<Object> object,
-                           DatabaseOpenConfiguration&& open_config,
-                           bool open,
-                           bool allow_load_extension)
+                   Local<Object> object,
+                   DatabaseOpenConfiguration&& open_config,
+                   bool open,
+                   bool allow_load_extension)
     : BaseObject(env, object), open_config_(std::move(open_config)) {
   MakeWeak();
   connection_ = nullptr;
@@ -674,6 +720,14 @@ void Database::RemoveBackup(BackupJob* job) {
   backups_.erase(job);
 }
 
+void Database::AddAsyncTask(ThreadPoolWork* async_task) {
+  async_tasks_.insert(async_task);
+}
+
+void Database::RemoveAsyncTask(ThreadPoolWork* async_task) {
+  async_tasks_.erase(async_task);
+}
+
 void Database::DeleteSessions() {
   // all attached sessions need to be deleted before the database is closed
   // https://www.sqlite.org/session/sqlite3session_create.html
@@ -685,6 +739,7 @@ void Database::DeleteSessions() {
 
 Database::~Database() {
   FinalizeBackups();
+  async_tasks_.clear();
 
   if (IsOpen()) {
     FinalizeStatements();
@@ -841,7 +896,8 @@ std::optional<std::string> ValidateDatabasePath(Environment* env,
   return std::nullopt;
 }
 
-inline void DatabaseNew(const FunctionCallbackInfo<Value>& args, bool async = true) {
+inline void DatabaseNew(const FunctionCallbackInfo<Value>& args,
+                        bool async = true) {
   Environment* env = Environment::GetCurrent(args);
   if (!args.IsConstructCall()) {
     THROW_ERR_CONSTRUCT_CALL_REQUIRED(env);
@@ -1053,8 +1109,7 @@ void Database::IsOpenGetter(const FunctionCallbackInfo<Value>& args) {
   args.GetReturnValue().Set(db->IsOpen());
 }
 
-void Database::IsTransactionGetter(
-    const FunctionCallbackInfo<Value>& args) {
+void Database::IsTransactionGetter(const FunctionCallbackInfo<Value>& args) {
   Database* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
@@ -1116,13 +1171,47 @@ void Database::Exec(const FunctionCallbackInfo<Value>& args) {
     return;
   }
 
+  Isolate* isolate = env->isolate();
+  auto sql = Utf8Value(isolate, args[0].As<String>()).ToString();
+  auto task = [sql, db]() -> int {
+    return sqlite3_exec(
+        db->connection_, sql.c_str(), nullptr, nullptr, nullptr);
+  };
+
   if (db->open_config_.get_async()) {
-    // TODO(geeksilva97): Support async by returning a Promise
-    std::cout << "This is async" << std::endl;
+    Local<Promise::Resolver> resolver;
+    if (!Promise::Resolver::New(env->context()).ToLocal(&resolver)) {
+      return;
+    }
+
+    auto after = [db, env, isolate](int exec_result,
+                                    Local<Promise::Resolver> resolver) {
+      if (exec_result != SQLITE_OK) {
+        if (db->ShouldIgnoreSQLiteError()) {
+          db->SetIgnoreNextSQLiteError(false);
+          return;
+        }
+
+        Local<Object> e;
+        if (!CreateSQLiteError(isolate, db->Connection()).ToLocal(&e)) {
+          return;
+        }
+
+        resolver->Reject(env->context(), e).FromJust();
+        return;
+      }
+
+      resolver->Resolve(env->context(), Undefined(env->isolate())).FromJust();
+    };
+
+    auto* work = new SQLiteAsyncTask<int>(env, db, resolver, task, after);
+    work->ScheduleWork();
+    db->AddAsyncTask(work);
+    args.GetReturnValue().Set(resolver->GetPromise());
+    return;
   }
 
-  Utf8Value sql(env->isolate(), args[0].As<String>());
-  int r = sqlite3_exec(db->connection_, *sql, nullptr, nullptr, nullptr);
+  int r = task();
   CHECK_ERROR_OR_THROW(env->isolate(), db, r, SQLITE_OK, void());
 }
 
@@ -1776,8 +1865,7 @@ void Database::ApplyChangeset(const FunctionCallbackInfo<Value>& args) {
   THROW_ERR_SQLITE_ERROR(env->isolate(), r);
 }
 
-void Database::EnableLoadExtension(
-    const FunctionCallbackInfo<Value>& args) {
+void Database::EnableLoadExtension(const FunctionCallbackInfo<Value>& args) {
   Database* db;
   ASSIGN_OR_RETURN_UNWRAP(&db, args.This());
   Environment* env = Environment::GetCurrent(args);
@@ -2470,8 +2558,9 @@ Local<FunctionTemplate> StatementSync::GetConstructorTemplate(
   return tmpl;
 }
 
-BaseObjectPtr<StatementSync> StatementSync::Create(
-    Environment* env, BaseObjectPtr<Database> db, sqlite3_stmt* stmt) {
+BaseObjectPtr<StatementSync> StatementSync::Create(Environment* env,
+                                                   BaseObjectPtr<Database> db,
+                                                   sqlite3_stmt* stmt) {
   Local<Object> obj;
   if (!GetConstructorTemplate(env)
            ->InstanceTemplate()
@@ -2730,12 +2819,15 @@ void DefineConstants(Local<Object> target) {
   NODE_DEFINE_CONSTANT(target, SQLITE_CHANGESET_FOREIGN_KEY);
 }
 
-void DefineAsyncInterface(Isolate* isolate, Local<Object> target, Local<Context> context) {
+void DefineAsyncInterface(Isolate* isolate,
+                          Local<Object> target,
+                          Local<Context> context) {
   Local<FunctionTemplate> db_async_tmpl =
       NewFunctionTemplate(isolate, Database::NewAsync);
   db_async_tmpl->InstanceTemplate()->SetInternalFieldCount(
       Database::kInternalFieldCount);
 
+  SetProtoMethod(isolate, db_async_tmpl, "close", Database::Close);
   SetProtoMethod(isolate, db_async_tmpl, "exec", Database::Exec);
   SetConstructorFunction(context, target, "Database", db_async_tmpl);
 }

src/node_sqlite.h

@@ -6,6 +6,7 @@
 #include "base_object.h"
 #include "node_mem.h"
 #include "sqlite3.h"
+#include "threadpoolwork-inl.h"
 #include "util.h"
 
 #include <map>
@@ -114,6 +115,8 @@ class Database : public BaseObject {
   void FinalizeStatements();
   void RemoveBackup(BackupJob* backup);
   void AddBackup(BackupJob* backup);
+  void AddAsyncTask(ThreadPoolWork* async_task);
+  void RemoveAsyncTask(ThreadPoolWork* async_task);
   void FinalizeBackups();
   void UntrackStatement(StatementSync* statement);
   bool IsOpen();
@@ -149,6 +152,7 @@ class Database : public BaseObject {
   bool ignore_next_sqlite_error_;
 
   std::set<BackupJob*> backups_;
+  std::set<ThreadPoolWork*> async_tasks_;
   std::set<sqlite3_session*> sessions_;
   std::unordered_set<StatementSync*> statements_;
 

test/parallel/test-sqlite-database-async.js

@@ -0,0 +1,71 @@
+import { skipIfSQLiteMissing } from '../common/index.mjs';
+import tmpdir from '../common/tmpdir.js';
+import { suite, test } from 'node:test';
+import { join } from 'node:path';
+import { Database } from 'node:sqlite';
+skipIfSQLiteMissing();
+
+tmpdir.refresh();
+
+let cnt = 0;
+function nextDb() {
+  return join(tmpdir.path, `database-${cnt++}.db`);
+}
+
+suite('Database.prototype.exec()', () => {
+  test('executes SQL', async (t) => {
+    const db = new Database(nextDb());
+    t.after(() => { db.close(); });
+    const result = await db.exec(`
+      CREATE TABLE data(
+        key INTEGER PRIMARY KEY,
+        val INTEGER
+      ) STRICT;
+      INSERT INTO data (key, val) VALUES (1, 2);
+      INSERT INTO data (key, val) VALUES (8, 9);
+    `);
+    t.assert.strictEqual(result, undefined);
+  });
+
+  test('reports errors from SQLite', async (t) => {
+    const db = new Database(nextDb());
+    t.after(() => { db.close(); });
+
+    await t.assert.rejects(db.exec('CREATE TABLEEEE'), {
+      code: 'ERR_SQLITE_ERROR',
+      message: /syntax error/,
+    });
+  });
+
+  test('throws if the URL does not have the file: scheme', (t) => {
+    t.assert.throws(() => {
+      new Database(new URL('http://example.com'));
+    }, {
+      code: 'ERR_INVALID_URL_SCHEME',
+      message: 'The URL must be of scheme file:',
+    });
+  });
+
+  test('throws if database is not open', (t) => {
+    const db = new Database(nextDb(), { open: false });
+
+    t.assert.throws(() => {
+      db.exec();
+    }, {
+      code: 'ERR_INVALID_STATE',
+      message: /database is not open/,
+    });
+  });
+
+  test('throws if sql is not a string', (t) => {
+    const db = new Database(nextDb());
+    t.after(() => { db.close(); });
+
+    t.assert.throws(() => {
+      db.exec();
+    }, {
+      code: 'ERR_INVALID_ARG_TYPE',
+      message: /The "sql" argument must be a string/,
+    });
+  });
+});