Implement Turso support (basically SQLite)

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- The `riversqlite` driver is now tested against Turso, an in-process SQLite-compatible database written in Rust. [PR #1311](https://github.com/riverqueue/river/pull/1311).
+
 ### Changed
 
 - Both hooks and middleware should now be registered under a general `Config.Plugins` instead, a change that simpifies things somewhat, and better handles cases where a hook or middleware might be _both_ a hook and a middleware as opposed to only one or the other. `Config.Hooks` and `Config.Middleware` are still available for use, but considered deprecated in favor of the more general `Config.Plugins`. [PR #1284](https://github.com/riverqueue/river/pull/1284).

riverdbtest/riverdbtest.go

@@ -245,6 +245,12 @@ func TestSchema[TTx any](ctx context.Context, tb testutil.TestingTB, driver rive
 
 		var sb strings.Builder
 		sb.WriteString(driver.DatabaseName())
+		if opts.ProcurePool != nil {
+			// SQLite and Turso both use riversqlite and therefore report the same
+			// database name, but their database files aren't compatible. Keep their
+			// idle schemas separate based on the function that procures their pool.
+			fmt.Fprintf(&sb, ",procure_pool:%p", opts.ProcurePool)
+		}
 
 		for _, line := range lines {
 			sb.WriteString(",")

riverdriver/riverdrivertest/driver_client_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	_ "github.com/tursodatabase/libsql-client-go/libsql"
 	_ "modernc.org/sqlite"
+	_ "turso.tech/database/tursogo"
 
 	"github.com/riverqueue/river"
 	"github.com/riverqueue/river/riverdbtest"
@@ -131,6 +132,28 @@ func TestClientWithDriverRiverSQLiteModernC(t *testing.T) {
 	)
 }
 
+func TestClientWithDriverRiverTurso(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	ExerciseClient(ctx, t,
+		func(ctx context.Context, t *testing.T) (riverdriver.Driver[*sql.Tx], string) {
+			t.Helper()
+
+			var (
+				driver = riversqlite.New(nil)
+				schema = riverdbtest.TestSchema(ctx, t, driver, &riverdbtest.TestSchemaOpts{
+					ProcurePool: func(ctx context.Context, schema string) (any, string) {
+						return riversharedtest.DBPoolTurso(ctx, t, schema), "" // could also be `main` instead of empty string
+					},
+				})
+			)
+			return driver, schema
+		},
+	)
+}
+
 type noOpArgs struct {
 	Name string `json:"name"`
 }

riverdriver/riverdrivertest/driver_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/stretchr/testify/require"
 	_ "github.com/tursodatabase/libsql-client-go/libsql"
 	_ "modernc.org/sqlite"
+	_ "turso.tech/database/tursogo"
 
 	"github.com/riverqueue/river"
 	"github.com/riverqueue/river/riverdbtest"
@@ -255,6 +256,58 @@ func TestDriverRiverSQLiteModernC(t *testing.T) { //nolint:dupl
 		})
 }
 
+func TestDriverRiverTurso(t *testing.T) { //nolint:dupl
+	t.Parallel()
+
+	var (
+		ctx         = context.Background()
+		procurePool = func(ctx context.Context, schema string) (any, string) {
+			return riversharedtest.DBPoolTurso(ctx, t, schema), "" // could also be `main` instead of empty string
+		}
+	)
+
+	riverdrivertest.Exercise(ctx, t,
+		func(ctx context.Context, t *testing.T, opts *riverdbtest.TestSchemaOpts) (riverdriver.Driver[*sql.Tx], string) {
+			t.Helper()
+
+			if opts == nil {
+				opts = &riverdbtest.TestSchemaOpts{}
+			}
+			opts.ProcurePool = procurePool
+
+			var (
+				// Driver will have its pool set by TestSchema.
+				driver = riversqlite.New(nil)
+				schema = riverdbtest.TestSchema(ctx, t, driver, opts)
+			)
+			return driver, schema
+		},
+		func(ctx context.Context, t *testing.T) (riverdriver.Executor, riverdriver.Driver[*sql.Tx]) {
+			t.Helper()
+
+			// Driver will have its pool set by TestSchema.
+			driver := riversqlite.New(nil)
+
+			tx, _ := riverdbtest.TestTx(ctx, t, driver, &riverdbtest.TestTxOpts{
+				// Unfortunately, the normal test transaction schema sharing has
+				// to be disabled for SQLite. When enabled, there's too much
+				// contention on the shared test databases and operations fail
+				// with `database is locked (5) (SQLITE_BUSY)`, which is a
+				// common concurrency error in SQLite whose recommended
+				// remediation is a backoff and retry. I tried various
+				// techniques like journal_mode=WAL, but it didn't seem to help
+				// enough. SQLite databases are just local files anyway, and
+				// test transactions can still reuse schemas freed by other
+				// tests through TestSchema, so this should be okay performance
+				// wise.
+				DisableSchemaSharing: true,
+
+				ProcurePool: procurePool,
+			})
+			return driver.UnwrapExecutor(tx), driver
+		})
+}
+
 func BenchmarkDriverRiverDatabaseSQLLibPQ(b *testing.B) {
 	ctx := context.Background()
 

riverdriver/riverdrivertest/example_turso_test.go

@@ -0,0 +1,58 @@
+package riverdrivertest_test
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	_ "turso.tech/database/tursogo"
+
+	"github.com/riverqueue/river"
+	"github.com/riverqueue/river/riverdriver/riversqlite"
+)
+
+// Example_turso demonstrates use of River's SQLite driver with a local Turso
+// database.
+func Example_turso() {
+	ctx := context.Background()
+
+	tempDir, err := os.MkdirTemp("", "river-example-turso-*")
+	if err != nil {
+		panic(err)
+	}
+	defer os.RemoveAll(tempDir)
+
+	dbPool, err := sql.Open("turso", filepath.Join(tempDir, "example_turso.db"))
+	if err != nil {
+		panic(err)
+	}
+	dbPool.SetMaxOpenConns(1)
+	defer dbPool.Close()
+
+	driver := riversqlite.New(dbPool)
+
+	if err := migrateDB(ctx, driver); err != nil {
+		panic(err)
+	}
+
+	riverClient, err := river.NewClient(driver, initTestConfig(ctx, nil, &river.Config{}))
+	if err != nil {
+		panic(err)
+	}
+
+	insertResult, err := riverClient.Insert(ctx, SortArgs{
+		Strings: []string{
+			"whale", "tiger", "bear",
+		},
+	}, nil)
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Printf("Inserted job kind: %s\n", insertResult.Job.Kind)
+
+	// Output:
+	// Inserted job kind: sort
+}

riverdriver/riverdrivertest/go.mod

@@ -19,15 +19,17 @@ require (
 	github.com/stretchr/testify v1.11.1
 	github.com/tidwall/gjson v1.19.0
 	github.com/tidwall/sjson v1.2.5
-	github.com/tursodatabase/libsql-client-go v0.0.0-20240902231107-85af5b9d094d
+	github.com/tursodatabase/libsql-client-go v0.0.0-20260528064733-9d5d30a29a60
 	golang.org/x/text v0.40.0
 	modernc.org/sqlite v1.53.0
+	turso.tech/database/tursogo v0.7.0
 )
 
 require (
 	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/coder/websocket v1.8.12 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/ebitengine/purego v0.9.1 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
@@ -38,6 +40,7 @@ require (
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/tidwall/match v1.2.0 // indirect
 	github.com/tidwall/pretty v1.2.1 // indirect
+	github.com/tursodatabase/turso-go-platform-libs v0.7.0 // indirect
 	go.uber.org/goleak v1.3.0 // indirect
 	golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 // indirect
 	golang.org/x/sync v0.22.0 // indirect

riverdriver/riverdrivertest/go.sum

@@ -7,6 +7,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/ebitengine/purego v0.9.1 h1:a/k2f2HQU3Pi399RPW1MOaZyhKJL9w/xFpKAg4q1s0A=
+github.com/ebitengine/purego v0.9.1/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
 github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
@@ -31,6 +33,8 @@ github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ=
 github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
 github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mattn/go-sqlite3 v1.14.42 h1:MigqEP4ZmHw3aIdIT7T+9TLa90Z6smwcthx+Azv4Cgo=
+github.com/mattn/go-sqlite3 v1.14.42/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ=
 github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w=
 github.com/ncruces/go-strftime v1.0.0/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -71,8 +75,10 @@ github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
 github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
 github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
-github.com/tursodatabase/libsql-client-go v0.0.0-20240902231107-85af5b9d094d h1:dOMI4+zEbDI37KGb0TI44GUAwxHF9cMsIoDTJ7UmgfU=
-github.com/tursodatabase/libsql-client-go v0.0.0-20240902231107-85af5b9d094d/go.mod h1:l8xTsYB90uaVdMHXMCxKKLSgw5wLYBwBKKefNIUnm9s=
+github.com/tursodatabase/libsql-client-go v0.0.0-20260528064733-9d5d30a29a60 h1:TfQEwhr0Q9t+Bgs0TNk2eHZ9EGD107Mimic0kcoGS1M=
+github.com/tursodatabase/libsql-client-go v0.0.0-20260528064733-9d5d30a29a60/go.mod h1:08inkKyguB6CGGssc/JzhmQWwBgFQBgjlYFjxjRh7nU=
+github.com/tursodatabase/turso-go-platform-libs v0.7.0 h1:NPTPklRhKsPwuwpvvCyJRPklXExdLyP1iRN9meQm1W4=
+github.com/tursodatabase/turso-go-platform-libs v0.7.0/go.mod h1:bo+Lpv5OYOX1gRV9L5DLKMsYxmDs56SkZwnCOLEFcxU=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 golang.org/x/exp v0.0.0-20251023183803-a4bb9ffd2546 h1:mgKeJMpvi0yx/sU5GsxQ7p6s2wtOnGAHZWCHUM4KGzY=
@@ -122,3 +128,5 @@ modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
 modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
 modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
 modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=
+turso.tech/database/tursogo v0.7.0 h1:SER1jXs1rAFlUlibX5YL3Qf9DzcFRCKXee+Oi0qrcN8=
+turso.tech/database/tursogo v0.7.0/go.mod h1:6CMOLO6jRD3PcPfyngEDISFonVj+Ucx5LBAtQI0GuJ0=

riverdriver/riverdrivertest/riverdrivertest.go

@@ -106,6 +106,7 @@ func requireMissingRelation(t *testing.T, err error, schema, missingRelation str
 	} else {
 		// lib/pq: pq: relation %s.%s does not exist
 		// SQLite: no such table: %s.%s
-		require.Regexp(t, fmt.Sprintf(`(pq: relation "%s\.%s" does not exist|no such table: %s\.%s)`, schema, missingRelation, schema, missingRelation), err.Error())
+		// Turso: turso: error: Invalid argument supplied: no such database: %s
+		require.Regexp(t, fmt.Sprintf(`(pq: relation "%s\.%s" does not exist|no such table: %s\.%s|no such database: %s)`, schema, missingRelation, schema, missingRelation, schema), err.Error())
 	}
 }

riverdriver/riversqlite/internal/dbsqlc/river_job.sql

@@ -486,7 +486,7 @@ LIMIT @max;
 -- name: JobRescue :exec
 UPDATE /* TEMPLATE: schema */river_job
 SET
-    errors = jsonb_insert(coalesce(errors, jsonb('[]')), '$[#]', jsonb(@error)),
+    errors = jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error))),
     finalized_at = cast(sqlc.narg('finalized_at') as text),
     scheduled_at = @scheduled_at,
     metadata = jsonb_set(
@@ -572,7 +572,7 @@ RETURNING *;
 
 -- name: JobScheduleSetDiscarded :many
 UPDATE /* TEMPLATE: schema */river_job
-SET metadata = jsonb_patch(metadata, jsonb('{"unique_key_conflict": "scheduler_discarded"}')),
+SET metadata = jsonb_patch(json(metadata), json('{"unique_key_conflict": "scheduler_discarded"}')),
     finalized_at = coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec')),
     state = 'discarded'
 WHERE id IN (sqlc.slice('id'))
@@ -582,7 +582,7 @@ RETURNING *;
 -- for JobSetStateIfRunning to use when falling back to non-running jobs.
 -- name: JobSetMetadataIfNotRunning :one
 UPDATE /* TEMPLATE: schema */river_job
-SET metadata = jsonb_patch(metadata, jsonb(@metadata_updates))
+SET metadata = jsonb_patch(json(metadata), json(@metadata_updates))
 WHERE id = @id
     AND state != 'running'
 RETURNING *;
@@ -600,15 +600,15 @@ SET
                         THEN @attempt
                         ELSE attempt END,
     errors       = CASE WHEN cast(@errors_do_update AS boolean)
-                        THEN jsonb_insert(coalesce(errors, jsonb('[]')), '$[#]', jsonb(@error))
+                        THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(@error)))
                         ELSE errors END,
     finalized_at = CASE WHEN /* should_cancel */((@state = 'retryable' OR @state = 'scheduled') AND (metadata -> 'cancel_attempted_at') iS NOT NULL)
                         THEN coalesce(cast(sqlc.narg('now') AS text), datetime('now', 'subsec'))
                         WHEN cast(@finalized_at_do_update AS boolean)
                         THEN @finalized_at
                         ELSE finalized_at END,
     metadata     = CASE WHEN cast(@metadata_do_merge AS boolean)
