Merge 'Rn fixes 2' from Nikita Sivukhin

This PR fixes few issues in the RN bindings and also add `setup`
function which allow user to proxy tursdb logs to the application layer

Reviewed-by: Jussi Saurio <jussi.saurio@gmail.com>

Closes #5095

Author: sivukhin

bindings/react-native/cpp/TursoConnectionHostObject.cpp

@@ -166,8 +166,13 @@ jsi::Value TursoConnectionHostObject::setBusyTimeout(jsi::Runtime &rt, const jsi
 }
 
 jsi::Value TursoConnectionHostObject::close(jsi::Runtime &rt) {
+    if (!conn_) {
+        return jsi::Value::undefined();
+    }
+
     const char* error = nullptr;
     turso_status_code_t status = turso_connection_close(conn_, &error);
+    conn_ = nullptr; // Prevent destructor from calling deinit on closed connection
 
     if (status != TURSO_OK) {
         throwError(rt, error);

bindings/react-native/cpp/TursoHostObject.cpp

@@ -4,6 +4,10 @@
 
 #include <cstdio>   // For FILE, fopen, fread, fwrite, fclose, fseek, ftell, remove, rename
 #include <cstdlib>  // For additional standard library functions
+#include <cstring>  // For strrchr
+#include <unistd.h> // For fsync, close
+#include <fcntl.h>  // For open, O_RDONLY
+#include <mutex>    // For std::mutex, std::lock_guard
 
 extern "C" {
 #include <turso.h>
@@ -13,11 +17,32 @@ extern "C" {
 namespace turso
 {
 
+    /**
+     * Durable fsync: on Apple, fsync() only flushes to disk cache,
+     * so we need F_FULLFSYNC for true persistence. On Linux/Android,
+     * plain fsync() is sufficient.
+     */
+    static int durable_fsync(int fd)
+    {
+#ifdef __APPLE__
+        return fcntl(fd, F_FULLFSYNC);
+#else
+        return fsync(fd);
+#endif
+    }
+
     using namespace facebook;
 
     // Global base path for database files
     static std::string g_basePath;
 
+    // Logger callback state — the Rust tracing subscriber may fire from any thread,
+    // so we copy log data in the C callback and schedule JS execution via CallInvoker.
+    static jsi::Runtime* g_runtime = nullptr;
+    static std::shared_ptr<react::CallInvoker> g_callInvoker;
+    static std::shared_ptr<jsi::Function> g_loggerFn;
+    static std::mutex g_loggerMutex;
+
     /**
      * Normalize a database path:
      * - If path is absolute (starts with '/'), use as-is
@@ -49,12 +74,79 @@ namespace turso
         }
     }
 
+    /**
+     * Map turso_tracing_level_t enum to JS-friendly string.
+     */
+    static const char* tracingLevelToString(turso_tracing_level_t level)
+    {
+        switch (level)
+        {
+            case TURSO_TRACING_LEVEL_ERROR: return "error";
+            case TURSO_TRACING_LEVEL_WARN:  return "warn";
+            case TURSO_TRACING_LEVEL_INFO:  return "info";
+            case TURSO_TRACING_LEVEL_DEBUG: return "debug";
+            case TURSO_TRACING_LEVEL_TRACE: return "trace";
+            default:                        return "error";
+        }
+    }
+
+    /**
+     * C callback invoked by the Rust tracing subscriber (possibly from any thread).
+     * Copies all string data synchronously, then schedules a JS call on the JS thread.
+     */
+    static void turso_logger_callback(const turso_log_t *log)
+    {
+        std::lock_guard<std::mutex> lock(g_loggerMutex);
+        if (!g_loggerFn || !g_callInvoker || !g_runtime)
+        {
+            return;
+        }
+
+        // Copy all data — the turso_log_t fields are only valid during this callback.
+        std::string message = log->message ? log->message : "";
+        std::string target  = log->target  ? log->target  : "";
+        std::string file    = log->file    ? log->file    : "";
+        uint64_t timestamp  = log->timestamp;
+        size_t line         = log->line;
+        const char* level   = tracingLevelToString(log->level);
+        std::string levelStr(level);
+
+        // Prevent captures from preventing cleanup — capture shared_ptr copies
+        auto callInvoker = g_callInvoker;
+        auto loggerFn    = g_loggerFn;
+
+        callInvoker->invokeAsync(
+            [loggerFn, message = std::move(message), target = std::move(target),
+             file = std::move(file), timestamp, line, levelStr = std::move(levelStr)]
+            (jsi::Runtime &rt)
+            {
+                try
+                {
+                    jsi::Object logObj(rt);
+                    logObj.setProperty(rt, "message",   jsi::String::createFromUtf8(rt, message));
+                    logObj.setProperty(rt, "target",    jsi::String::createFromUtf8(rt, target));
+                    logObj.setProperty(rt, "file",      jsi::String::createFromUtf8(rt, file));
+                    logObj.setProperty(rt, "timestamp", static_cast<double>(timestamp));
+                    logObj.setProperty(rt, "line",      static_cast<double>(line));
+                    logObj.setProperty(rt, "level",     jsi::String::createFromUtf8(rt, levelStr));
+
+                    loggerFn->call(rt, logObj);
+                }
+                catch (...)
+                {
+                    // Logger must never crash the app — swallow all exceptions.
+                }
+            });
+    }
+
     void install(
         jsi::Runtime &rt,
         const std::shared_ptr<react::CallInvoker> &invoker,
         const char *basePath)
     {
         g_basePath = basePath ? basePath : "";
+        g_runtime = &rt;
+        g_callInvoker = invoker;
 
         // Create the module object
         jsi::Object module(rt);
@@ -163,7 +255,7 @@ namespace turso
                 sync_config.path = normalizedPath.c_str();
 
                 // remoteUrl (optional)
-                static std::string remoteUrl;
+                std::string remoteUrl;
                 if (syncConfigObj.hasProperty(rt, "remoteUrl"))
                 {
                     jsi::Value remoteUrlVal = syncConfigObj.getProperty(rt, "remoteUrl");
@@ -183,7 +275,7 @@ namespace turso
                 }
 
                 // clientName (optional)
-                static std::string clientName;
+                std::string clientName;
                 if (syncConfigObj.hasProperty(rt, "clientName"))
                 {
                     jsi::Value clientNameVal = syncConfigObj.getProperty(rt, "clientName");
@@ -274,7 +366,7 @@ namespace turso
                     sync_config.partial_bootstrap_strategy_prefix = 0;
                 }
 
-                static std::string partialBootstrapStrategyQuery;
+                std::string partialBootstrapStrategyQuery;
                 if (syncConfigObj.hasProperty(rt, "partialBootstrapStrategyQuery"))
                 {
                     jsi::Value queryVal = syncConfigObj.getProperty(rt, "partialBootstrapStrategyQuery");
@@ -328,7 +420,7 @@ namespace turso
                 }
 
                 // Remote encryption options
-                static std::string remoteEncryptionKey;
+                std::string remoteEncryptionKey;
                 if (syncConfigObj.hasProperty(rt, "remoteEncryptionKey"))
                 {
                     jsi::Value keyVal = syncConfigObj.getProperty(rt, "remoteEncryptionKey");
@@ -347,7 +439,7 @@ namespace turso
                     sync_config.remote_encryption_key = nullptr;
                 }
 
-                static std::string remoteEncryptionCipher;
+                std::string remoteEncryptionCipher;
                 if (syncConfigObj.hasProperty(rt, "remoteEncryptionCipher"))
                 {
                     jsi::Value cipherVal = syncConfigObj.getProperty(rt, "remoteEncryptionCipher");
@@ -409,8 +501,7 @@ namespace turso
 
                 jsi::Object options = args[0].asObject(rt);
 
-                // Store log level in a static variable to ensure lifetime
-                static std::string logLevelStr;
+                std::string logLevelStr;
 
                 // Get log level if provided
                 if (options.hasProperty(rt, "logLevel"))
@@ -424,6 +515,21 @@ namespace turso
 
                 turso_config_t config = {nullptr, logLevelStr.empty() ? nullptr : logLevelStr.c_str()};
 
+                // Wire up logger callback if provided
+                if (options.hasProperty(rt, "logger"))
+                {
+                    jsi::Value loggerVal = options.getProperty(rt, "logger");
+                    if (loggerVal.isObject() && loggerVal.asObject(rt).isFunction(rt))
+                    {
+                        {
+                            std::lock_guard<std::mutex> lock(g_loggerMutex);
+                            g_loggerFn = std::make_shared<jsi::Function>(
+                                loggerVal.asObject(rt).asFunction(rt));
+                        }
+                        config.logger = turso_logger_callback;
+                    }
+                }
+
                 // Call turso_setup
                 const char *error = nullptr;
                 turso_status_code_t status = turso_setup(&config, &error);
@@ -504,33 +610,35 @@ namespace turso
                 std::string path = args[0].asString(rt).utf8(rt);
                 jsi::ArrayBuffer buffer = args[1].asObject(rt).getArrayBuffer(rt);
 
-                // Write atomically using temporary file + rename
+                // Write atomically: write to temp, fsync, rename, fsync dir
                 std::string tempPath = path + ".tmp";
 
-                // Open temp file for writing
                 FILE* file = fopen(tempPath.c_str(), "wb");
                 if (!file)
                 {
                     throw jsi::JSError(rt, "Failed to open file for writing");
                 }
 
-                // Write data
                 size_t size = buffer.size(rt);
                 if (size > 0)
                 {
                     size_t written = fwrite(buffer.data(rt), 1, size, file);
-                    fclose(file);
-
                     if (written != size)
                     {
+                        fclose(file);
                         remove(tempPath.c_str());
                         throw jsi::JSError(rt, "Failed to write complete file");
                     }
                 }
-                else
+
+                // Flush to OS and sync to disk before rename
+                if (fflush(file) != 0 || durable_fsync(fileno(file)) != 0)
                 {
                     fclose(file);
+                    remove(tempPath.c_str());
+                    throw jsi::JSError(rt, "Failed to sync file to disk");
                 }
+                fclose(file);
 
                 // Atomic rename (replaces old file)
                 if (rename(tempPath.c_str(), path.c_str()) != 0)
@@ -539,6 +647,20 @@ namespace turso
                     throw jsi::JSError(rt, "Failed to rename temp file");
                 }
 
+                // Fsync parent directory to ensure rename is durable
+                std::string dirPath = path;
+                auto lastSlash = dirPath.rfind('/');
+                if (lastSlash != std::string::npos)
+                {
+                    dirPath.resize(lastSlash);
+                    int dirFd = open(dirPath.c_str(), O_RDONLY);
+                    if (dirFd >= 0)
+                    {
+                        durable_fsync(dirFd);
+                        close(dirFd);
+                    }
+                }
+
                 return jsi::Value::undefined();
             });
 
@@ -555,7 +677,10 @@ namespace turso
 
     void invalidate()
     {
-        // Cleanup if needed
+        std::lock_guard<std::mutex> lock(g_loggerMutex);
+        g_loggerFn.reset();
+        g_callInvoker.reset();
+        g_runtime = nullptr;
     }
 
 } // namespace turso

bindings/react-native/cpp/TursoStatementHostObject.cpp

@@ -286,9 +286,17 @@ jsi::Value TursoStatementHostObject::reset(jsi::Runtime &rt) {
 }
 
 jsi::Value TursoStatementHostObject::finalize(jsi::Runtime &rt) {
+    if (!stmt_) {
+        return jsi::Value(static_cast<int>(TURSO_DONE));
+    }
+
     const char* error = nullptr;
     turso_status_code_t status = turso_statement_finalize(stmt_, &error);
 
+    if (status == TURSO_DONE) {
+        stmt_ = nullptr; // Prevent destructor from calling deinit on finalized statement
+    }
+
     if (status != TURSO_DONE && status != TURSO_IO) {
         throwError(rt, error);
     }

bindings/react-native/package-lock.json

@@ -94,4 +94,4 @@
       ]
     ]
   }
-}
+}

