diff --git a/CHANGELOG.md b/CHANGELOG.md index 09d691c2..f45ba769 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). + ### Fixed - Guard against empty job slice returned by `JobSetStateIfRunningMany` when a job has been deleted mid-run. [PR #1308](https://github.com/riverqueue/river/pull/1308). 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 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 f4cc1971..b5a08834 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) { @@ -1171,7 +1185,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)