diff --git a/docs/content/docs/sqlite.mdx b/docs/content/docs/sqlite.mdx index a0dbcc43f9..5797930d5a 100644 --- a/docs/content/docs/sqlite.mdx +++ b/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(...)`. + + +**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. + + +```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. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e8d9fcae90..9a24d5c7cb 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -3806,6 +3806,9 @@ importers: semver: specifier: ^7.6.0 version: 7.7.4 + yaml: + specifier: ^2.9.0 + version: 2.9.0 devDependencies: '@types/node': specifier: ^24.3.0 diff --git a/rivetkit-typescript/packages/effect/src/Actor.test-d.ts b/rivetkit-typescript/packages/effect/src/Actor.test-d.ts index 5be7817214..b1cedaba7e 100644 --- a/rivetkit-typescript/packages/effect/src/Actor.test-d.ts +++ b/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(); + ).toEqualTypeOf(); expectTypeOf( wakeOptions.rawRivetkitContext.db.transaction( async () => {}, { name: "effect-operation", timeout: 1_000 }, ), ).toEqualTypeOf>(); + expectTypeOf( + wakeOptions.rawRivetkitContext.db.executeSync<{ + count: number; + }>("SELECT COUNT(*) AS count"), + ).toEqualTypeOf<{ count: number }[]>(); + expectTypeOf( + wakeOptions.rawRivetkitContext.db.transactionSync(() => 42), + ).toEqualTypeOf(); return { Ping: () => Effect.succeed(0), diff --git a/rivetkit-typescript/packages/rivetkit-napi/index.d.ts b/rivetkit-typescript/packages/rivetkit-napi/index.d.ts index 6e980d29c6..de062a406a 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/index.d.ts +++ b/rivetkit-typescript/packages/rivetkit-napi/index.d.ts @@ -379,19 +379,27 @@ export declare class JsNativeDatabase { run(sql: string, params?: Array | undefined | null): Promise query(sql: string, params?: Array | undefined | null): Promise execute(sql: string, params?: Array | undefined | null): Promise + executeSync(sql: string, params?: Array | undefined | null): NativeExecuteResult executeBatch(statements: Array): Promise> exec(sql: string): Promise + execSync(sql: string): QueryResult close(): Promise beginTransaction(timeoutMs?: number | undefined | null, name?: string | undefined | null): Promise + beginTransactionSync(timeoutMs?: number | undefined | null, name?: string | undefined | null): JsSqliteTransaction } export declare class JsSqliteTransaction { execute(sql: string, params?: Array | undefined | null): Promise + executeSync(sql: string, params?: Array | undefined | null): NativeExecuteResult exec(sql: string): Promise + execSync(sql: string): QueryResult commit(): Promise + commitSync(): void rollback(): Promise + rollbackSync(): void } export declare class JsActorStateTransaction { execute(sql: string, params?: Array | undefined | null): Promise + executeSync(sql: string, params?: Array | undefined | null): NativeExecuteResult commit(payload: StateDeltaPayload): Promise rollback(): Promise } diff --git a/rivetkit-typescript/packages/rivetkit-napi/src/database.rs b/rivetkit-typescript/packages/rivetkit-napi/src/database.rs index fad6e3c0ef..f4d1dfe61e 100644 --- a/rivetkit-typescript/packages/rivetkit-napi/src/database.rs +++ b/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>, + ) -> napi::Result { + 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 { + 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, + name: Option, + ) -> napi::Result { + 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>, + ) -> napi::Result { + 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 { self.transaction @@ -242,6 +287,12 @@ impl JsSqliteTransaction { .map_err(crate::napi_anyhow_error) } + #[napi] + pub fn exec_sync(&self, sql: String) -> napi::Result { + 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,6 +301,12 @@ 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 @@ -257,6 +314,12 @@ impl JsSqliteTransaction { .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>, + ) -> napi::Result { + 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(future: F) -> napi::Result +where + F: Future>, +{ + 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 { 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); + } +} diff --git a/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts b/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts index 731f95cccc..8fd33d33c5 100644 --- a/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts +++ b/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 }; diff --git a/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/index.ts b/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/index.ts index accac8eec4..0b6ac788ba 100644 --- a/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/index.ts +++ b/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( undefined, AgentOsActorVars, undefined, - DatabaseProvider, + DatabaseProvider, { sessionEvent: typeof sessionEventToken; permissionRequest: typeof permissionRequestToken; @@ -182,7 +185,7 @@ export function agentOs( undefined, AgentOsActorVars, undefined, - DatabaseProvider, + DatabaseProvider, { sessionEvent: typeof sessionEventToken; permissionRequest: typeof permissionRequestToken; diff --git a/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/preview.ts b/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/preview.ts index 3e5f9f7e96..8c5e8b5794 100644 --- a/rivetkit-typescript/packages/rivetkit/src/agent-os/actor/preview.ts +++ b/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 = RequestContext< undefined, AgentOsActorVars, undefined, - DatabaseProvider + DatabaseProvider >; export function buildOnRequestHandler( diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts index 7ff08c39e3..465a954554 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts @@ -84,7 +84,12 @@ export interface SqliteDatabase { sql: string, callback?: (row: unknown[], columns: string[]) => void, ): Promise; + execSync?( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void; execute(sql: string, params?: SqliteBindings): Promise; + executeSync?(sql: string, params?: SqliteBindings): SqliteExecuteResult; executeBatch( statements: SqliteBatchStatement[], ): Promise; @@ -92,6 +97,10 @@ export interface SqliteDatabase { timeoutMs?: number, name?: string, ): Promise; + beginTransactionSync?( + timeoutMs?: number, + name?: string, + ): SynchronousSqliteTransactionDatabase; run(sql: string, params?: SqliteBindings): Promise; query(sql: string, params?: SqliteBindings): Promise; nativeMetrics?(): @@ -106,11 +115,26 @@ export interface SqliteTransactionDatabase { sql: string, callback?: (row: unknown[], columns: string[]) => void, ): Promise; + execSync?( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void; execute(sql: string, params?: SqliteBindings): Promise; + executeSync?(sql: string, params?: SqliteBindings): SqliteExecuteResult; commit(): Promise; rollback(): Promise; } +export type SynchronousSqliteTransactionDatabase = SqliteTransactionDatabase & { + execSync( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void; + executeSync(sql: string, params?: SqliteBindings): SqliteExecuteResult; + commitSync(): void; + rollbackSync(): void; +}; + /** * Provider for opening native databases from the active runtime. */ @@ -204,11 +228,28 @@ type ExecuteFunction = < ...args: unknown[] ) => Promise; +type ExecuteSyncFunction = < + TRow extends Record = Record, +>( + query: string, + ...args: unknown[] +) => TRow[]; + +/** SQL operations available inside a synchronous transaction callback. */ +export type SynchronousTransactionAccess = { + executeSync: ExecuteSyncFunction; +}; + 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: ( callback: (tx: RawAccess) => Promise | T, @@ -226,3 +267,16 @@ export type RawAccess = { */ close: () => Promise; }; + +/** Raw database access with synchronous operations provided by the Node.js runtime. */ +export type SynchronousRawAccess = RawAccess & { + executeSync: ExecuteSyncFunction; + /** + * Runs a synchronous callback in an isolated SQLite transaction. + * The callback must not return a promise. + */ + transactionSync: ( + callback: (tx: SynchronousTransactionAccess) => T, + options?: Omit, + ) => T; +}; diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/mod.test.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/mod.test.ts index 21904e386f..40e25800b2 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/mod.test.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/mod.test.ts @@ -7,6 +7,7 @@ import type { SqliteDatabase, SqliteExecuteResult, SqliteTransactionDatabase, + SynchronousSqliteTransactionDatabase, } from "./config"; import { db, registerNativeStateTransactionOpener } from "./mod"; @@ -19,7 +20,20 @@ class FakeSqliteDatabase implements SqliteDatabase { transactionTimeouts: Array = []; transactionNames: Array = []; - async exec(): Promise {} + async exec( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): Promise { + this.execSync(sql, callback); + } + + execSync( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void { + this.record(sql); + callback?.([1], ["value"]); + } async execute( sql: string, @@ -29,21 +43,40 @@ class FakeSqliteDatabase implements SqliteDatabase { return emptyResult(); } + executeSync(sql: string, params?: SqliteBindings): SqliteExecuteResult { + this.record(sql, params); + return emptyResult(); + } + async beginTransaction( timeoutMs?: number, name?: string, ): Promise { + return this.beginTransactionSync(timeoutMs, name); + } + + beginTransactionSync( + timeoutMs?: number, + name?: string, + ): SynchronousSqliteTransactionDatabase { this.transactionTimeouts.push(timeoutMs); this.transactionNames.push(name); this.record("BEGIN"); return { exec: async () => {}, + execSync: () => {}, execute: async (sql, params) => { this.record(sql, params); return emptyResult(); }, + executeSync: (sql, params) => { + this.record(sql, params); + return emptyResult(); + }, commit: async () => this.record("COMMIT"), + commitSync: () => this.record("COMMIT"), rollback: async () => this.record("ROLLBACK"), + rollbackSync: () => this.record("ROLLBACK"), }; } async beginStateTransaction( @@ -53,10 +86,15 @@ class FakeSqliteDatabase implements SqliteDatabase { this.record("BEGIN_STATE"); return { exec: async () => {}, + execSync: () => {}, execute: async (sql, params) => { this.record(sql, params); return emptyResult(); }, + executeSync: (sql, params) => { + this.record(sql, params); + return emptyResult(); + }, commit: async () => this.record("COMMIT"), rollback: async () => this.record("ROLLBACK"), }; @@ -173,6 +211,129 @@ describe("db", () => { }); }); + test("exposes synchronous raw queries on the built-in client", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + client.executeSync("SELECT ?", 42); + expect( + client.executeSync<{ value: number }>("SELECT 1; SELECT 2"), + ).toEqual([{ value: 1 }]); + + expect(nativeDb.executeCalls).toEqual([ + { + sql: "SELECT ?", + params: [42], + }, + { + sql: "SELECT 1; SELECT 2", + params: undefined, + }, + ]); + }); + + test("commits synchronous transaction work before returning", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + const value = client.transactionSync( + (tx) => { + tx.executeSync("INSERT INTO items(value) VALUES (?)", "inside"); + return 42; + }, + { name: "sync-insert", timeout: 120_000 }, + ); + + expect(value).toBe(42); + expect(nativeDb.transactionTimeouts).toEqual([120_000]); + expect(nativeDb.transactionNames).toEqual(["sync-insert"]); + expect(nativeDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "INSERT INTO items(value) VALUES (?)", + "COMMIT", + ]); + }); + + test("rolls back synchronous transactions on callback and commit errors", async () => { + const callbackDb = new FakeSqliteDatabase(); + const callbackClient = await db().createClient( + testProviderContext(callbackDb), + ); + expect(() => + callbackClient.transactionSync(() => { + throw new Error("callback failed"); + }), + ).toThrow("callback failed"); + expect(callbackDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "ROLLBACK", + ]); + + const commitDb = new FakeSqliteDatabase(); + commitDb.failSql.set("COMMIT", new Error("commit failed")); + const commitClient = await db().createClient( + testProviderContext(commitDb), + ); + expect(() => commitClient.transactionSync(() => 1)).toThrow( + "commit failed", + ); + expect(commitDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "COMMIT", + "ROLLBACK", + ]); + }); + + test("rejects async synchronous-transaction callbacks and outer-client queries", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + expect(() => client.transactionSync(async () => undefined)).toThrow( + "must not return a promise", + ); + expect(nativeDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "ROLLBACK", + ]); + + nativeDb.executeCalls = []; + let outerQuery: Promise[]> | undefined; + client.transactionSync((tx) => { + expect(Object.keys(tx)).toEqual(["executeSync"]); + expect(() => client.executeSync("SELECT 1")).toThrow( + "transaction callback's tx value", + ); + outerQuery = client.execute("SELECT 1"); + expect(() => client.transactionSync(() => undefined)).toThrow( + "Nested synchronous SQLite transactions", + ); + tx.executeSync("SELECT 2"); + }); + await expect(outerQuery).rejects.toThrow( + "transaction callback's tx value", + ); + expect(nativeDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "SELECT 2", + "COMMIT", + ]); + }); + + test("validates synchronous transaction options before beginning", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + for (const timeout of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) { + expect(() => + client.transactionSync(() => undefined, { timeout }), + ).toThrow("positive finite"); + } + expect(() => + client.transactionSync(() => undefined, { name: "" }), + ).toThrow("must not be empty"); + expect(nativeDb.executeCalls).toEqual([]); + }); + test("rolls back migrations when onMigrate fails", async () => { const nativeDb = new FakeSqliteDatabase(); const provider = db({ diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/mod.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/mod.ts index 5f29b65fb2..b1226940ca 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/mod.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/mod.ts @@ -7,17 +7,20 @@ import type { SqliteProfilingOptions, SqliteTransactionDatabase, SqliteTransactionOptions, + SynchronousRawAccess, + SynchronousTransactionAccess, } from "./config"; import { isManualTransactionControl, isSqliteBindingObject, MIGRATION_TRANSACTION_TIMEOUT_MS, + runSqliteTransactionSync, toSqliteBindings, validateTransactionName, validateTransactionTimeout, } from "./shared"; -export type { RawAccess } from "./config"; +export type { RawAccess, SynchronousRawAccess } from "./config"; export interface DatabaseFactoryConfig { onMigrate?: (db: RawAccess) => Promise | void; @@ -84,8 +87,8 @@ export function db({ onMigrate, warnOnManualTransactions = true, profiling, -}: DatabaseFactoryConfig = {}): DatabaseProvider { - const provider: DatabaseProvider = { +}: DatabaseFactoryConfig = {}): DatabaseProvider { + const provider: DatabaseProvider = { sqliteProfiling: profiling, createClient: async (ctx) => { const nativeDatabaseProvider = ctx.nativeDatabaseProvider; @@ -98,6 +101,7 @@ export function db({ const db = await nativeDatabaseProvider.open(ctx.actorId); let closed = false; let manualTransactionWarned = false; + let synchronousTransactionActive = false; const ensureOpen = () => { if (closed) { throw new Error( @@ -105,13 +109,22 @@ export function db({ ); } }; + const ensureSynchronousTransactionClient = ( + transactionScoped: boolean, + ) => { + if (!transactionScoped && synchronousTransactionActive) { + throw new Error( + "Use the transaction callback's tx value for queries inside db.transactionSync().", + ); + } + }; const createClient = ( target: SqliteDatabase | SqliteTransactionDatabase, transactionScoped = false, stateTransactionContext?: NativeStateTransactionContext, - ): RawAccess => { - const client: RawAccess = { + ): SynchronousRawAccess => { + const client: SynchronousRawAccess = { execute: async < TRow extends Record = Record< string, @@ -122,6 +135,7 @@ export function db({ ...args: unknown[] ): Promise => { ensureOpen(); + ensureSynchronousTransactionClient(transactionScoped); if ( !transactionScoped && warnOnManualTransactions && @@ -132,7 +146,7 @@ export function db({ manualTransactionWarned = true; getLogger("database").warn( { actorId: ctx.actorId }, - "Manual cross-call SQLite transactions can interleave with other actor work. Use db.transaction() for coordinated transactions. Set warnOnManualTransactions: false in your db(...) configuration to disable this warning.", + "Manual cross-call SQLite transactions can interleave with other actor work. Use db.transaction() or db.transactionSync() for coordinated transactions. Set warnOnManualTransactions: false in your db(...) configuration to disable this warning.", ); } @@ -188,10 +202,91 @@ export function db({ } } }, + executeSync: < + TRow extends Record = Record< + string, + unknown + >, + >( + query: string, + ...args: unknown[] + ): TRow[] => { + ensureOpen(); + ensureSynchronousTransactionClient(transactionScoped); + if (!target.executeSync) { + throw new Error( + "Synchronous SQLite queries are only available in the Node.js native runtime.", + ); + } + if ( + !transactionScoped && + warnOnManualTransactions && + !manualTransactionWarned && + !hasMultipleStatements(query) && + isManualTransactionControl(query) + ) { + manualTransactionWarned = true; + getLogger("database").warn( + { actorId: ctx.actorId }, + "Manual cross-call SQLite transactions can interleave with other actor work. Use db.transaction() or db.transactionSync() for coordinated transactions. Set warnOnManualTransactions: false in your db(...) configuration to disable this warning.", + ); + } + + const kvReadsBefore = ctx.metrics?.totalKvReads ?? 0; + const kvWritesBefore = ctx.metrics?.totalKvWrites ?? 0; + const start = performance.now(); + + try { + if (args.length > 0) { + const bindings = + args.length === 1 && + isSqliteBindingObject(args[0]) + ? toSqliteBindings(args[0]) + : toSqliteBindings(args); + const { rows, columns } = target.executeSync( + query, + bindings, + ); + return rows.map((row) => + rowToObject(row, columns), + ); + } + + if (!hasMultipleStatements(query)) { + const { rows, columns } = target.executeSync( + query, + undefined, + ); + return rows.map((row) => + rowToObject(row, columns), + ); + } + + return execMultiStatementSync(target, query); + } finally { + const durationMs = performance.now() - start; + ctx.metrics?.trackSql(query, durationMs); + if (ctx.metrics) { + const kvReads = + ctx.metrics.totalKvReads - kvReadsBefore; + const kvWrites = + ctx.metrics.totalKvWrites - kvWritesBefore; + ctx.log?.debug({ + msg: "sql query", + query: query.slice(0, 120), + durationMs, + kvReads, + kvWrites, + }); + } + } + }, transaction: async ( callback: (tx: RawAccess) => Promise | T, options?: SqliteTransactionOptions, ): Promise => { + ensureOpen(); + ensureSynchronousTransactionClient(transactionScoped); validateTransactionTimeout(options?.timeout); validateTransactionName(options?.name); if ( @@ -249,6 +344,39 @@ export function db({ } } }, + transactionSync: ( + callback: (tx: SynchronousTransactionAccess) => T, + options?: Omit< + SqliteTransactionOptions, + "experimental" + >, + ): T => { + ensureOpen(); + if (transactionScoped || synchronousTransactionActive) { + throw new Error( + "Nested synchronous SQLite transactions are not supported.", + ); + } + return runSqliteTransactionSync( + db, + (transaction) => { + const transactionClient = createClient( + transaction, + true, + ); + const tx: SynchronousTransactionAccess = { + executeSync: transactionClient.executeSync, + }; + synchronousTransactionActive = true; + try { + return callback(tx); + } finally { + synchronousTransactionActive = false; + } + }, + options, + ); + }, close: async () => { if (!closed) { closed = true; @@ -304,6 +432,26 @@ async function execMultiStatement>( return results as TRow[]; } +function execMultiStatementSync>( + db: Pick, + query: string, +): TRow[] { + if (!db.execSync) { + throw new Error( + "Synchronous SQLite queries are only available in the Node.js native runtime.", + ); + } + const results: Record[] = []; + let columnNames: string[] | null = null; + db.execSync(query, (row: unknown[], columns: string[]) => { + if (!columnNames) { + columnNames = columns; + } + results.push(rowToObject(row, columnNames)); + }); + return results as TRow[]; +} + async function withMigrationSavepoint( client: RawAccess, callback: (leased: RawAccess) => Promise | T, diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.test.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.test.ts index 0e5da38a9a..491a0bb5f3 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.test.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.test.ts @@ -20,28 +20,66 @@ function deferred() { class FakeNativeDatabase implements JsNativeDatabaseLike { async beginTransaction() { + this.transactionEvents.push("BEGIN"); + return this.#transaction(); + } + + beginTransactionSync() { + this.transactionEvents.push("BEGIN_SYNC"); + return this.#transaction(); + } + + #transaction() { return { exec: async (_sql: string) => this.exec(), + execSync: (_sql: string) => this.execSync(), execute: async (sql: string, params?: NativeParams) => this.execute(sql, params), - commit: async () => {}, - rollback: async () => {}, + executeSync: (sql: string, params?: NativeParams) => + this.executeSync(sql, params), + commit: async () => { + this.transactionEvents.push("COMMIT"); + }, + commitSync: () => { + this.transactionEvents.push("COMMIT_SYNC"); + }, + rollback: async () => { + this.transactionEvents.push("ROLLBACK"); + }, + rollbackSync: () => { + this.transactionEvents.push("ROLLBACK_SYNC"); + }, }; } active = 0; maxActive = 0; closed = false; executeCalls: { sql: string; params?: NativeParams; write: boolean }[] = []; + transactionEvents: string[] = []; #pending: ReturnType>[] = []; async exec() { return { columns: [], rows: [] }; } + execSync() { + return { columns: ["value"], rows: [[1], [2]] }; + } + async execute(sql: string, params?: NativeParams) { return await this.#startExecute(sql, params, false); } + executeSync(sql: string, params?: NativeParams): NativeExecuteResult { + this.executeCalls.push({ sql, params, write: false }); + return { + columns: ["value"], + rows: [[1]], + changes: 0, + lastInsertRowId: null, + }; + } + async query(sql: string, params?: NativeParams) { const { columns, rows } = await this.execute(sql, params); return { columns, rows }; @@ -155,6 +193,47 @@ describe("wrapJsNativeDatabase", () => { }); }); + test("executes synchronously with normalized bindings", () => { + const native = new FakeNativeDatabase(); + const db = wrapJsNativeDatabase(native); + + const result = db.executeSync?.("SELECT ?, ?", [true, "text"]); + const execRows: unknown[][] = []; + db.execSync?.("SELECT 1; SELECT 2", (row) => execRows.push(row)); + + expect(native.executeCalls[0]?.params).toEqual([ + { kind: "int", intValue: 1 }, + { kind: "text", textValue: "text" }, + ]); + expect(result).toMatchObject({ + columns: ["value"], + rows: [[1]], + }); + expect(execRows).toEqual([[1], [2]]); + }); + + test("wraps synchronous transaction lifecycle methods", () => { + const native = new FakeNativeDatabase(); + const db = wrapJsNativeDatabase(native); + + const committed = db.beginTransactionSync?.(1_000, "commit"); + if (!committed) + throw new Error("missing synchronous transaction support"); + committed.executeSync("SELECT 1"); + committed.commitSync(); + const rolledBack = db.beginTransactionSync?.(1_000, "rollback"); + if (!rolledBack) + throw new Error("missing synchronous transaction support"); + rolledBack.rollbackSync(); + + expect(native.transactionEvents).toEqual([ + "BEGIN_SYNC", + "COMMIT_SYNC", + "BEGIN_SYNC", + "ROLLBACK_SYNC", + ]); + }); + test("returns native execute metadata", async () => { const native = new FakeNativeDatabase(); const db = wrapJsNativeDatabase(native); diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.ts index 62e1cfd152..81cc981ed5 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/native-database.ts @@ -6,6 +6,7 @@ import type { SqliteExecuteResult, SqliteNativeMetrics, SqliteTransactionDatabase, + SynchronousSqliteTransactionDatabase, } from "./config"; type NativeBindNoValues = { @@ -74,10 +75,15 @@ interface NativeBatchStatement { export interface JsNativeDatabaseLike { exec(sql: string): Promise; + execSync(sql: string): NativeExecResult; execute( sql: string, params?: NativeBindParam[] | null, ): Promise; + executeSync( + sql: string, + params?: NativeBindParam[] | null, + ): NativeExecuteResult; executeBatch?( statements: NativeBatchStatement[], ): Promise; @@ -88,7 +94,11 @@ export interface JsNativeDatabaseLike { beginTransaction( timeoutMs?: number, name?: string, - ): Promise; + ): Promise; + beginTransactionSync( + timeoutMs?: number, + name?: string, + ): JsNativeSynchronousTransactionLike; query( sql: string, params?: NativeBindParam[] | null, @@ -110,14 +120,36 @@ export type StateAwareSqliteDatabase = SqliteDatabase & { export interface JsNativeTransactionLike { exec(sql: string): Promise; + execSync(sql: string): NativeExecResult; execute( sql: string, params?: NativeBindParam[] | null, ): Promise; + executeSync( + sql: string, + params?: NativeBindParam[] | null, + ): NativeExecuteResult; commit(): Promise; rollback(): Promise; } +export interface JsNativeSynchronousTransactionLike + extends JsNativeTransactionLike { + commitSync(): void; + rollbackSync(): void; +} + +function isSynchronousTransaction( + transaction: JsNativeTransactionLike, +): transaction is JsNativeSynchronousTransactionLike { + const candidate = + transaction as Partial; + return ( + typeof candidate.commitSync === "function" && + typeof candidate.rollbackSync === "function" + ); +} + function shouldAttachNativeKvError(message: string): boolean { return /i\/o error|unable to open database file/i.test(message); } @@ -358,6 +390,34 @@ export function wrapJsNativeDatabase( release(); } }; + const executeNativeSync = ( + sql: string, + params?: SqliteBindings, + ): SqliteExecuteResult => { + const lastInsertRowIdColumn = lastInsertRowIdColumnName(sql); + if (lastInsertRowIdColumn) { + return { + columns: [lastInsertRowIdColumn], + rows: [[lastInsertRowId ?? 0]], + changes: 0, + lastInsertRowId, + }; + } + + const release = gate.enter(); + try { + const nativeParams = toNativeBindings(sql, params); + const result = database.executeSync(sql, nativeParams); + if (result.lastInsertRowId !== undefined) { + lastInsertRowId = result.lastInsertRowId; + } + return result; + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + }; return { async exec( @@ -380,12 +440,34 @@ export function wrapJsNativeDatabase( callback(row, result.columns); } }, + execSync( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void { + const release = gate.enter(); + let result: NativeExecResult; + try { + result = database.execSync(sql); + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + if (callback) { + for (const row of result.rows) { + callback(row, result.columns); + } + } + }, async execute( sql: string, params?: SqliteBindings, ): Promise { return await executeNative(sql, params); }, + executeSync(sql: string, params?: SqliteBindings): SqliteExecuteResult { + return executeNativeSync(sql, params); + }, async executeBatch( statements: SqliteBatchStatement[], ): Promise { @@ -435,7 +517,7 @@ export function wrapJsNativeDatabase( name?: string, ): Promise { const release = gate.enter(); - let transaction: JsNativeTransactionLike; + let transaction: JsNativeSynchronousTransactionLike; try { transaction = await database.beginTransaction(timeoutMs, name); } catch (error) { @@ -449,6 +531,25 @@ export function wrapJsNativeDatabase( } }); }, + beginTransactionSync( + timeoutMs?: number, + name?: string, + ): SynchronousSqliteTransactionDatabase { + const release = gate.enter(); + let transaction: JsNativeSynchronousTransactionLike; + try { + transaction = database.beginTransactionSync(timeoutMs, name); + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + return wrapTransaction(database, transaction, gate, (result) => { + if (result.lastInsertRowId !== undefined) { + lastInsertRowId = result.lastInsertRowId; + } + }); + }, async beginStateTransaction( timeoutMs?: number, context?: unknown, @@ -491,13 +592,25 @@ export function wrapJsNativeDatabase( }; } +function wrapTransaction( + database: JsNativeDatabaseLike, + transaction: JsNativeSynchronousTransactionLike, + gate: NativeCloseGate, + onExecute: (result: NativeExecuteResult) => void, +): SynchronousSqliteTransactionDatabase; +function wrapTransaction( + database: JsNativeDatabaseLike, + transaction: JsNativeTransactionLike, + gate: NativeCloseGate, + onExecute: (result: NativeExecuteResult) => void, +): SqliteTransactionDatabase; function wrapTransaction( database: JsNativeDatabaseLike, transaction: JsNativeTransactionLike, gate: NativeCloseGate, onExecute: (result: NativeExecuteResult) => void, ): SqliteTransactionDatabase { - return { + const wrapped: SqliteTransactionDatabase = { async exec(sql, callback) { const release = gate.enter(); let result: NativeExecResult; @@ -512,6 +625,20 @@ function wrapTransaction( for (const row of result.rows) callback(row, result.columns); } }, + execSync(sql, callback) { + const release = gate.enter(); + let result: NativeExecResult; + try { + result = transaction.execSync(sql); + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + if (callback) { + for (const row of result.rows) callback(row, result.columns); + } + }, async execute(sql, params) { const release = gate.enter(); try { @@ -527,6 +654,21 @@ function wrapTransaction( release(); } }, + executeSync(sql, params) { + const release = gate.enter(); + try { + const result = transaction.executeSync( + sql, + toNativeBindings(sql, params), + ); + onExecute(result); + return result; + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + }, async commit() { const release = gate.enter(); try { @@ -548,6 +690,33 @@ function wrapTransaction( } }, }; + + if (isSynchronousTransaction(transaction)) { + return Object.assign(wrapped, { + commitSync() { + const release = gate.enter(); + try { + transaction.commitSync(); + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + }, + rollbackSync() { + const release = gate.enter(); + try { + transaction.rollbackSync(); + } catch (error) { + enrichNativeDatabaseError(database, error); + } finally { + release(); + } + }, + }) as SynchronousSqliteTransactionDatabase; + } + + return wrapped; } function lastInsertRowIdColumnName(sql: string): string | undefined { diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/shared.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/shared.ts index 21ea39cf71..fcd47d774b 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/shared.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/shared.ts @@ -1,4 +1,9 @@ -import type { SqliteBindings } from "./config"; +import type { + SqliteBindings, + SqliteDatabase, + SqliteTransactionOptions, + SynchronousSqliteTransactionDatabase, +} from "./config"; /** Migrations may legitimately do substantially more work than request transactions. */ export const MIGRATION_TRANSACTION_TIMEOUT_MS = 5 * 60_000; @@ -24,6 +29,49 @@ export function validateTransactionName(name: string | undefined): void { } } +export function isPromiseLike(value: unknown): value is PromiseLike { + return (typeof value === "object" && value !== null) || + typeof value === "function" + ? typeof (value as { then?: unknown }).then === "function" + : false; +} + +export function runSqliteTransactionSync( + database: SqliteDatabase, + callback: (transaction: SynchronousSqliteTransactionDatabase) => T, + options?: Omit, +): T { + validateTransactionTimeout(options?.timeout); + validateTransactionName(options?.name); + if (!database.beginTransactionSync) { + throw new Error( + "Synchronous SQLite transactions are only available in the Node.js native runtime.", + ); + } + + const transaction = database.beginTransactionSync( + options?.timeout, + options?.name, + ); + try { + const result = callback(transaction); + if (isPromiseLike(result)) { + throw new Error( + "db.transactionSync() callback must complete synchronously and must not return a promise.", + ); + } + transaction.commitSync(); + return result; + } catch (error) { + try { + transaction.rollbackSync(); + } catch { + // Preserve the callback or commit error after cleanup failure. + } + throw error; + } +} + type SqliteBindingObject = Record; function isSqliteBindingValue(value: unknown): boolean { diff --git a/rivetkit-typescript/packages/rivetkit/src/db/drizzle.test.ts b/rivetkit-typescript/packages/rivetkit/src/db/drizzle.test.ts index f806e679c7..615c4e469f 100644 --- a/rivetkit-typescript/packages/rivetkit/src/db/drizzle.test.ts +++ b/rivetkit-typescript/packages/rivetkit/src/db/drizzle.test.ts @@ -6,6 +6,7 @@ import type { SqliteDatabase, SqliteExecuteResult, SqliteTransactionDatabase, + SynchronousSqliteTransactionDatabase, } from "@/common/database/config"; import { db } from "./drizzle"; @@ -14,7 +15,20 @@ class FakeSqliteDatabase implements SqliteDatabase { transactionTimeouts: Array = []; transactionNames: Array = []; - async exec(): Promise {} + async exec( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): Promise { + this.execSync(sql, callback); + } + + execSync( + sql: string, + callback?: (row: unknown[], columns: string[]) => void, + ): void { + this.executeCalls.push({ sql }); + callback?.([1], ["value"]); + } async execute( sql: string, @@ -24,6 +38,11 @@ class FakeSqliteDatabase implements SqliteDatabase { return emptyResult(); } + executeSync(sql: string, params?: SqliteBindings): SqliteExecuteResult { + this.executeCalls.push({ sql, params }); + return emptyResult(); + } + async executeBatch( statements: SqliteBatchStatement[], ): Promise { @@ -47,21 +66,39 @@ class FakeSqliteDatabase implements SqliteDatabase { timeoutMs?: number, name?: string, ): Promise { + return this.beginTransactionSync(timeoutMs, name); + } + + beginTransactionSync( + timeoutMs?: number, + name?: string, + ): SynchronousSqliteTransactionDatabase { this.transactionTimeouts.push(timeoutMs); this.transactionNames.push(name); this.executeCalls.push({ sql: "BEGIN" }); return { exec: async () => {}, + execSync: () => {}, execute: async (sql, params) => { this.executeCalls.push({ sql, params }); return emptyResult(); }, + executeSync: (sql, params) => { + this.executeCalls.push({ sql, params }); + return emptyResult(); + }, commit: async () => { this.executeCalls.push({ sql: "COMMIT" }); }, + commitSync: () => { + this.executeCalls.push({ sql: "COMMIT" }); + }, rollback: async () => { this.executeCalls.push({ sql: "ROLLBACK" }); }, + rollbackSync: () => { + this.executeCalls.push({ sql: "ROLLBACK" }); + }, }; } @@ -151,6 +188,50 @@ describe("Drizzle database transactions", () => { ]); }); + test("exposes synchronous raw queries", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + client.executeSync("SELECT ?", 42); + expect( + client.executeSync<{ value: number }>("SELECT 1; SELECT 2"), + ).toEqual([{ value: 1 }]); + + expect(nativeDb.executeCalls).toEqual([ + { sql: "SELECT ?", params: [42] }, + { sql: "SELECT 1; SELECT 2" }, + ]); + }); + + test("commits and rolls back synchronous raw transactions", async () => { + const nativeDb = new FakeSqliteDatabase(); + const client = await db().createClient(testProviderContext(nativeDb)); + + const value = client.transactionSync( + (tx) => { + tx.executeSync("INSERT INTO items(value) VALUES (?)", "inside"); + return 42; + }, + { name: "sync-drizzle", timeout: 120_000 }, + ); + expect(value).toBe(42); + expect(() => + client.transactionSync(() => { + throw new Error("callback failed"); + }), + ).toThrow("callback failed"); + + expect(nativeDb.transactionTimeouts).toEqual([120_000, undefined]); + expect(nativeDb.transactionNames).toEqual(["sync-drizzle", undefined]); + expect(nativeDb.executeCalls.map(({ sql }) => sql)).toEqual([ + "BEGIN", + "INSERT INTO items(value) VALUES (?)", + "COMMIT", + "BEGIN", + "ROLLBACK", + ]); + }); + test("validates zero, negative, and non-finite transaction timeouts", async () => { const client = await db().createClient( testProviderContext(new FakeSqliteDatabase()), diff --git a/rivetkit-typescript/packages/rivetkit/src/db/drizzle.ts b/rivetkit-typescript/packages/rivetkit/src/db/drizzle.ts index 94aea12ef4..f9e39f7055 100644 --- a/rivetkit-typescript/packages/rivetkit/src/db/drizzle.ts +++ b/rivetkit-typescript/packages/rivetkit/src/db/drizzle.ts @@ -6,20 +6,22 @@ import { import type { DatabaseProvider, DatabaseProviderContext, - RawAccess, - SqliteProfilingOptions, - SqliteTransactionOptions, SqliteDatabase, + SqliteProfilingOptions, SqliteTransactionDatabase, + SqliteTransactionOptions, + SynchronousRawAccess, + SynchronousTransactionAccess, } from "@/common/database/config"; -import { getLogger } from "@/common/log"; import { isManualTransactionControl, MIGRATION_TRANSACTION_TIMEOUT_MS, + runSqliteTransactionSync, toSqliteBindings, validateTransactionName, validateTransactionTimeout, } from "@/common/database/shared"; +import { getLogger } from "@/common/log"; import { sha256Hex } from "@/utils/crypto"; export type { SQLiteTable } from "drizzle-orm/sqlite-core"; @@ -42,7 +44,7 @@ type DrizzleDatabase = Omit< SqliteRemoteDatabase, "transaction" > & - Omit & { + Omit & { transaction: ( callback: (tx: DrizzleDatabase) => Promise | T, options?: SqliteTransactionOptions, @@ -114,6 +116,7 @@ export function db>({ const nativeDb = await nativeDatabaseProvider.open(ctx.actorId); let closed = false; let manualTransactionWarned = false; + let synchronousTransactionActive = false; const ensureOpen = () => { if (closed) { throw new Error( @@ -121,6 +124,15 @@ export function db>({ ); } }; + const ensureSynchronousTransactionClient = ( + transactionScoped: boolean, + ) => { + if (!transactionScoped && synchronousTransactionActive) { + throw new Error( + "Use the transaction callback's tx value for queries inside db.transactionSync().", + ); + } + }; const createDrizzleClient = ( target: SqliteDatabase | SqliteTransactionDatabase, @@ -132,6 +144,7 @@ export function db>({ method: "run" | "all" | "values" | "get", ) => { ensureOpen(); + ensureSynchronousTransactionClient(transactionScoped); warnForManualTransaction(query, transactionScoped); const start = performance.now(); @@ -186,6 +199,7 @@ export function db>({ query: string, ...args: unknown[] ): Promise => { + ensureSynchronousTransactionClient(transactionScoped); return await executeRaw( target, ctx, @@ -196,12 +210,34 @@ export function db>({ warnForManualTransaction(query, transactionScoped), ); }; + drizzleDb.executeSync = < + TRow extends Record = Record< + string, + unknown + >, + >( + query: string, + ...args: unknown[] + ): TRow[] => { + ensureSynchronousTransactionClient(transactionScoped); + return executeRawSync( + target, + ctx, + ensureOpen, + query, + args, + () => + warnForManualTransaction(query, transactionScoped), + ); + }; drizzleDb.transaction = async ( transactionCallback: ( tx: DrizzleDatabase, ) => Promise | T, options?: SqliteTransactionOptions, ): Promise => { + ensureOpen(); + ensureSynchronousTransactionClient(transactionScoped); validateTransactionTimeout(options?.timeout); validateTransactionName(options?.name); const transaction = await nativeDb.beginTransaction( @@ -222,6 +258,38 @@ export function db>({ throw error; } }; + drizzleDb.transactionSync = ( + transactionCallback: ( + tx: SynchronousTransactionAccess, + ) => T, + options?: Omit, + ): T => { + ensureOpen(); + if (transactionScoped || synchronousTransactionActive) { + throw new Error( + "Nested synchronous SQLite transactions are not supported.", + ); + } + return runSqliteTransactionSync( + nativeDb, + (transaction) => { + const transactionClient = createDrizzleClient( + transaction, + true, + ); + const tx: SynchronousTransactionAccess = { + executeSync: transactionClient.executeSync, + }; + synchronousTransactionActive = true; + try { + return transactionCallback(tx); + } finally { + synchronousTransactionActive = false; + } + }, + options, + ); + }; drizzleDb.close = async () => { if (!closed) { closed = true; @@ -248,7 +316,7 @@ export function db>({ manualTransactionWarned = true; getLogger("database").warn( { actorId: ctx.actorId }, - "Manual cross-call SQLite transactions can interleave with other actor work. Use db.transaction() for coordinated transactions. Set warnOnManualTransactions: false in your db(...) configuration to disable this warning.", + "Manual cross-call SQLite transactions can interleave with other actor work. Use db.transaction() or db.transactionSync() for coordinated transactions. Set warnOnManualTransactions: false in your db(...) configuration to disable this warning.", ); }; @@ -431,3 +499,65 @@ async function executeRaw>( } } } + +function executeRawSync>( + db: SqliteDatabase | SqliteTransactionDatabase, + ctx: DatabaseProviderContext, + ensureOpen: () => void, + query: string, + args: unknown[], + warnForManualTransaction: () => void, +): TRow[] { + ensureOpen(); + warnForManualTransaction(); + if (!db.executeSync) { + throw new Error( + "Synchronous SQLite queries are only available in the Node.js native runtime.", + ); + } + + const start = performance.now(); + const kvReadsBefore = ctx.metrics?.totalKvReads ?? 0; + const kvWritesBefore = ctx.metrics?.totalKvWrites ?? 0; + try { + if (args.length > 0) { + const { rows, columns } = db.executeSync( + query, + toSqliteBindings(args), + ); + return rows.map((row) => rowToObject(row, columns)); + } + + if (!hasMultipleStatements(query)) { + const { rows, columns } = db.executeSync(query, undefined); + return rows.map((row) => rowToObject(row, columns)); + } + + if (!db.execSync) { + throw new Error( + "Synchronous SQLite queries are only available in the Node.js native runtime.", + ); + } + const results: Record[] = []; + let columnNames: string[] | null = null; + db.execSync(query, (row, columns) => { + if (!columnNames) { + columnNames = columns; + } + results.push(rowToObject(row, columnNames)); + }); + return results as TRow[]; + } finally { + const durationMs = performance.now() - start; + ctx.metrics?.trackSql(query, durationMs); + if (ctx.metrics) { + ctx.log?.debug({ + msg: "sql query", + query: query.slice(0, 120), + durationMs, + kvReads: ctx.metrics.totalKvReads - kvReadsBefore, + kvWrites: ctx.metrics.totalKvWrites - kvWritesBefore, + }); + } + } +} diff --git a/rivetkit-typescript/packages/rivetkit/src/db/mod.ts b/rivetkit-typescript/packages/rivetkit/src/db/mod.ts index cea213bd6c..a5e8977a4e 100644 --- a/rivetkit-typescript/packages/rivetkit/src/db/mod.ts +++ b/rivetkit-typescript/packages/rivetkit/src/db/mod.ts @@ -12,5 +12,7 @@ export type { SqliteProfilingOptions, SqliteQueryResult, SqliteTransactionOptions, + SynchronousRawAccess, + SynchronousTransactionAccess, } from "@/common/database/config"; export { db } from "@/common/database/mod"; diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/napi-runtime.ts b/rivetkit-typescript/packages/rivetkit/src/registry/napi-runtime.ts index 3d589d5242..6c419c8e7a 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/napi-runtime.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/napi-runtime.ts @@ -743,6 +743,13 @@ export class NapiCoreRuntime implements CoreRuntime { return await this.#actorSql(ctx).exec(sql); } + actorSqlExecSync( + ctx: ActorContextHandle, + sql: string, + ): RuntimeSqlExecResult { + return this.#actorSql(ctx).execSync(sql); + } + async actorSqlExecute( ctx: ActorContextHandle, sql: string, @@ -755,6 +762,18 @@ export class NapiCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorSqlExecuteSync( + ctx: ActorContextHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + const result = this.#actorSql(ctx).executeSync( + sql, + toNapiSqlBindParams(params), + ); + return normalizeRuntimeSqlExecuteResult(result); + } + async actorSqlExecuteBatch( ctx: ActorContextHandle, statements: RuntimeSqlBatchStatement[], @@ -776,6 +795,17 @@ export class NapiCoreRuntime implements CoreRuntime { )) as unknown as SqliteTransactionHandle; } + actorSqlBeginTransactionSync( + ctx: ActorContextHandle, + timeoutMs?: number, + name?: string, + ): SqliteTransactionHandle { + return this.#actorSql(ctx).beginTransactionSync( + timeoutMs, + name, + ) as unknown as SqliteTransactionHandle; + } + async actorSqlTransactionExec( transaction: SqliteTransactionHandle, sql: string, @@ -783,6 +813,13 @@ export class NapiCoreRuntime implements CoreRuntime { return await asNativeSqlTransaction(transaction).exec(sql); } + actorSqlTransactionExecSync( + transaction: SqliteTransactionHandle, + sql: string, + ): RuntimeSqlExecResult { + return asNativeSqlTransaction(transaction).execSync(sql); + } + async actorSqlTransactionExecute( transaction: SqliteTransactionHandle, sql: string, @@ -795,17 +832,39 @@ export class NapiCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorSqlTransactionExecuteSync( + transaction: SqliteTransactionHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + const result = asNativeSqlTransaction(transaction).executeSync( + sql, + toNapiSqlBindParams(params), + ); + return normalizeRuntimeSqlExecuteResult(result); + } + async actorSqlTransactionCommit( transaction: SqliteTransactionHandle, ): Promise { await asNativeSqlTransaction(transaction).commit(); } + actorSqlTransactionCommitSync(transaction: SqliteTransactionHandle): void { + asNativeSqlTransaction(transaction).commitSync(); + } + async actorSqlTransactionRollback( transaction: SqliteTransactionHandle, ): Promise { await asNativeSqlTransaction(transaction).rollback(); } + + actorSqlTransactionRollbackSync( + transaction: SqliteTransactionHandle, + ): void { + asNativeSqlTransaction(transaction).rollbackSync(); + } async actorBeginStateTransaction( ctx: ActorContextHandle, timeoutMs?: number, @@ -827,6 +886,18 @@ export class NapiCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorStateTransactionExecuteSync( + transaction: ActorStateTransactionHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + const result = asNativeActorStateTransaction(transaction).executeSync( + sql, + toNapiSqlBindParams(params), + ); + return normalizeRuntimeSqlExecuteResult(result); + } + async actorStateTransactionCommit( transaction: ActorStateTransactionHandle, payload: RuntimeStateDeltaPayload, diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts index cee301a9e2..7514492b77 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts @@ -55,7 +55,10 @@ import { db as defaultDb, registerNativeStateTransactionOpener, } from "@/common/database/mod"; -import { wrapJsNativeDatabase } from "@/common/database/native-database"; +import { + type JsNativeSynchronousTransactionLike, + wrapJsNativeDatabase, +} from "@/common/database/native-database"; import { assertJsonCompatValue, type JsonCompatValue } from "@/common/encoding"; import { isResponseLike } from "@/common/fetch-like"; import { @@ -118,6 +121,7 @@ import type { RuntimeScheduledFireInfo, RuntimeServeConfig, RuntimeStateDeltaPayload, + SqliteTransactionHandle, WebSocketHandle, } from "./runtime"; import { loadWasmRuntime } from "./wasm-runtime"; @@ -563,6 +567,26 @@ async function closeNativeDatabaseClient( } } +function createNativeSqlTransaction( + runtime: CoreRuntime, + transaction: SqliteTransactionHandle, +): JsNativeSynchronousTransactionLike { + return { + exec: (sql) => runtime.actorSqlTransactionExec(transaction, sql), + execSync: (sql) => + runtime.actorSqlTransactionExecSync(transaction, sql), + execute: (sql, params) => + runtime.actorSqlTransactionExecute(transaction, sql, params), + executeSync: (sql, params) => + runtime.actorSqlTransactionExecuteSync(transaction, sql, params), + commit: () => runtime.actorSqlTransactionCommit(transaction), + commitSync: () => runtime.actorSqlTransactionCommitSync(transaction), + rollback: () => runtime.actorSqlTransactionRollback(transaction), + rollbackSync: () => + runtime.actorSqlTransactionRollbackSync(transaction), + }; +} + function getOrCreateNativeSqlDatabase( runtime: CoreRuntime, ctx: ActorContextHandle, @@ -575,7 +599,10 @@ function getOrCreateNativeSqlDatabase( const database = wrapJsNativeDatabase({ exec: (sql) => runtime.actorSqlExec(ctx, sql), + execSync: (sql) => runtime.actorSqlExecSync(ctx, sql), execute: (sql, params) => runtime.actorSqlExecute(ctx, sql, params), + executeSync: (sql, params) => + runtime.actorSqlExecuteSync(ctx, sql, params), executeBatch: (statements) => runtime.actorSqlExecuteBatch(ctx, statements), beginTransaction: async (timeoutMs, name) => { @@ -584,19 +611,15 @@ function getOrCreateNativeSqlDatabase( timeoutMs, name, ); - return { - exec: (sql) => - runtime.actorSqlTransactionExec(transaction, sql), - execute: (sql, params) => - runtime.actorSqlTransactionExecute( - transaction, - sql, - params, - ), - commit: () => runtime.actorSqlTransactionCommit(transaction), - rollback: () => - runtime.actorSqlTransactionRollback(transaction), - }; + return createNativeSqlTransaction(runtime, transaction); + }, + beginTransactionSync: (timeoutMs, name) => { + const transaction = runtime.actorSqlBeginTransactionSync( + ctx, + timeoutMs, + name, + ); + return createNativeSqlTransaction(runtime, transaction); }, beginStateTransaction: async (timeoutMs, context) => { const scope = context as NativeStateTransactionScope | undefined; @@ -615,12 +638,23 @@ function getOrCreateNativeSqlDatabase( "actor state transactions only support single-statement execute calls", ); }, + execSync: () => { + throw new Error( + "actor state transactions only support single-statement execute calls", + ); + }, execute: (sql, params) => runtime.actorStateTransactionExecute( transaction, sql, params, ), + executeSync: (sql, params) => + runtime.actorStateTransactionExecuteSync( + transaction, + sql, + params, + ), commit: async () => { await runtime.actorStateTransactionCommit( transaction, diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts b/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts index 383cec1d1b..6915f9531c 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/runtime.ts @@ -250,10 +250,15 @@ export interface RuntimeSqlRunResult { export interface RuntimeSqlDatabase { exec(sql: string): Promise; + execSync(sql: string): RuntimeSqlExecResult; execute( sql: string, params?: RuntimeSqlBindParams, ): Promise; + executeSync( + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult; executeBatch( statements: RuntimeSqlBatchStatement[], ): Promise; @@ -269,6 +274,10 @@ export interface RuntimeSqlDatabase { timeoutMs?: number, name?: string, ): Promise; + beginTransactionSync( + timeoutMs?: number, + name?: string, + ): RuntimeSqlTransactionDatabase; metrics?(): SqliteNativeMetrics | null; takeLastKvError?(): string | null; close(): Promise; @@ -276,12 +285,19 @@ export interface RuntimeSqlDatabase { export interface RuntimeSqlTransactionDatabase { exec(sql: string): Promise; + execSync(sql: string): RuntimeSqlExecResult; execute( sql: string, params?: RuntimeSqlBindParams, ): Promise; + executeSync( + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult; commit(): Promise; + commitSync(): void; rollback(): Promise; + rollbackSync(): void; } export interface RuntimeActorConfig { @@ -615,11 +631,20 @@ export interface CoreRuntime { ctx: ActorContextHandle, sql: string, ): Promise; + actorSqlExecSync( + ctx: ActorContextHandle, + sql: string, + ): RuntimeSqlExecResult; actorSqlExecute( ctx: ActorContextHandle, sql: string, params?: RuntimeSqlBindParams, ): Promise; + actorSqlExecuteSync( + ctx: ActorContextHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult; actorSqlExecuteBatch( ctx: ActorContextHandle, statements: RuntimeSqlBatchStatement[], @@ -629,21 +654,37 @@ export interface CoreRuntime { timeoutMs?: number, name?: string, ): Promise; + actorSqlBeginTransactionSync( + ctx: ActorContextHandle, + timeoutMs?: number, + name?: string, + ): SqliteTransactionHandle; actorSqlTransactionExec( transaction: SqliteTransactionHandle, sql: string, ): Promise; + actorSqlTransactionExecSync( + transaction: SqliteTransactionHandle, + sql: string, + ): RuntimeSqlExecResult; actorSqlTransactionExecute( transaction: SqliteTransactionHandle, sql: string, params?: RuntimeSqlBindParams, ): Promise; + actorSqlTransactionExecuteSync( + transaction: SqliteTransactionHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult; actorSqlTransactionCommit( transaction: SqliteTransactionHandle, ): Promise; + actorSqlTransactionCommitSync(transaction: SqliteTransactionHandle): void; actorSqlTransactionRollback( transaction: SqliteTransactionHandle, ): Promise; + actorSqlTransactionRollbackSync(transaction: SqliteTransactionHandle): void; actorBeginStateTransaction( ctx: ActorContextHandle, timeoutMs?: number, @@ -653,6 +694,11 @@ export interface CoreRuntime { sql: string, params?: RuntimeSqlBindParams, ): Promise; + actorStateTransactionExecuteSync( + transaction: ActorStateTransactionHandle, + sql: string, + params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult; actorStateTransactionCommit( transaction: ActorStateTransactionHandle, payload: RuntimeStateDeltaPayload, diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/wasm-runtime.ts b/rivetkit-typescript/packages/rivetkit/src/registry/wasm-runtime.ts index b0fb460ccf..0386dff180 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/wasm-runtime.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/wasm-runtime.ts @@ -145,6 +145,12 @@ function wasmNumber(value: number | bigint): number { return typeof value === "bigint" ? Number(value) : value; } +function synchronousSqliteUnavailable(): never { + throw new Error( + "Synchronous SQLite operations are only available in the Node.js native runtime.", + ); +} + function normalizeKvEntry(entry: RuntimeKvEntry): RuntimeKvEntry { return { key: toBytes(entry.key), @@ -757,6 +763,13 @@ export class WasmCoreRuntime implements CoreRuntime { return await callWasm(() => this.#actorSql(ctx).exec(sql)); } + actorSqlExecSync( + _ctx: ActorContextHandle, + _sql: string, + ): RuntimeSqlExecResult { + return synchronousSqliteUnavailable(); + } + async actorSqlExecute( ctx: ActorContextHandle, sql: string, @@ -768,6 +781,14 @@ export class WasmCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorSqlExecuteSync( + _ctx: ActorContextHandle, + _sql: string, + _params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + return synchronousSqliteUnavailable(); + } + async actorSqlExecuteBatch( ctx: ActorContextHandle, statements: RuntimeSqlBatchStatement[], @@ -788,6 +809,14 @@ export class WasmCoreRuntime implements CoreRuntime { )) as unknown as SqliteTransactionHandle; } + actorSqlBeginTransactionSync( + _ctx: ActorContextHandle, + _timeoutMs?: number, + _name?: string, + ): SqliteTransactionHandle { + return synchronousSqliteUnavailable(); + } + async actorSqlTransactionExec( transaction: SqliteTransactionHandle, sql: string, @@ -801,6 +830,13 @@ export class WasmCoreRuntime implements CoreRuntime { ); } + actorSqlTransactionExecSync( + _transaction: SqliteTransactionHandle, + _sql: string, + ): RuntimeSqlExecResult { + return synchronousSqliteUnavailable(); + } + async actorSqlTransactionExecute( transaction: SqliteTransactionHandle, sql: string, @@ -819,6 +855,14 @@ export class WasmCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorSqlTransactionExecuteSync( + _transaction: SqliteTransactionHandle, + _sql: string, + _params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + return synchronousSqliteUnavailable(); + } + async actorSqlTransactionCommit( transaction: SqliteTransactionHandle, ): Promise { @@ -827,6 +871,10 @@ export class WasmCoreRuntime implements CoreRuntime { ); } + actorSqlTransactionCommitSync(_transaction: SqliteTransactionHandle): void { + synchronousSqliteUnavailable(); + } + async actorSqlTransactionRollback( transaction: SqliteTransactionHandle, ): Promise { @@ -836,6 +884,13 @@ export class WasmCoreRuntime implements CoreRuntime { ).rollback(), ); } + + actorSqlTransactionRollbackSync( + _transaction: SqliteTransactionHandle, + ): void { + synchronousSqliteUnavailable(); + } + async actorBeginStateTransaction( ctx: ActorContextHandle, timeoutMs?: number, @@ -869,6 +924,14 @@ export class WasmCoreRuntime implements CoreRuntime { return normalizeRuntimeSqlExecuteResult(result); } + actorStateTransactionExecuteSync( + _transaction: ActorStateTransactionHandle, + _sql: string, + _params?: RuntimeSqlBindParams, + ): RuntimeSqlExecuteResult { + return synchronousSqliteUnavailable(); + } + async actorStateTransactionCommit( transaction: ActorStateTransactionHandle, payload: RuntimeStateDeltaPayload, diff --git a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts index 304969f794..5c11d5a290 100644 --- a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts +++ b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts @@ -707,6 +707,40 @@ describeDriverMatrix( dbTestTimeout, ); + test( + "supports synchronous queries only in the Node.js native runtime", + async (c) => { + const { client } = await setupDriverTest( + c, + driverTestConfig, + ); + const actor = getDbActor(client, variant).getOrCreate([ + `db-${variant}-sync-${crypto.randomUUID()}`, + ]); + + if (driverTestConfig.runtime === "wasm") { + await expect( + actor.synchronousQueries("sync"), + ).rejects.toThrow( + "only available in the Node.js native runtime", + ); + return; + } + + await expect( + actor.synchronousQueries("sync"), + ).resolves.toEqual({ + value: "sync", + // exec() returns rows from the final statement, matching + // the existing native database contract. + multiStatementValues: [2], + transactionCount: 2, + rollbackCount: 0, + }); + }, + dbTestTimeout, + ); + test( "handles transactions", async (c) => { diff --git a/rivetkit-typescript/packages/rivetkit/tests/wasm-runtime.test.ts b/rivetkit-typescript/packages/rivetkit/tests/wasm-runtime.test.ts index a9bf9d42d2..41dfca431b 100644 --- a/rivetkit-typescript/packages/rivetkit/tests/wasm-runtime.test.ts +++ b/rivetkit-typescript/packages/rivetkit/tests/wasm-runtime.test.ts @@ -243,6 +243,59 @@ describe("WasmCoreRuntime", () => { acceptRuntime(new NapiCoreRuntime({} as never)); }); + test("routes synchronous SQLite lifecycle through NAPI and rejects it in wasm", () => { + const executeSync = vi.fn(() => ({ + columns: ["value"], + rows: [[42]], + changes: 0, + lastInsertRowId: null, + })); + const transaction = { + commitSync: vi.fn(), + rollbackSync: vi.fn(), + }; + const beginTransactionSync = vi.fn(() => transaction); + const context = { + sql: () => ({ executeSync, beginTransactionSync }), + } as unknown as ActorContextHandle; + const params = [{ kind: "int" as const, intValue: 42 }]; + const napiRuntime = new NapiCoreRuntime({} as never); + + expect( + napiRuntime.actorSqlExecuteSync(context, "SELECT ?", params), + ).toEqual({ + columns: ["value"], + rows: [[42]], + changes: 0, + lastInsertRowId: null, + }); + expect(executeSync).toHaveBeenCalledWith("SELECT ?", params); + const transactionHandle = napiRuntime.actorSqlBeginTransactionSync( + context, + 1_000, + "sync-test", + ); + napiRuntime.actorSqlTransactionCommitSync(transactionHandle); + napiRuntime.actorSqlTransactionRollbackSync(transactionHandle); + expect(beginTransactionSync).toHaveBeenCalledWith(1_000, "sync-test"); + expect(transaction.commitSync).toHaveBeenCalledOnce(); + expect(transaction.rollbackSync).toHaveBeenCalledOnce(); + + const wasmRuntime = new WasmCoreRuntime(fakeWasmBindings()); + expect(() => + wasmRuntime.actorSqlExecuteSync(context, "SELECT ?", params), + ).toThrow("only available in the Node.js native runtime"); + expect(() => wasmRuntime.actorSqlBeginTransactionSync(context)).toThrow( + "only available in the Node.js native runtime", + ); + expect(() => + wasmRuntime.actorSqlTransactionCommitSync(transactionHandle), + ).toThrow("only available in the Node.js native runtime"); + expect(() => + wasmRuntime.actorSqlTransactionRollbackSync(transactionHandle), + ).toThrow("only available in the Node.js native runtime"); + }); + test("maps raw wasm registry, factory, and cancellation handles", () => { const runtime = new WasmCoreRuntime(fakeWasmBindings()); const registry = runtime.createRegistry(); diff --git a/scripts/publish/package.json b/scripts/publish/package.json index c717d22f77..976fb31b8f 100644 --- a/scripts/publish/package.json +++ b/scripts/publish/package.json @@ -6,13 +6,15 @@ "scripts": { "ci": "tsx src/ci/bin.ts", "release": "tsx src/local/cut-release.ts", + "test": "tsx --test 'src/**/*.test.ts'", "check-types": "tsc --noEmit" }, "dependencies": { "commander": "^12.1.0", "execa": "^8.0.1", "glob": "^10.3.10", - "semver": "^7.6.0" + "semver": "^7.6.0", + "yaml": "^2.9.0" }, "devDependencies": { "@types/node": "^24.3.0", diff --git a/scripts/publish/src/lib/version.test.ts b/scripts/publish/src/lib/version.test.ts new file mode 100644 index 0000000000..e9d4ed4bae --- /dev/null +++ b/scripts/publish/src/lib/version.test.ts @@ -0,0 +1,84 @@ +import assert from "node:assert/strict"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +import { + bumpPackageJsons, + parseWorkspaceCatalogs, + resolveCatalogDependency, +} from "./version.js"; + +test("resolves default and named pnpm catalog dependencies", () => { + const catalogs = parseWorkspaceCatalogs(` +catalog: + drizzle-orm: "0.45.2" +catalogs: + react19: + react: "^19.0.0" +`); + + assert.equal( + resolveCatalogDependency("drizzle-orm", "catalog:", catalogs), + "0.45.2", + ); + assert.equal( + resolveCatalogDependency("react", "catalog:react19", catalogs), + "^19.0.0", + ); + assert.equal(resolveCatalogDependency("zod", "^4.0.0", catalogs), undefined); +}); + +test("rejects unresolved catalog references before publication", () => { + const catalogs = parseWorkspaceCatalogs("catalog:\n drizzle-orm: 0.45.2\n"); + + assert.throws( + () => resolveCatalogDependency("missing", "catalog:", catalogs), + /missing from pnpm catalog default/, + ); + assert.throws( + () => resolveCatalogDependency("react", "catalog:react19", catalogs), + /missing pnpm catalog react19/, + ); +}); + +test("rewrites catalog specs in the manifest passed to npm", async () => { + const root = await mkdtemp(join(tmpdir(), "rivet-publish-catalog-")); + const packageDirectory = join(root, "packages", "rivetkit"); + try { + await mkdir(packageDirectory, { recursive: true }); + await writeFile( + join(root, "pnpm-workspace.yaml"), + 'catalog:\n drizzle-orm: "0.45.2"\npackages:\n - packages/*\n', + ); + await writeFile( + join(packageDirectory, "package.json"), + JSON.stringify({ + dependencies: { "drizzle-orm": "catalog:" }, + name: "rivetkit", + version: "1.0.0", + }), + ); + + await bumpPackageJsons(root, "0.0.0-preview.abcdef0", { + repository: "rivet-dev/rivet", + }); + const manifest: unknown = JSON.parse( + await readFile(join(packageDirectory, "package.json"), "utf8"), + ); + if (typeof manifest !== "object" || manifest === null) { + throw new Error("rewritten package manifest is not an object"); + } + const dependencies: unknown = Reflect.get(manifest, "dependencies"); + if (typeof dependencies !== "object" || dependencies === null) { + throw new Error("rewritten package dependencies are not an object"); + } + assert.equal( + Reflect.get(dependencies, "drizzle-orm"), + "0.45.2", + ); + } finally { + await rm(root, { force: true, recursive: true }); + } +}); diff --git a/scripts/publish/src/lib/version.ts b/scripts/publish/src/lib/version.ts index 3752c00fc3..44db3d77da 100644 --- a/scripts/publish/src/lib/version.ts +++ b/scripts/publish/src/lib/version.ts @@ -20,6 +20,7 @@ import { join, resolve as resolvePath } from "node:path"; import { $ } from "execa"; import { glob } from "glob"; import * as semver from "semver"; +import { parse } from "yaml"; import { scoped } from "./logger.js"; import { buildMetaPlatformMap, @@ -55,6 +56,76 @@ const DEP_FIELDS = [ "optionalDependencies", ] as const; +export interface WorkspaceCatalogs { + default: ReadonlyMap; + named: ReadonlyMap>; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function parseCatalog( + value: unknown, + label: string, +): ReadonlyMap { + if (value === undefined) return new Map(); + if (!isRecord(value)) { + throw new Error(`${label} must be a dependency-to-version mapping`); + } + const catalog = new Map(); + for (const [dependency, spec] of Object.entries(value)) { + if (typeof spec !== "string" || spec.length === 0) { + throw new Error(`${label}.${dependency} must be a non-empty string`); + } + catalog.set(dependency, spec); + } + return catalog; +} + +export function parseWorkspaceCatalogs(source: string): WorkspaceCatalogs { + const parsed: unknown = parse(source); + if (!isRecord(parsed)) { + throw new Error("pnpm-workspace.yaml must contain a mapping"); + } + const named = new Map>(); + if (parsed.catalogs !== undefined) { + if (!isRecord(parsed.catalogs)) { + throw new Error("catalogs must be a named catalog mapping"); + } + for (const [name, value] of Object.entries(parsed.catalogs)) { + named.set(name, parseCatalog(value, `catalogs.${name}`)); + } + } + return { + default: parseCatalog(parsed.catalog, "catalog"), + named, + }; +} + +export function resolveCatalogDependency( + dependency: string, + spec: string, + catalogs: WorkspaceCatalogs, +): string | undefined { + if (!spec.startsWith("catalog:")) return undefined; + const catalogName = spec.slice("catalog:".length); + const catalog = + catalogName.length === 0 ? catalogs.default : catalogs.named.get(catalogName); + if (catalog === undefined) { + throw new Error( + `dependency ${dependency} references missing pnpm catalog ${catalogName}`, + ); + } + const resolved = catalog.get(dependency); + if (resolved === undefined) { + throw new Error( + `dependency ${dependency} is missing from pnpm catalog ${catalogName || "default"}`, + ); + } + return resolved; +} + const PUBLISHED_RUST_WORKSPACE_DEPS = new Set([ "rivet-error-macros", "rivet-error", @@ -148,6 +219,11 @@ export async function bumpPackageJsons( const packageNames = new Set(packages.map((p) => p.name)); const metaPlatformMap = buildMetaPlatformMap(packages); const versionOnly = opts.versionOnly ?? false; + const catalogs = versionOnly + ? undefined + : parseWorkspaceCatalogs( + await fs.readFile(join(repoRoot, "pnpm-workspace.yaml"), "utf8"), + ); // Cache `npm view version` lookups for out-of-scope dependencies so a // dep referenced by several packages is only resolved once. @@ -195,6 +271,17 @@ export async function bumpPackageJsons( const deps = pkgJson[field]; if (!deps) continue; for (const [dep, spec] of Object.entries(deps)) { + const catalogVersion = + catalogs === undefined + ? undefined + : resolveCatalogDependency(dep, spec, catalogs); + if (catalogVersion !== undefined) { + deps[dep] = catalogVersion; + log.info( + `resolving catalog dep ${pkg.name} -> ${dep}@${catalogVersion}`, + ); + continue; + } const isWorkspace = typeof spec === "string" && spec.startsWith("workspace:"); if (!isWorkspace) continue;