bindings/react-native/package.json

@@ -223,8 +223,7 @@ export class Database {
     }
 
     const nativeStmt = this._connection.prepareSingle(sql);
-    // Pass extraIo callback for partial sync support
-    return new Statement(nativeStmt, this._extraIo);
+    return new Statement(nativeStmt, this._connection!, this._extraIo);
   }
 
   /**
@@ -245,12 +244,12 @@ export class Database {
     while (remaining.length > 0) {
       const result = this._connection.prepareFirst(remaining);
 
-      if (!result.statement) {
-        break; // No more statements
+      if (!result) {
+        break; // No more statements (C++ returns null when nothing to parse)
       }
 
       // Wrap in Statement to get IO handling
-      const stmt = new Statement(result.statement, this._extraIo);
+      const stmt = new Statement(result.statement, this._connection!, this._extraIo);
       try {
         // Execute - will handle IO if needed
         await stmt.run();

bindings/react-native/src/Database.ts

@@ -6,6 +6,7 @@
  */
 
 import type {
+  NativeConnection,
   NativeStatement,
   SQLiteValue,
   BindParams,
@@ -19,11 +20,13 @@ import { TursoStatus, TursoType } from './types';
  */
 export class Statement {
   private _statement: NativeStatement;
+  private _connection: NativeConnection;
   private _finalized = false;
   private _extraIo?: () => Promise<void>;
 
-  constructor(statement: NativeStatement, extraIo?: () => Promise<void>) {
+  constructor(statement: NativeStatement, connection: NativeConnection, extraIo?: () => Promise<void>) {
     this._statement = statement;
+    this._connection = connection;
     this._extraIo = extraIo;
   }
 
@@ -137,7 +140,7 @@ export class Statement {
 
     return {
       changes: result.rowsChanged,
-      lastInsertRowid: 0, // Not available from execute, would need connection
+      lastInsertRowid: this._connection ? this._connection.lastInsertRowid() : 0,
     };
   }