From 487158fb667103a8b718ef078e30adb50d27a688 Mon Sep 17 00:00:00 2001 From: Brandur Date: Mon, 22 Jun 2026 15:03:05 -0600 Subject: [PATCH] Implement Turso support (basically SQLite) Here, start testing against Turso, which is a SQLite-compatible database rewritten in Rust and with some nice additional features. There are a couple minor tweaks we need to make to SQLite queries to make this work, but nothing major, so the idea here is that like libSQL, we can get Turso support without any major maintenance overhead and hopefully make it a bit of a PR win. I tried to do this originally a few months ago, but Turso had a few bugs around JSON handling that we couldn't work our way around. These were fixed recently, so now it's possible to do. --- CHANGELOG.md | 4 ++ riverdbtest/riverdbtest.go | 6 ++ .../riverdrivertest/driver_client_test.go | 23 +++++++ riverdriver/riverdrivertest/driver_test.go | 53 +++++++++++++++ .../riverdrivertest/example_turso_test.go | 58 ++++++++++++++++ riverdriver/riverdrivertest/go.mod | 5 +- riverdriver/riverdrivertest/go.sum | 12 +++- .../riverdrivertest/riverdrivertest.go | 3 +- .../riversqlite/internal/dbsqlc/river_job.sql | 12 ++-- .../internal/dbsqlc/river_job.sql.go | 12 ++-- .../internal/dbsqlc/river_queue.sql | 2 +- .../internal/dbsqlc/river_queue.sql.go | 5 +- .../main/003_river_job_tags_non_null.down.sql | 4 +- .../main/003_river_job_tags_non_null.up.sql | 4 +- .../riversqlite/river_sqlite_driver.go | 66 ++++++++++++------- .../riversharedtest/riversharedtest.go | 18 ++++- 16 files changed, 239 insertions(+), 48 deletions(-) create mode 100644 riverdriver/riverdrivertest/example_turso_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ced190a..2d60c4f3 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/riverdbtest/riverdbtest.go b/riverdbtest/riverdbtest.go index acdb3db7..64468ca9 100644 --- a/riverdbtest/riverdbtest.go +++ b/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(",") diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go index 8f99cec4..923c7a2b 100644 --- a/riverdriver/riverdrivertest/driver_client_test.go +++ b/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"` } diff --git a/riverdriver/riverdrivertest/driver_test.go b/riverdriver/riverdrivertest/driver_test.go index 50d30675..367ee972 100644 --- a/riverdriver/riverdrivertest/driver_test.go +++ b/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() diff --git a/riverdriver/riverdrivertest/example_turso_test.go b/riverdriver/riverdrivertest/example_turso_test.go new file mode 100644 index 00000000..4aa976d5 --- /dev/null +++ b/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 +} diff --git a/riverdriver/riverdrivertest/go.mod b/riverdriver/riverdrivertest/go.mod index 12f8403d..353da0b9 100644 --- a/riverdriver/riverdrivertest/go.mod +++ b/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 diff --git a/riverdriver/riverdrivertest/go.sum b/riverdriver/riverdrivertest/go.sum index a7330f8e..195bcf2c 100644 --- a/riverdriver/riverdrivertest/go.sum +++ b/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= diff --git a/riverdriver/riverdrivertest/riverdrivertest.go b/riverdriver/riverdrivertest/riverdrivertest.go index ad4eb976..b970f80d 100644 --- a/riverdriver/riverdrivertest/riverdrivertest.go +++ b/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()) } } diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql index b0600dc9..46eb1434 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/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,7 +600,7 @@ 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')) @@ -608,7 +608,7 @@ SET 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 *; diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go index 20b64838..e886476d 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1241,7 +1241,7 @@ func (q *Queries) JobList(ctx context.Context, db DBTX, max int64) ([]*RiverJob, const jobRescue = `-- name: JobRescue :exec UPDATE /* TEMPLATE: schema */river_job SET - errors = jsonb_insert(coalesce(errors, jsonb('[]')), '$[#]', jsonb(?1)), + errors = jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?1))), finalized_at = cast(?2 as text), scheduled_at = ?3, metadata = jsonb_set( @@ -1522,7 +1522,7 @@ func (q *Queries) JobScheduleSetAvailable(ctx context.Context, db DBTX, id []int const jobScheduleSetDiscarded = `-- 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(?1 AS text), datetime('now', 'subsec')), state = 'discarded' WHERE id IN (/*SLICE:id*/?) @@ -1589,7 +1589,7 @@ func (q *Queries) JobScheduleSetDiscarded(ctx context.Context, db DBTX, arg *Job const jobSetMetadataIfNotRunning = `-- name: JobSetMetadataIfNotRunning :one UPDATE /* TEMPLATE: schema */river_job -SET metadata = jsonb_patch(metadata, jsonb(?1)) +SET metadata = jsonb_patch(json(metadata), json(?1)) WHERE id = ?2 AND state != 'running' RETURNING id, json(args), attempt, attempted_at, json(attempted_by), created_at, json(errors), finalized_at, kind, max_attempts, json(metadata), priority, queue, state, scheduled_at, json(tags), unique_key, unique_states @@ -1638,7 +1638,7 @@ SET THEN ?3 ELSE attempt END, errors = CASE WHEN cast(?4 AS boolean) - THEN jsonb_insert(coalesce(errors, jsonb('[]')), '$[#]', jsonb(?5)) + THEN jsonb(json_insert(json(coalesce(errors, jsonb('[]'))), '$[#]', json(?5))) ELSE errors END, finalized_at = CASE WHEN /* should_cancel */((?1 = 'retryable' OR ?1 = 'scheduled') AND (metadata -> 'cancel_attempted_at') iS NOT NULL) THEN coalesce(cast(?6 AS text), datetime('now', 'subsec')) @@ -1646,7 +1646,7 @@ SET THEN ?8 ELSE finalized_at END, metadata = CASE WHEN cast(?9 AS boolean) - THEN jsonb_patch(metadata, jsonb(?10)) + THEN jsonb_patch(json(metadata), json(?10)) ELSE metadata END, scheduled_at = CASE WHEN /* NOT should_cancel */(cast(?1 AS text) <> 'retryable' AND ?1 <> 'scheduled' OR (metadata -> 'cancel_attempted_at') IS NULL) AND cast(?11 AS boolean) THEN ?12 @@ -1721,7 +1721,7 @@ func (q *Queries) JobSetStateIfRunning(ctx context.Context, db DBTX, arg *JobSet const jobUpdate = `-- name: JobUpdate :one UPDATE /* TEMPLATE: schema */river_job SET - metadata = CASE WHEN cast(?1 AS boolean) THEN jsonb_patch(metadata, jsonb(?2)) ELSE metadata END + metadata = CASE WHEN cast(?1 AS boolean) THEN jsonb_patch(json(metadata), json(?2)) ELSE metadata END WHERE id = ?3 RETURNING id, json(args), attempt, attempted_at, json(attempted_by), created_at, json(errors), finalized_at, kind, max_attempts, json(metadata), priority, queue, state, scheduled_at, json(tags), unique_key, unique_states ` diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql b/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql index 7e769d3f..3071dc6f 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql +++ b/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 ) diff --git a/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql.go b/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql.go index fa52751c..8cadb45d 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_queue.sql.go @@ -8,7 +8,6 @@ package dbsqlc import ( "context" "strings" - "time" ) const queueCreateOrSetUpdatedAt = `-- name: QueueCreateOrSetUpdatedAt :one @@ -62,7 +61,7 @@ DELETE FROM /* TEMPLATE: schema */river_queue WHERE name IN ( SELECT name FROM /* TEMPLATE: schema */river_queue - WHERE river_queue.updated_at < ?1 + WHERE river_queue.updated_at < cast(?1 AS text) ORDER BY name ASC LIMIT ?2 ) @@ -70,7 +69,7 @@ RETURNING name, created_at, json(metadata), paused_at, updated_at ` type QueueDeleteExpiredParams struct { - UpdatedAtHorizon time.Time + UpdatedAtHorizon string Max int64 } diff --git a/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.down.sql b/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.down.sql index cac0a6ce..8d314cf0 100644 --- a/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.down.sql +++ b/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.down.sql @@ -1,4 +1,6 @@ -- -- Normally `river_job.tags` is set back to nullable here, but since SQLite was -- added well after 003 came about, we push that to version 006 index. --- \ No newline at end of file +-- + +SELECT 1; diff --git a/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.up.sql b/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.up.sql index 5af35c58..d4e1e240 100644 --- a/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.up.sql +++ b/riverdriver/riversqlite/migration/main/003_river_job_tags_non_null.up.sql @@ -1,4 +1,6 @@ -- -- Normally `river_job.tags` is set to `NOT NULL` with a `DEFAULT` here, but since -- SQLite was added well after 003 came about, we push that to version 006 index. --- \ No newline at end of file +-- + +SELECT 1; diff --git a/riverdriver/riversqlite/river_sqlite_driver.go b/riverdriver/riversqlite/river_sqlite_driver.go index c7df4964..d2acda27 100644 --- a/riverdriver/riversqlite/river_sqlite_driver.go +++ b/riverdriver/riversqlite/river_sqlite_driver.go @@ -1,6 +1,5 @@ // Package riversqlite provides a River driver implementation for SQLite. It's -// also tested against libSQL (a SQLite fork), and that should continue to work -// as long they keep to their commitment in maintaining API compatibility. +// also tested against libSQL (a SQLite fork) and local Turso databases. // // This driver is currently in early testing. It's exercised reasonably // thoroughly in the test suite, but has minimal real world use as of yet. @@ -19,7 +18,6 @@ package riversqlite import ( - "cmp" "context" "database/sql" "embed" @@ -65,7 +63,7 @@ type Driver struct { } // New returns a new SQLite driver for use with River. It also works with libSQL -// (a SQLite fork). +// (a SQLite fork) and local Turso databases. // // It takes an sql.DB to use for use with River. The pool should already be // configured to use the schema specified in the client's Schema field. The pool @@ -181,17 +179,31 @@ func (e *Executor) Begin(ctx context.Context) (riverdriver.ExecutorTx, error) { } func (e *Executor) ColumnExists(ctx context.Context, params *riverdriver.ColumnExistsParams) (bool, error) { - // Unfortunately this doesn't work in sqlc because of the "table value" - // pragma isn't supported. This seems like it should be fixable, but for now - // run the raw SQL to accomplish it. - const sql = ` + // Unfortunately this doesn't work in sqlc because table-valued pragmas + // aren't supported. This seems like it should be fixable, but for now run + // raw SQL to accomplish it. + const sqlMain = ` + SELECT EXISTS ( + SELECT 1 + FROM pragma_table_info(?) + WHERE name = ? + )` + const sqlSchema = ` SELECT EXISTS ( SELECT 1 FROM pragma_table_info WHERE schema = ? AND arg = ? AND name = ? )` + + sql := sqlMain + args := []any{params.Table, params.Column} + if params.Schema != "" && params.Schema != "main" { + sql = sqlSchema + args = []any{params.Schema, params.Table, params.Column} + } + var exists int64 - if err := e.dbtx.QueryRowContext(ctx, sql, cmp.Or(params.Schema, "main"), params.Table, params.Column).Scan(&exists); err != nil { + if err := e.dbtx.QueryRowContext(ctx, sql, args...).Scan(&exists); err != nil { return false, interpretError(err) } @@ -450,23 +462,29 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel // //nolint:gochecknoglobals var jobGetAvailableAttemptedBySQL = strings.TrimSpace(` - jsonb_insert( - ( - SELECT jsonb_group_array(value) - FROM ( - SELECT * + CASE + WHEN attempted_by IS NULL + THEN jsonb(json_array(@attempted_by)) + WHEN json_array_length(attempted_by) < @max_attempted_by + THEN jsonb(json_insert(json(attempted_by), '$[#]', @attempted_by)) + ELSE jsonb(json_insert( + ( + SELECT json_group_array(value) FROM ( SELECT * - FROM json_each(attempted_by) - ORDER BY key DESC - LIMIT @max_attempted_by - 1 + FROM ( + SELECT * + FROM json_each(json(attempted_by)) + ORDER BY key DESC + LIMIT @max_attempted_by - 1 + ) + ORDER BY key ) - ORDER BY key - ) - ), - '$[#]', - @attempted_by - ) + ), + '$[#]', + @attempted_by + )) + END `) func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { @@ -1170,7 +1188,7 @@ func (e *Executor) QueueCreateOrSetUpdatedAt(ctx context.Context, params *riverd func (e *Executor) QueueDeleteExpired(ctx context.Context, params *riverdriver.QueueDeleteExpiredParams) ([]string, error) { queues, err := dbsqlc.New().QueueDeleteExpired(schemaTemplateParam(ctx, params.Schema), e.dbtx, &dbsqlc.QueueDeleteExpiredParams{ Max: int64(params.Max), - UpdatedAtHorizon: params.UpdatedAtHorizon.UTC(), + UpdatedAtHorizon: timeString(params.UpdatedAtHorizon), }) if err != nil { return nil, interpretError(err) diff --git a/rivershared/riversharedtest/riversharedtest.go b/rivershared/riversharedtest/riversharedtest.go index d6f2b5ba..cf62498d 100644 --- a/rivershared/riversharedtest/riversharedtest.go +++ b/rivershared/riversharedtest/riversharedtest.go @@ -127,12 +127,24 @@ func DBPoolSQLite(ctx context.Context, tb testing.TB, schema string) *sql.DB { return dbPoolSQLite(ctx, tb, schema, "sqlite") } +// DBPoolTurso gets a database pool appropriate for use with Turso in testing. +func DBPoolTurso(ctx context.Context, tb testing.TB, schema string) *sql.DB { + tb.Helper() + + return dbPoolSQLite(ctx, tb, schema, "turso") +} + func dbPoolSQLite(ctx context.Context, tb testing.TB, schema, driverName string) *sql.DB { //nolint:unparam tb.Helper() var databaseURLBuilder strings.Builder - databaseURLBuilder.WriteString("file:" + filepath.Join(sqliteTestDir(), schema+".sqlite3")) + databasePath := filepath.Join(sqliteTestDir(), schema+".sqlite3") + if driverName == "turso" { + databaseURLBuilder.WriteString(databasePath) + } else { + databaseURLBuilder.WriteString("file:" + databasePath) + } // This innocuous line turns out to be quite important at the tail. // @@ -163,7 +175,9 @@ func dbPoolSQLite(ctx context.Context, tb testing.TB, schema, driverName string) // I write all this because this line is a little dangerous. Removing it // will probably still allow a basic test run to pass so it might seem okay, // but actually it opens the door to intermittency hell. - databaseURLBuilder.WriteString("?_pragma=journal_mode(WAL)") + if driverName != "turso" { + databaseURLBuilder.WriteString("?_pragma=journal_mode(WAL)") + } dbPool, err := sql.Open(driverName, databaseURLBuilder.String()) require.NoError(tb, err)