Migrations

River needs a small set of tables in the database to operate, and provides a command line tool which executes migrations.

River persists jobs to a Postgres database, and needs a small set of tables created to insert jobs and carry out leader election. It's bundled with a command line tool which executes migrations, and which future-proofs River in case other migration steps need to be run in future versions.

Install the River CLI:

go install github.com/riverqueue/river/cmd/river@latest

With the DATABASE_URL of a target database (looks like postgres://host:5432/db), migrate up:

river migrate-up --database-url "$DATABASE_URL"

See also using an alternate Postgres schema.

River tables can be removed through an equivalent down migration:

river migrate-down --database-url "$DATABASE_URL" --max-steps 10

This is a destructive action that'll remove River's job table along with all the jobs that were in it. river migrate-down defaults to one migration step at a time. --max-steps is set to a high number so all steps are removed.

For users that would like to use their own migration framework instead of the one built into River, the CLI also supports dumping the raw SQL so it can be imported elsewhere.

Print a single version using river migrate-get along with a --version parameter and either --up or --down:

river migrate-get --version 3 --up > river3.up.sql
river migrate-get --version 3 --down > river3.down.sql

The contents of river3.up.sql will be:

-- River migration 003 [up]
ALTER TABLE river_job ALTER COLUMN tags SET DEFAULT '{}';
UPDATE river_job SET tags = '{}' WHERE tags IS NULL;
ALTER TABLE river_job ALTER COLUMN tags SET NOT NULL;

When bootstrapping new projects, River's full set of migrations are available with --all. Version 1 contains the tables for River's internal migration framework, so a common pattern is to use --all, but exclude version 1 in both directions:

river migrate-get --all --exclude-version 1 --up > river_all.up.sql
river migrate-get --all --exclude-version 1 --down > river_all.down.sql

River provides a Go API to run migrations through the rivermigrate package, for those who prefer it over a CLI.

Use is similar to the River client. Instantiate a migrator using a database driver like riverpgxv5:

migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)

Then migrate up:

res, err := migrator.MigrateTx(ctx, tx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
    TargetVersion: <target_version>,
})
if err != nil {
    // handle error
}

See the Migrate example for complete code.

The migrate function has both non-transactional (Migrate) and transactional (MigrateTx) variants, and can be used in a variety of ways:

• With no or empty MigrateOpts, fully migrates to the latest River schema version.
• MaxSteps specifies the maximum number of steps to apply.
• TargetVersion targets a specific schema version. This is the recommended use when pairing River's migration API with another migration framework like Goose. Find the current River migration by looking for the higest number in its migrations directory, and target that. As new River migrations are released in future versions, add new migrations that target them.

Migrate down:

res, err = migrator.MigrateTx(ctx, tx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
    MaxSteps: 1,
})

Unlike migrating up, migrating down applies only one step by default (when MigrateOpts is nil or empty). The special value TargetVersion: -1 will remove all River schema additions and delete any data from its tables.

River provides a riverdatabasesql driver that lets it run with a sql.Tx from the standard library for use with tools like the Goose migration framework.

Goose Go migrations involve building a custom binary (see example), then creating migration files that are compiled into it. Add contents similar to the below to a file like 00001_raise_river.go:

package main

import (
    "context"
    "database/sql"

    "github.com/pressly/goose/v3"
    "github.com/riverqueue/river/riverdriver/riverdatabasesql"
    "github.com/riverqueue/river/rivermigrate"
)

var migrator = rivermigrate.New(riverdatabasesql.New(nil), nil)

func init() {
    goose.AddMigration(Up, Down)
}

func Up(tx *sql.Tx) error {
    // Migrate up. An empty MigrateOpts will migrate all the way up, but
    // best practice is to specify a specific target version.
    _, err := migrator.MigrateTx(context.Background(), tx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
        TargetVersion: <target_version>,
    })
    return err
}

func Down(tx *sql.Tx) error {
    // TargetVersion -1 removes River's schema completely.
    _, err := migrator.MigrateTx(context.Background(), tx, rivermigrate.DirectionDown, &rivermigrate.MigrateOpts{
        TargetVersion: -1,
    })
    return err
}

Best practice is to write migrations so they target River's latest version. Get it by looking for the biggest number in River's migrations directory.

The main.go file for the custom Goose binary will connect to Postgres with a driver like pgx (again, see the full example):

import _ "github.com/jackc/pgx/v5/stdlib"

...

db, err := goose.OpenDBWithDriver("pgx", dbstring)
if err != nil {
    // handle error
}

Then built and run with:

$ go build -o goose-custom *.go
$ ./goose-custom postgres "$DATABASE_URL" up

Here's what each of River's jobs is used for:

• river_job: The main jobs table where all the work happens. Jobs are inserted as new rows and clients read out of it in bulk as they lock jobs for work.

• river_leader: An unlogged table used to elect a leader that will run queue maintenance services.

• river_migration: Stores which River migrations have been applied.