-                        THEN jsonb_patch(metadata, jsonb(@metadata_updates))
+                        THEN jsonb_patch(json(metadata), json(@metadata_updates))
                         ELSE metadata END,
     scheduled_at = CASE WHEN /* NOT should_cancel */(cast(@state AS text) <> 'retryable' AND @state <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(@scheduled_at_do_update AS boolean)
                         THEN @scheduled_at
@@ -623,7 +623,7 @@ RETURNING *;
 -- name: JobUpdate :one
 UPDATE /* TEMPLATE: schema */river_job
 SET
-    metadata = CASE WHEN cast(@metadata_do_merge AS boolean) THEN jsonb_patch(metadata, jsonb(@metadata)) ELSE metadata END
+    metadata = CASE WHEN cast(@metadata_do_merge AS boolean) THEN jsonb_patch(json(metadata), json(@metadata)) ELSE metadata END
 WHERE id = @id
 RETURNING *;
 

riverdriver/riversqlite/internal/dbsqlc/river_queue.sql

@@ -29,7 +29,7 @@ DELETE FROM /* TEMPLATE: schema */river_queue
 WHERE name IN (
     SELECT name
     FROM /* TEMPLATE: schema */river_queue
-    WHERE river_queue.updated_at < @updated_at_horizon
+    WHERE river_queue.updated_at < cast(@updated_at_horizon AS text)
     ORDER BY name ASC
     LIMIT @max
 )