From 2bd03c54fde7c22037e456c878d6d612449011d8 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 | 1 + riverdbtest/riverdbtest.go | 6 ++ .../riverdrivertest/driver_client_test.go | 23 ++++++++ riverdriver/riverdrivertest/driver_test.go | 40 +++++++++++++ .../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 | 40 ++++++++----- .../riversharedtest/riversharedtest.go | 21 ++++++- 16 files changed, 211 insertions(+), 37 deletions(-) create mode 100644 riverdriver/riverdrivertest/example_turso_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 0fa5db07..540838e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Added `rivertype.HookMetricEmit` for receiving metrics emitted by River. Initial metrics report the duration of successful job fetches with `JobGetAvailableDurationMetric` and the number of jobs fetched with `JobGetAvailableCountMetric`. [PR #1285](https://github.com/riverqueue/river/pull/1285). +- 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 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..bbf9db89 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,45 @@ func TestDriverRiverSQLiteModernC(t *testing.T) { //nolint:dupl }) } +func TestDriverRiverTurso(t *testing.T) { + 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{ + 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 386957dc..56561800 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.54.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 9ac852ec..e946393b 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 35628049..94b433ca 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql @@ -487,7 +487,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( @@ -573,7 +573,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')) @@ -583,7 +583,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 *; @@ -601,7 +601,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')) @@ -609,7 +609,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 @@ -624,7 +624,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 1ffe44b5..5bd39b2c 100644 --- a/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go +++ b/riverdriver/riversqlite/internal/dbsqlc/river_job.sql.go @@ -1243,7 +1243,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( @@ -1524,7 +1524,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*/?) @@ -1591,7 +1591,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 @@ -1640,7 +1640,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')) @@ -1648,7 +1648,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 @@ -1723,7 +1723,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 1b833f2e..88ad262e 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,33 @@ 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 = ? )` + + var ( + 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,14 +464,14 @@ func (e *Executor) JobDeleteMany(ctx context.Context, params *riverdriver.JobDel // //nolint:gochecknoglobals var jobGetAvailableAttemptedBySQL = strings.TrimSpace(` - jsonb_insert( + jsonb(json_insert( ( SELECT jsonb_group_array(value) FROM ( SELECT * FROM ( SELECT * - FROM json_each(attempted_by) + FROM json_each(coalesce(attempted_by, jsonb('[]'))) ORDER BY key DESC LIMIT @max_attempted_by - 1 ) @@ -466,7 +480,7 @@ var jobGetAvailableAttemptedBySQL = strings.TrimSpace(` ), '$[#]', @attempted_by - ) + )) `) func (e *Executor) JobGetAvailable(ctx context.Context, params *riverdriver.JobGetAvailableParams) ([]*rivertype.JobRow, error) { @@ -1172,7 +1186,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..a5905ad1 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,12 @@ 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)") + // + // Turso ignores `_pragma` so we don't bother passing it. Turso has better + // defaults and WAL is enabled out of the box. + if driverName != "turso" { + databaseURLBuilder.WriteString("?_pragma=journal_mode(WAL)") + } dbPool, err := sql.Open(driverName, databaseURLBuilder.String()) require.NoError(tb, err)