internal/jobs persists jobs in the jobs table, which is also the queue: a
bounded pool of workers claims queued jobs oldest first. Handlers are
registered per kind before Start with typed payloads and results, and can
publish partial results to the row while they run. Only typed results and
*jobs.Failure messages are stored; any other handler error is logged and
stored as "internal error", so dependency error text cannot surface a
secret. Start fails jobs a previous process left running with
"interrupted by restart" before any worker runs.
GET /api/v1/jobs/{id} returns {id, kind, status, result, error} inside the
protected /api/v1 group; an unknown id is a 404 not_found.
Migration 0003_jobs_status rebuilds jobs with status queued, running,
succeeded, failed, mapping legacy pending and done rows.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
312 lines
9.8 KiB
Go
312 lines
9.8 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"io/fs"
|
|
"slices"
|
|
"strings"
|
|
"testing"
|
|
"testing/fstest"
|
|
"time"
|
|
|
|
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
|
|
)
|
|
|
|
var testMigrations = fstest.MapFS{
|
|
"0001_first.sql": {Data: []byte("CREATE TABLE first (id INTEGER PRIMARY KEY);")},
|
|
"0002_second.sql": {Data: []byte("CREATE TABLE second (id INTEGER PRIMARY KEY);\nCREATE TABLE third (id INTEGER PRIMARY KEY);")},
|
|
"README.md": {Data: []byte("not a migration")},
|
|
}
|
|
|
|
type migrationRow struct {
|
|
version string
|
|
appliedAt string
|
|
}
|
|
|
|
func migrationRows(t *testing.T, s *Store) []migrationRow {
|
|
t.Helper()
|
|
rows, err := s.Reader.Query("SELECT version, applied_at FROM schema_migrations ORDER BY version")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer rows.Close()
|
|
var out []migrationRow
|
|
for rows.Next() {
|
|
var r migrationRow
|
|
if err := rows.Scan(&r.version, &r.appliedAt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out = append(out, r)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func versions(rows []migrationRow) string {
|
|
vs := make([]string, len(rows))
|
|
for i, r := range rows {
|
|
vs[i] = r.version
|
|
}
|
|
return strings.Join(vs, ",")
|
|
}
|
|
|
|
func tableExists(t *testing.T, s *Store, name string) bool {
|
|
t.Helper()
|
|
var n int
|
|
if err := s.Reader.QueryRow("SELECT count(*) FROM sqlite_schema WHERE type = 'table' AND name = ?", name).Scan(&n); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return n == 1
|
|
}
|
|
|
|
func TestMigrateFreshDatabaseReachesLatest(t *testing.T) {
|
|
s := openTestStore(t)
|
|
if err := Migrate(context.Background(), s.Writer, testMigrations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
rows := migrationRows(t, s)
|
|
if got := versions(rows); got != "0001_first,0002_second" {
|
|
t.Fatalf("versions = %q", got)
|
|
}
|
|
for _, r := range rows {
|
|
if _, err := time.Parse(time.RFC3339, r.appliedAt); err != nil || !strings.HasSuffix(r.appliedAt, "Z") {
|
|
t.Errorf("applied_at %q is not UTC RFC 3339", r.appliedAt)
|
|
}
|
|
}
|
|
for _, table := range []string{"first", "second", "third"} {
|
|
if !tableExists(t, s, table) {
|
|
t.Errorf("table %s missing", table)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestMigrateSecondRunIsNoOp(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := openTestStore(t)
|
|
if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before := migrationRows(t, s)
|
|
if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
|
|
t.Fatalf("second run: %v", err)
|
|
}
|
|
after := migrationRows(t, s)
|
|
if len(after) != len(before) {
|
|
t.Fatalf("rows after second run = %v, want %v", after, before)
|
|
}
|
|
for i := range before {
|
|
if after[i] != before[i] {
|
|
t.Errorf("row %d changed: %v -> %v", i, before[i], after[i])
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestMigrateBrokenMigrationRollsBack(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := openTestStore(t)
|
|
if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
broken := fstest.MapFS{"0003_broken.sql": {Data: []byte(
|
|
// The first statement succeeds; the second is invalid, so the
|
|
// table created by the first must be rolled back with it.
|
|
"CREATE TABLE fourth (id INTEGER PRIMARY KEY);\nCREATE TABLE oops (;",
|
|
)}}
|
|
for name, f := range testMigrations {
|
|
broken[name] = f
|
|
}
|
|
err := Migrate(ctx, s.Writer, broken)
|
|
if err == nil {
|
|
t.Fatal("broken migration returned no error")
|
|
}
|
|
if !strings.Contains(err.Error(), "0003_broken.sql") {
|
|
t.Errorf("error %q does not name the failing migration", err)
|
|
}
|
|
if got := versions(migrationRows(t, s)); got != "0001_first,0002_second" {
|
|
t.Errorf("versions after failure = %q, want previous version intact", got)
|
|
}
|
|
if tableExists(t, s, "fourth") {
|
|
t.Error("table from the failed migration was not rolled back")
|
|
}
|
|
for _, table := range []string{"first", "second", "third"} {
|
|
if !tableExists(t, s, table) {
|
|
t.Errorf("previous table %s missing after failed migration", table)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestMigrateAppliesInFilenameOrder(t *testing.T) {
|
|
s := openTestStore(t)
|
|
// 0002 depends on 0001; MapFS iteration order is random, so this only
|
|
// passes if migrations are sorted by name.
|
|
ordered := fstest.MapFS{
|
|
"0002_child.sql": {Data: []byte("CREATE TABLE child (parent_id INTEGER REFERENCES parent(id)); INSERT INTO child VALUES (1);")},
|
|
"0001_parent.sql": {Data: []byte("CREATE TABLE parent (id INTEGER PRIMARY KEY); INSERT INTO parent VALUES (1);")},
|
|
}
|
|
if err := Migrate(context.Background(), s.Writer, ordered); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestMigrateUnknownAppliedVersionFails(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := openTestStore(t)
|
|
if err := Migrate(ctx, s.Writer, testMigrations); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
older := fstest.MapFS{"0001_first.sql": testMigrations["0001_first.sql"]}
|
|
if err := Migrate(ctx, s.Writer, older); err == nil {
|
|
t.Fatal("migrating a database newer than the build returned no error")
|
|
}
|
|
}
|
|
|
|
func TestMigrateNoMigrationsFails(t *testing.T) {
|
|
s := openTestStore(t)
|
|
if err := Migrate(context.Background(), s.Writer, fstest.MapFS{}); err == nil {
|
|
t.Fatal("empty migrations returned no error")
|
|
}
|
|
}
|
|
|
|
func TestMigrateEmbeddedMigrations(t *testing.T) {
|
|
s := openTestStore(t)
|
|
if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
|
|
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
|
|
}
|
|
}
|
|
|
|
func TestMigrateBootstrapThenEmbeddedInitApplies(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := openTestStore(t)
|
|
bootstrap, err := fs.ReadFile(migrations.FS, "0000_bootstrap.sql")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// A database migrated by the bootstrap-only build upgrades to the
|
|
// embedded set, which adds the later migrations alongside the unchanged
|
|
// bootstrap.
|
|
if err := Migrate(ctx, s.Writer, fstest.MapFS{"0000_bootstrap.sql": {Data: bootstrap}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
|
|
t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
|
|
}
|
|
if !tableExists(t, s, "dinners") {
|
|
t.Error("table from 0001_init was not created")
|
|
}
|
|
}
|
|
|
|
// schemaSQL returns every schema object's definition, so a test can prove a
|
|
// later Migrate created or altered nothing.
|
|
func schemaSQL(t *testing.T, s *Store) []string {
|
|
t.Helper()
|
|
rows, err := s.Reader.Query("SELECT type || ' ' || name || ': ' || coalesce(sql, '') FROM sqlite_schema ORDER BY type, name")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer rows.Close()
|
|
var out []string
|
|
for rows.Next() {
|
|
var def string
|
|
if err := rows.Scan(&def); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out = append(out, def)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out
|
|
}
|
|
|
|
func TestMigrateReopenSameDataDirIsNoOp(t *testing.T) {
|
|
ctx := context.Background()
|
|
dir := t.TempDir()
|
|
|
|
first, err := Open(ctx, dir)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := Migrate(ctx, first.Writer, migrations.FS); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before, beforeSchema := migrationRows(t, first), schemaSQL(t, first)
|
|
if err := first.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
second, err := Open(ctx, dir)
|
|
if err != nil {
|
|
t.Fatalf("reopen: %v", err)
|
|
}
|
|
t.Cleanup(func() {
|
|
if err := second.Close(); err != nil {
|
|
t.Error(err)
|
|
}
|
|
})
|
|
if err := Migrate(ctx, second.Writer, migrations.FS); err != nil {
|
|
t.Fatalf("migrate after reopen: %v", err)
|
|
}
|
|
if after := migrationRows(t, second); !slices.Equal(after, before) {
|
|
t.Errorf("schema_migrations changed on reopen:\n got %v\nwant %v", after, before)
|
|
}
|
|
if afterSchema := schemaSQL(t, second); !slices.Equal(afterSchema, beforeSchema) {
|
|
t.Errorf("schema changed on reopen:\n got %v\nwant %v", afterSchema, beforeSchema)
|
|
}
|
|
}
|
|
|
|
// TestMigrateJobsStatusMapsLegacyRows upgrades a database holding jobs in
|
|
// every 0001 status and checks 0003 carries each row over with its status
|
|
// mapped and every other column unchanged.
|
|
func TestMigrateJobsStatusMapsLegacyRows(t *testing.T) {
|
|
ctx := context.Background()
|
|
s := openTestStore(t)
|
|
before := fstest.MapFS{}
|
|
for _, name := range []string{"0000_bootstrap.sql", "0001_init.sql", "0002_household_settings_updated_by.sql"} {
|
|
body, err := fs.ReadFile(migrations.FS, name)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
before[name] = &fstest.MapFile{Data: body}
|
|
}
|
|
if err := Migrate(ctx, s.Writer, before); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
legacy := map[string]string{"pending": "queued", "running": "running", "done": "succeeded", "failed": "failed"}
|
|
for old := range legacy {
|
|
if _, err := s.Writer.Exec(`INSERT INTO jobs (id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at)
|
|
VALUES (?, 'suggest', ?, '{"n":1}', '{"r":2}', 'boom', 's', 'f', 'c', 'u')`, "job-"+old, old); err != nil {
|
|
t.Fatalf("insert %s job: %v", old, err)
|
|
}
|
|
}
|
|
|
|
if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for old, want := range legacy {
|
|
var kind, status, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt string
|
|
if err := s.Reader.QueryRow(`SELECT kind, status, payload, result, error, started_at, finished_at, created_at, updated_at FROM jobs WHERE id = ?`, "job-"+old).
|
|
Scan(&kind, &status, &payload, &result, &errText, &startedAt, &finishedAt, &createdAt, &updatedAt); err != nil {
|
|
t.Fatalf("read %s job: %v", old, err)
|
|
}
|
|
if status != want {
|
|
t.Errorf("%s job status = %q, want %q", old, status, want)
|
|
}
|
|
got := []string{kind, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt}
|
|
if wantCols := []string{"suggest", `{"n":1}`, `{"r":2}`, "boom", "s", "f", "c", "u"}; !slices.Equal(got, wantCols) {
|
|
t.Errorf("%s job columns = %v, want %v", old, got, wantCols)
|
|
}
|
|
}
|
|
if tableExists(t, s, "jobs_new") {
|
|
t.Error("jobs_new left behind")
|
|
}
|
|
}
|