feat(rivetkit): add synchronous sqlite operations

docs/content/docs/sqlite.mdx

@@ -83,6 +83,36 @@ const rows = await c.db.execute(
 );
 ```
 
+### Synchronous operations in Node.js
+
+The Node.js native runtime also provides `c.db.executeSync(...)` and `c.db.transactionSync(...)` for integrations that cannot use an asynchronous API. `executeSync(...)` accepts the same SQL and parameters as `execute(...)`.
+
+<Warning>
+**Use synchronous SQLite only when absolutely necessary.** Every synchronous operation blocks the entire Node.js runtime until SQLite finishes. This pauses the current actor and can also prevent other actors hosted by the same runtime from running. Prefer `await c.db.execute(...)` and `await c.db.transaction(...)` whenever possible.
+
+Worker-thread actor isolation is coming soon. It will allow each actor to run on its own thread so a synchronous operation in one actor does not block other actors. Even with that isolation, use the synchronous API only when an integration requires it.
+</Warning>
+
+```ts @nocheck
+const rows = c.db.executeSync(
+  "SELECT id, title FROM todos WHERE title LIKE ?",
+  `%${query}%`,
+);
+```
+
+Use `transactionSync(...)` when multiple synchronous queries must commit or roll back together. The transaction commits before `transactionSync(...)` returns and rolls back if the callback throws.
+
+```ts @nocheck
+const todoId = c.db.transactionSync((tx) => {
+  tx.executeSync("INSERT INTO todos (title) VALUES (?)", title);
+  return tx.executeSync<{ id: number }>(
+    "SELECT last_insert_rowid() AS id",
+  )[0].id;
+});
+```
+
+The callback must be synchronous and must use its `tx` value, which exposes only `executeSync(...)`. It must not return a promise. `{ name, timeout }` options are supported, matching `transaction(...)`. Synchronous operations are unavailable in WebAssembly runtimes.
+
 ### Transactions
 
 Use transactions when multiple writes must succeed or fail together.

rivetkit-typescript/packages/effect/src/Actor.test-d.ts

@@ -6,7 +6,7 @@ import {
 	Schema,
 	SchemaTransformation,
 } from "effect";
-import type { RawAccess } from "rivetkit/db";
+import type { SynchronousRawAccess } from "rivetkit/db";
 import { db } from "rivetkit/db";
 import { describe, expectTypeOf, it, test } from "@effect/vitest";
 
@@ -287,13 +287,21 @@ describe("Actor.make(...).toLayer", () => {
 			(wakeOptions) => {
 				expectTypeOf(
 					wakeOptions.rawRivetkitContext.db,
-				).toEqualTypeOf<RawAccess>();
+				).toEqualTypeOf<SynchronousRawAccess>();
 				expectTypeOf(
 					wakeOptions.rawRivetkitContext.db.transaction(
 						async () => {},
 						{ name: "effect-operation", timeout: 1_000 },
 					),
 				).toEqualTypeOf<Promise<void>>();
+				expectTypeOf(
+					wakeOptions.rawRivetkitContext.db.executeSync<{
+						count: number;
+					}>("SELECT COUNT(*) AS count"),
+				).toEqualTypeOf<{ count: number }[]>();
+				expectTypeOf(
+					wakeOptions.rawRivetkitContext.db.transactionSync(() => 42),
+				).toEqualTypeOf<number>();
 
 				return {
 					Ping: () => Effect.succeed(0),

rivetkit-typescript/packages/rivetkit-napi/index.d.ts

@@ -379,19 +379,27 @@ export declare class JsNativeDatabase {
   run(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<ExecuteResult>
   query(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<QueryResult>
   execute(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<NativeExecuteResult>
+  executeSync(sql: string, params?: Array<JsBindParam> | undefined | null): NativeExecuteResult
   executeBatch(statements: Array<JsSqliteBatchStatement>): Promise<Array<NativeExecuteResult>>
   exec(sql: string): Promise<QueryResult>
+  execSync(sql: string): QueryResult
   close(): Promise<void>
   beginTransaction(timeoutMs?: number | undefined | null, name?: string | undefined | null): Promise<JsSqliteTransaction>
+  beginTransactionSync(timeoutMs?: number | undefined | null, name?: string | undefined | null): JsSqliteTransaction
 }
 export declare class JsSqliteTransaction {
   execute(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<NativeExecuteResult>
+  executeSync(sql: string, params?: Array<JsBindParam> | undefined | null): NativeExecuteResult
   exec(sql: string): Promise<QueryResult>
+  execSync(sql: string): QueryResult
   commit(): Promise<void>
+  commitSync(): void
   rollback(): Promise<void>
+  rollbackSync(): void
 }
 export declare class JsActorStateTransaction {
   execute(sql: string, params?: Array<JsBindParam> | undefined | null): Promise<NativeExecuteResult>
+  executeSync(sql: string, params?: Array<JsBindParam> | undefined | null): NativeExecuteResult
   commit(payload: StateDeltaPayload): Promise<void>
   rollback(): Promise<void>
 }

rivetkit-typescript/packages/rivetkit-napi/src/database.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{future::Future, time::Duration};
 
 use crate::actor_context::{StateDeltaPayload, state_deltas_from_payload};
 use napi::bindgen_prelude::Buffer;
@@ -176,6 +176,18 @@ impl JsNativeDatabase {
 		Ok(core_execute_result_to_js(result))
 	}
 
+	#[napi]
+	pub fn execute_sync(
+		&self,
+		sql: String,
+		params: Option<Vec<JsBindParam>>,
+	) -> napi::Result<NativeExecuteResult> {
+		let params = params.map(js_bind_params_to_core).transpose()?;
+		let db = self.db.clone();
+		wait_for_runtime(async move { db.execute(sql, params).await })
+			.map(core_execute_result_to_js)
+	}
+
 	#[napi]
 	pub async fn execute_batch(
 		&self,
@@ -196,6 +208,12 @@ impl JsNativeDatabase {
 		Ok(core_query_result_to_js(result))
 	}
 
+	#[napi]
+	pub fn exec_sync(&self, sql: String) -> napi::Result<QueryResult> {
+		let db = self.db.clone();
+		wait_for_runtime(async move { db.exec(sql).await }).map(core_query_result_to_js)
+	}
+
 	#[napi]
 	pub async fn close(&self) -> napi::Result<()> {
 		self.db.close().await.map_err(crate::napi_anyhow_error)
@@ -215,6 +233,21 @@ impl JsNativeDatabase {
 			.map_err(crate::napi_anyhow_error)?;
 		Ok(JsSqliteTransaction { transaction })
 	}
+
+	#[napi]
+	pub fn begin_transaction_sync(
+		&self,
+		timeout_ms: Option<f64>,
+		name: Option<String>,
+	) -> napi::Result<JsSqliteTransaction> {
+		let timeout = timeout_ms.map(transaction_timeout).transpose()?;
+		let db = self.db.clone();
+		let transaction =
+			wait_for_runtime(
+				async move { db.begin_named_transaction(name.as_deref(), timeout).await },
+			)?;
+		Ok(JsSqliteTransaction { transaction })
+	}
 }
 
 #[napi]
@@ -233,6 +266,18 @@ impl JsSqliteTransaction {
 			.map_err(crate::napi_anyhow_error)
 	}
 
+	#[napi]
+	pub fn execute_sync(
+		&self,
+		sql: String,
+		params: Option<Vec<JsBindParam>>,
+	) -> napi::Result<NativeExecuteResult> {
+		let params = params.map(js_bind_params_to_core).transpose()?;
+		let transaction = self.transaction.clone();
+		wait_for_runtime(async move { transaction.execute(sql, params).await })
+			.map(core_execute_result_to_js)
+	}
+
 	#[napi]
 	pub async fn exec(&self, sql: String) -> napi::Result<QueryResult> {
 		self.transaction
@@ -242,6 +287,12 @@ impl JsSqliteTransaction {
 			.map_err(crate::napi_anyhow_error)
 	}
 
+	#[napi]
+	pub fn exec_sync(&self, sql: String) -> napi::Result<QueryResult> {
+		let transaction = self.transaction.clone();
+		wait_for_runtime(async move { transaction.exec(sql).await }).map(core_query_result_to_js)
+	}
+
 	#[napi]
 	pub async fn commit(&self) -> napi::Result<()> {
 		self.transaction
@@ -250,13 +301,25 @@ impl JsSqliteTransaction {
 			.map_err(crate::napi_anyhow_error)
 	}
 
+	#[napi]
+	pub fn commit_sync(&self) -> napi::Result<()> {
+		let transaction = self.transaction.clone();
+		wait_for_runtime(async move { transaction.commit().await })
+	}
+
 	#[napi]
 	pub async fn rollback(&self) -> napi::Result<()> {
 		self.transaction
 			.rollback()
 			.await
 			.map_err(crate::napi_anyhow_error)
 	}
+
+	#[napi]
+	pub fn rollback_sync(&self) -> napi::Result<()> {
+		let transaction = self.transaction.clone();
+		wait_for_runtime(async move { transaction.rollback().await })
+	}
 }
 
 #[napi]
@@ -275,6 +338,18 @@ impl JsActorStateTransaction {
 			.map_err(crate::napi_anyhow_error)
 	}
 
+	#[napi]
+	pub fn execute_sync(
+		&self,
+		sql: String,
+		params: Option<Vec<JsBindParam>>,
+	) -> napi::Result<NativeExecuteResult> {
+		let params = params.map(js_bind_params_to_core).transpose()?;
+		let transaction = self.transaction.clone();
+		wait_for_runtime(async move { transaction.execute(sql, params).await })
+			.map(core_execute_result_to_js)
+	}
+
 	#[napi]
 	pub async fn commit(&self, payload: StateDeltaPayload) -> napi::Result<()> {
 		self.transaction
@@ -292,6 +367,23 @@ impl JsActorStateTransaction {
 	}
 }
 
+fn wait_for_runtime<T, F>(future: F) -> napi::Result<T>
+where
+	F: Future<Output = anyhow::Result<T>>,
+{
+	let runtime = tokio::runtime::Handle::try_current().map_err(|error| {
+		napi_anyhow_error(
+			crate::NapiInvalidState {
+				state: "runtime".to_owned(),
+				reason: format!("cannot run synchronous SQLite operation: {error}"),
+			}
+			.build(),
+		)
+	})?;
+	// NAPI-RS enters its multithreaded runtime before invoking synchronous exports.
+	tokio::task::block_in_place(|| runtime.block_on(future)).map_err(crate::napi_anyhow_error)
+}
+
 pub(crate) fn transaction_timeout(timeout_ms: f64) -> napi::Result<Duration> {
 	if !timeout_ms.is_finite() || timeout_ms <= 0.0 {
 		return Err(napi_anyhow_error(
@@ -388,3 +480,20 @@ fn column_value_to_json(value: ColumnValue) -> serde_json::Value {
 		}
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	#[test]
+	fn synchronous_wait_uses_the_active_multithreaded_runtime() {
+		let runtime = tokio::runtime::Builder::new_multi_thread()
+			.enable_all()
+			.build()
+			.expect("runtime should build");
+		let _guard = runtime.enter();
+
+		let result = super::wait_for_runtime(async { Ok::<_, anyhow::Error>(42) })
+			.expect("future should complete");
+
+		assert_eq!(result, 42);
+	}
+}

rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts

@@ -254,6 +254,63 @@ export const dbActorRaw = actor({
 			);
 			return results[0].count;
 		},
+		synchronousQueries: async (c, value: string) => {
+			c.db.executeSync(
+				"INSERT INTO test_data (value, payload, created_at) VALUES (?, ?, ?)",
+				value,
+				"",
+				Date.now(),
+			);
+			const selected = c.db.executeSync<{ value: string }>(
+				"SELECT value FROM test_data WHERE value = ?",
+				value,
+			);
+			const multiStatementValues = c.db.executeSync<{ value: number }>(
+				"SELECT 1 AS value; SELECT 2 AS value",
+			);
+			const transactionCount = c.db.transactionSync((tx) => {
+				tx.executeSync(
+					"INSERT INTO test_data (value, payload, created_at) VALUES (?, ?, ?)",
+					`${value}-committed`,
+					"",
+					Date.now(),
+				);
+				return tx.executeSync<{ count: number }>(
+					"SELECT COUNT(*) AS count FROM test_data",
+				)[0]?.count;
+			});
+			const rolledBackValue = `${value}-rolled-back`;
+			try {
+				c.db.transactionSync((tx) => {
+					tx.executeSync(
+						"INSERT INTO test_data (value, payload, created_at) VALUES (?, ?, ?)",
+						rolledBackValue,
+						"",
+						Date.now(),
+					);
+					throw new Error("rollback sync transaction");
+				});
+			} catch (error) {
+				if (
+					!(error instanceof Error) ||
+					error.message !== "rollback sync transaction"
+				) {
+					throw error;
+				}
+			}
+			const rollbackCount = c.db.executeSync<{ count: number }>(
+				"SELECT COUNT(*) AS count FROM test_data WHERE value = ?",
+				rolledBackValue,
+			)[0]?.count;
+			return {
+				value: selected[0]?.value,
+				multiStatementValues: multiStatementValues.map(
+					(row) => row.value,
+				),
+				transactionCount,
+				rollbackCount,
+			};
+		},
 		insertMany: async (c, count: number) => {
 			if (count <= 0) {
 				return { count: 0 };

rivetkit-typescript/packages/rivetkit/src/agent-os/actor/index.ts

@@ -1,7 +1,10 @@
 import type { AgentOsOptions, MountConfig } from "@rivet-dev/agent-os-core";
 import { AgentOs, createInMemoryFileSystem } from "@rivet-dev/agent-os-core";
 import { type ActorDefinition, actor, event } from "@/actor/mod";
-import type { DatabaseProvider, RawAccess } from "@/common/database/config";
+import type {
+	DatabaseProvider,
+	SynchronousRawAccess,
+} from "@/common/database/config";
 import { db } from "@/common/database/mod";
 import {
 	type AgentOsActorConfig,
@@ -146,7 +149,7 @@ export function agentOs<TConnParams = undefined>(
 	undefined,
 	AgentOsActorVars,
 	undefined,
-	DatabaseProvider<RawAccess>,
+	DatabaseProvider<SynchronousRawAccess>,
 	{
 		sessionEvent: typeof sessionEventToken;
 		permissionRequest: typeof permissionRequestToken;
@@ -182,7 +185,7 @@ export function agentOs<TConnParams = undefined>(
 		undefined,
 		AgentOsActorVars,
 		undefined,
-		DatabaseProvider<RawAccess>,
+		DatabaseProvider<SynchronousRawAccess>,
 		{
 			sessionEvent: typeof sessionEventToken;
 			permissionRequest: typeof permissionRequestToken;