feat(rivetkit): add synchronous sqlite queries

Author: NathanFlurry

docs/content/docs/sqlite.mdx

@@ -83,6 +83,19 @@ const rows = await c.db.execute(
 );
 ```
 
+### Synchronous queries in Node.js
+
+The Node.js native runtime also provides `c.db.executeSync(...)` for integrations that require an immediate result. It accepts the same SQL and parameters as `execute(...)`.
+
+```ts @nocheck
+const rows = c.db.executeSync(
+  "SELECT id, title FROM todos WHERE title LIKE ?",
+  `%${query}%`,
+);
+```
+
+**Prefer `await c.db.execute(...)` for normal use.** `executeSync(...)` blocks the Node.js event loop until SQLite finishes, so a slow query also prevents the actor from handling other JavaScript work. It is unavailable in WebAssembly runtimes.
+
 ### Transactions
 
 Use transactions when multiple writes must succeed or fail together.

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

@@ -379,19 +379,24 @@ 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>
 }
 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>
   rollback(): Promise<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, sync::mpsc, time::Duration};
 
 use crate::actor_context::{StateDeltaPayload, state_deltas_from_payload};
 use napi::bindgen_prelude::Buffer;
@@ -176,6 +176,17 @@ 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 +207,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)
@@ -233,6 +250,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 +271,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
@@ -275,6 +310,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 +339,38 @@ impl JsActorStateTransaction {
 	}
 }
 
+fn wait_for_runtime<T, F>(future: F) -> napi::Result<T>
+where
+	T: Send + 'static,
+	F: Future<Output = anyhow::Result<T>> + Send + 'static,
+{
+	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 query: {error}"),
+			}
+			.build(),
+		)
+	})?;
+	let (sender, receiver) = mpsc::sync_channel(1);
+	runtime.spawn(async move {
+		let _ = sender.send(future.await);
+	});
+	receiver
+		.recv()
+		.map_err(|error| {
+			napi_anyhow_error(
+				crate::NapiInvalidState {
+					state: "runtime".to_owned(),
+					reason: format!("synchronous SQLite query ended without a result: {error}"),
+				}
+				.build(),
+			)
+		})?
+		.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 +467,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,32 @@ 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 transactionCount = await c.db.transaction((tx) => {
+				if (!tx.executeSync) {
+					throw new Error(
+						"synchronous transaction queries are unavailable",
+					);
+				}
+				return tx.executeSync<{ count: number }>(
+					"SELECT COUNT(*) AS count FROM test_data",
+				)[0]?.count;
+			});
+			return {
+				value: selected[0]?.value,
+				transactionCount,
+			};
+		},
 		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;

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

@@ -1,6 +1,9 @@
 import crypto from "node:crypto";
 import type { RequestContext } from "@/actor/config";
-import type { DatabaseProvider, RawAccess } from "@/common/database/config";
+import type {
+	DatabaseProvider,
+	SynchronousRawAccess,
+} from "@/common/database/config";
 import type { AgentOsActorConfig } from "../config";
 import type {
 	AgentOsActionContext,
@@ -46,7 +49,7 @@ type AgentOsRequestContext<TConnParams> = RequestContext<
 	undefined,
 	AgentOsActorVars,
 	undefined,
-	DatabaseProvider<RawAccess>
+	DatabaseProvider<SynchronousRawAccess>
 >;
 
 export function buildOnRequestHandler<TConnParams>(

rivetkit-typescript/packages/rivetkit/src/common/database/config.ts

@@ -84,7 +84,15 @@ export interface SqliteDatabase {
 		sql: string,
 		callback?: (row: unknown[], columns: string[]) => void,
 	): Promise<void>;
+	execSync?(
+		sql: string,
+		callback?: (row: unknown[], columns: string[]) => void,
+	): void;
 	execute(sql: string, params?: SqliteBindings): Promise<SqliteExecuteResult>;
+	executeSync?(
+		sql: string,
+		params?: SqliteBindings,
+	): SqliteExecuteResult;
 	executeBatch(
 		statements: SqliteBatchStatement[],
 	): Promise<SqliteExecuteResult[]>;
@@ -106,7 +114,15 @@ export interface SqliteTransactionDatabase {
 		sql: string,
 		callback?: (row: unknown[], columns: string[]) => void,
 	): Promise<void>;
+	execSync?(
+		sql: string,
+		callback?: (row: unknown[], columns: string[]) => void,
+	): void;
 	execute(sql: string, params?: SqliteBindings): Promise<SqliteExecuteResult>;
+	executeSync?(
+		sql: string,
+		params?: SqliteBindings,
+	): SqliteExecuteResult;
 	commit(): Promise<void>;
 	rollback(): Promise<void>;
 }
@@ -204,11 +220,23 @@ type ExecuteFunction = <
 	...args: unknown[]
 ) => Promise<TRow[]>;
 
+type ExecuteSyncFunction = <
+	TRow extends Record<string, unknown> = Record<string, unknown>,
+>(
+	query: string,
+	...args: unknown[]
+) => TRow[];
+
 export type RawAccess = {
 	/**
 	 * Executes a raw SQL query.
 	 */
 	execute: ExecuteFunction;
+	/**
+	 * Executes a raw SQL query synchronously when supported by the runtime.
+	 * This blocks the Node.js event loop. Prefer `execute` for normal use.
+	 */
+	executeSync?: ExecuteSyncFunction;
 	/** Runs a callback in an isolated SQLite transaction. */
 	transaction: <T>(
 		callback: (tx: RawAccess) => Promise<T> | T,
@@ -226,3 +254,8 @@ export type RawAccess = {
 	 */
 	close: () => Promise<void>;
 };
+
+/** Raw database access with synchronous queries provided by the Node.js runtime. */
+export type SynchronousRawAccess = RawAccess & {
+	executeSync: ExecuteSyncFunction;
+};