Make the flushing of reactive queries async

cpp/DBHostObject.cpp

@@ -17,11 +17,12 @@ namespace jsi = facebook::jsi;
 namespace react = facebook::react;
 
 #ifdef OP_SQLITE_USE_LIBSQL
-void DBHostObject::flush_pending_reactive_queries() {
-  // intentionally left blank
+void DBHostObject::flush_pending_reactive_queries(std::shared_ptr<jsi::Value> resolve) {
+    invoker->invokeAsync(
+        [this, resolve]() { resolve->asObject(rt).asFunction(rt).call(rt, {}); });
 }
 #else
-void DBHostObject::flush_pending_reactive_queries() {
+void DBHostObject::flush_pending_reactive_queries(std::shared_ptr<jsi::Value> resolve) {
   for (const auto &query_ptr : pending_reactive_queries) {
     auto query = query_ptr.get();
 
@@ -50,6 +51,9 @@ void DBHostObject::flush_pending_reactive_queries() {
           });
     }
   }
+
+  invoker->invokeAsync(
+      [this, resolve]() { resolve->asObject(rt).asFunction(rt).call(rt, {}); });
 }
 
 void DBHostObject::auto_register_update_hook() {
@@ -302,8 +306,7 @@ void DBHostObject::create_jsi_functions() {
     std::vector<JSVariant> params;
 
     if (count == 2) {
-      const jsi::Value &originalParams = args[1];
-      params = to_variant_vec(rt, originalParams);
+      params = to_variant_vec(rt, args[1]);
     }
 
     auto promiseCtr = rt.global().getPropertyAsFunction(rt, "Promise");
@@ -312,7 +315,7 @@ void DBHostObject::create_jsi_functions() {
       auto reject = std::make_shared<jsi::Value>(rt, args[1]);
 
       auto task = [&rt, this, query, params = std::move(params), resolve,
-                   reject, invoker = this->invoker]() {
+                   reject]() {
         try {
           std::vector<std::vector<JSVariant>> results;
 
@@ -361,21 +364,20 @@ void DBHostObject::create_jsi_functions() {
   });
 
   auto execute = HOSTFN("execute") {
-    const std::string query = args[0].asString(rt).utf8(rt);
+    std::string query = args[0].asString(rt).utf8(rt);
     std::vector<JSVariant> params;
 
     if (count == 2) {
       params = to_variant_vec(rt, args[1]);
     }
 
     auto promiseCtr = rt.global().getPropertyAsFunction(rt, "Promise");
-        auto promise = promiseCtr.callAsConstructor(rt, HOSTFN("executor") {
+    auto promise = promiseCtr.callAsConstructor(rt, HOSTFN("executor") {
       auto resolve = std::make_shared<jsi::Value>(rt, args[0]);
       auto reject = std::make_shared<jsi::Value>(rt, args[1]);
 
       auto task = [&rt, this, query = std::move(query),
-                   params = std::move(params), resolve, reject,
-                   invoker = this->invoker]() {
+                   params = std::move(params), resolve, reject]() {
         try {
 
 #ifdef OP_SQLITE_USE_LIBSQL
@@ -419,7 +421,7 @@ void DBHostObject::create_jsi_functions() {
       return {};
         }));
 
-        return promise;
+    return promise;
   });
 
   auto execute_with_host_objects = HOSTFN("executeWithHostObjects") {
@@ -811,8 +813,20 @@ void DBHostObject::create_jsi_functions() {
 
   auto flush_pending_reactive_queries_js =
       HOSTFN("flushPendingReactiveQueries") {
-    flush_pending_reactive_queries();
-    return {};
+    auto promiseCtr = rt.global().getPropertyAsFunction(rt, "Promise");
+          auto promise = promiseCtr.callAsConstructor(rt, HOSTFN("executor") {
+      auto resolve = std::make_shared<jsi::Value>(rt, args[0]);
+
+      auto task = [&rt, this, resolve]() {
+        flush_pending_reactive_queries(resolve);
+      };
+
+      thread_pool->queueWork(task);
+
+      return {};
+              }));
+
+          return promise;
   });
 
   function_map["attach"] = std::move(attach);

cpp/DBHostObject.h

@@ -64,7 +64,7 @@ class JSI_EXPORT DBHostObject : public jsi::HostObject {
     std::set<std::shared_ptr<ReactiveQuery>> pending_reactive_queries;
   void auto_register_update_hook();
   void create_jsi_functions();
-    void flush_pending_reactive_queries();
+  void flush_pending_reactive_queries(std::shared_ptr<jsi::Value> resolve);
 
   std::unordered_map<std::string, jsi::Value> function_map;
   std::string base_path;

example/ios/Podfile.lock

@@ -10,7 +10,7 @@ PODS:
   - hermes-engine (0.74.0):
     - hermes-engine/Pre-built (= 0.74.0)
   - hermes-engine/Pre-built (0.74.0)
-  - op-sqlite (9.2.1):
+  - op-sqlite (9.2.2):
     - React
     - React-callinvoker
     - React-Core
@@ -1393,7 +1393,7 @@ SPEC CHECKSUMS:
   GCDWebServer: 2c156a56c8226e2d5c0c3f208a3621ccffbe3ce4
   glog: c5d68082e772fa1c511173d6b30a9de2c05a69a2
   hermes-engine: 6eae7edb2f563ee41d7c1f91f4f2e57c26d8a5c3
-  op-sqlite: de7f4da4de0217c70e41bf0695967070ad6561d9
+  op-sqlite: be28804d262a275f6da8fe9950c4b99580eec531
   RCT-Folly: 045d6ecaa59d826c5736dfba0b2f4083ff8d79df
   RCTDeprecation: 3ca8b6c36bfb302e1895b72cfe7db0de0c92cd47
   RCTRequired: 9fc183af555fd0c89a366c34c1ae70b7e03b1dc5

example/package.json

@@ -68,7 +68,7 @@
     "performanceMode": "2",
     "iosSqlite": false,
     "fts5": true,
-    "libsql": true,
+    "libsql": false,
     "sqliteVec": true
   }
 }

src/index.ts

@@ -178,7 +178,7 @@ export type DB = {
     callback: (response: any) => void;
   }) => () => void;
   sync: () => void;
-  flushPendingReactiveQueries: () => void;
+  flushPendingReactiveQueries: () => Promise<void>;
 };
 
 type OPSQLiteProxy = {
@@ -349,9 +349,9 @@ function enhanceDB(db: DB, options: any): DB {
           );
         }
         const result = await enhancedDb.execute('COMMIT;');
-        console.log('BEFORE FLUSH');
-        enhancedDb.flushPendingReactiveQueries();
-        console.log('AFER FLUSH');
+
+        await enhancedDb.flushPendingReactiveQueries();
+
         isFinalized = true;
         return result;
       };