River

https://riverqueue.com/docs
Admin
River is a fast and reliable job queue for Go that processes asynchronous jobs from a PostgreSQL...
Tokens:76,872
Snippets:858
Trust Score:7.8
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install River UI from Source

Source: https://riverqueue.com/docs/river-ui

Installs the open-source River UI command-line tool using Go modules. Run the 'riverui' command after installation.

```bash
go install riverqueue.com/riverui/cmd/riverui@latest
riverui
```

--------------------------------

### Install River Pro UI from Source

Source: https://riverqueue.com/docs/river-ui

Installs the River Pro UI command-line tool using Go modules. Run the 'riverproui' command after installation.

```bash
go install riverqueue.com/riverui/riverproui/cmd/riverproui@latest
priverproui
```

--------------------------------

### Install River Pro CLI

Source: https://riverqueue.com/docs/pro/migrations

Installs the River Pro CLI using `go install`. Ensure your environment is configured for Pro, including `GOPROXY` and `GONOSUMDB`.

```bash
go install riverqueue.com/riverpro/cmd/riverpro@latest
```

--------------------------------

### Install River CLI and Run Migrations

Source: https://riverqueue.com/docs/river-ui

Installs the River CLI tool and applies database migrations. Ensure the DATABASE_URL environment variable is set.

```bash
go install github.com/riverqueue/river@latest
river migrate-up --database-url "$DATABASE_URL"
```

--------------------------------

### Install River CLI

Source: https://riverqueue.com/docs

Install the River command line tool to manage database migrations.

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

--------------------------------

### GitHub Actions Setup for River Pro

Source: https://riverqueue.com/docs/pro/go-proxy

Configure GitHub Actions to install River Pro CLI and set necessary environment variables for Go builds. This includes setting GOPROXY and GONOSUMDB using repository secrets.

```yaml
- name: Setup Go
  uses: actions/setup-go@v5
  with:
    go-version: "stable"
    check-latest: true
```

```yaml
- name: Install River Pro CLI
  run: go install riverqueue.com/riverpro/cmd/riverpro@latest
  env:
    GOPROXY: https://proxy.golang.org,https://river:${{ secrets.RIVER_PRO_SECRET }}@riverqueue.com/goproxy,direct
    GONOSUMDB: riverqueue.com/riverpro
```

```yaml
- name: Migrate River
  run: |
    riverpro migrate-up --database-url "$DATABASE_URL"
    riverpro migrate-up --database-url "$DATABASE_URL" --line pro
```

```yaml
- name: Go test
  run: go test ./...
  env:
    GOPROXY: https://proxy.golang.org,https://river:${{ secrets.RIVER_PRO_SECRET }}@riverqueue.com/goproxy,direct
    GONOSUMDB: riverqueue.com/riverpro
```

--------------------------------

### Install Prisma driver

Source: https://riverqueue.com/docs/typescript

Install the necessary packages to use River with Prisma.

```bash
pnpm add riverqueue @riverqueue/driver-prisma
```

--------------------------------

### Install River packages

Source: https://riverqueue.com/docs

Install the core River package and the pgxv5 driver in your Go project directory.

```bash
go get github.com/riverqueue/river
go get github.com/riverqueue/river/riverdriver/riverpgxv5
```

--------------------------------

### Starting a River Client

Source: https://riverqueue.com/docs/inserting-and-working-jobs

Initialize a client with a database pool and worker configuration to begin processing jobs.

```go

riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 100},
    },
    Workers: workers,
})
if err != nil {
    panic(err)
}


// Run the client inline. All executed jobs will inherit from ctx:
if err := riverClient.Start(ctx); err != nil {
    panic(err)
}
```

--------------------------------

### Example output of river3.up.sql

Source: https://riverqueue.com/docs/migrations

The generated SQL file for migration version 3.

```sql
-- 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;
```

--------------------------------

### Benchmark Output Example

Source: https://riverqueue.com/docs/benchmarks

Example output from the benchmark utility showing job throughput over time.

```text
bench: jobs worked [          0 ], inserted [    1000000 ], job/sec [        0.0 ] [0s]
bench: jobs worked [      82657 ], inserted [          0 ], job/sec [    41328.5 ] [2s]
bench: jobs worked [      96057 ], inserted [          0 ], job/sec [    48028.5 ] [2s]
bench: jobs worked [      89829 ], inserted [          0 ], job/sec [    44914.5 ] [2s]
bench: jobs worked [      96847 ], inserted [          0 ], job/sec [    48423.5 ] [2s]
bench: jobs worked [      96042 ], inserted [          0 ], job/sec [    48021.0 ] [2s]
bench: jobs worked [      87198 ], inserted [          0 ], job/sec [    43599.0 ] [2s]
bench: jobs worked [      96474 ], inserted [          0 ], job/sec [    48237.0 ] [2s]
bench: jobs worked [      94126 ], inserted [          0 ], job/sec [    47063.0 ] [2s]
bench: jobs worked [      85323 ], inserted [          0 ], job/sec [    42661.5 ] [2s]
bench: jobs worked [      94043 ], inserted [          0 ], job/sec [    47021.5 ] [2s]
bench: jobs worked [      81387 ], inserted [          0 ], job/sec [    40693.5 ] [2s]
bench: total jobs worked [    1000000 ], total jobs inserted [    1000000 ], overall job/sec [    45753.1 ], running 21.856442959s
```

--------------------------------

### Install riverqueue dependencies

Source: https://riverqueue.com/docs/typescript

Install the core package along with the required driver for either node-postgres or Prisma.

```bash
# Using node-postgres (pg)
pnpm add riverqueue @riverqueue/driver-pg pg


# Using Prisma
pnpm add riverqueue @riverqueue/driver-prisma
```

--------------------------------

### Install pgtestdb

Source: https://riverqueue.com/docs/testing

Install the pgtestdb package to facilitate testing with real Postgres databases.

```bash
go get github.com/peterldowns/pgtestdb@latest
```

--------------------------------

### Install River CLI

Source: https://riverqueue.com/docs/benchmarks

Installs the River CLI tool. Ensure you have a recent version for the latest optimizations.

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

--------------------------------

### Install node-postgres driver

Source: https://riverqueue.com/docs/typescript

Install the necessary packages to use River with node-postgres.

```bash
pnpm add riverqueue @riverqueue/driver-pg pg
```

--------------------------------

### Install otelriver dependency

Source: https://riverqueue.com/docs/metrics

Command to install the OpenTelemetry integration for River.

```bash
go get -u github.com/riverqueue/rivercontrib/otelriver
```

--------------------------------

### Install riverlog.Middleware as a plugin

Source: https://riverqueue.com/docs/job-logging

Configure the River client with the riverlog middleware, providing a function that initializes a slog handler with the provided io.Writer.

```go
riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{
        riverlog.NewMiddleware(func(w io.Writer) slog.Handler {
            return slog.NewJSONHandler(w, nil)
        }, nil),
    },
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Configuring River Client with ErrorHandler

Source: https://riverqueue.com/docs/error-handling

Example of how to provide a custom ErrorHandler implementation when initializing a River client.

```go
riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    ErrorHandler: &CustomErrorHandler{},
    ...
})
```

--------------------------------

### Install robfig/cron Package

Source: https://riverqueue.com/docs/periodic-jobs

Install the `robfig/cron` package to use complex cron schedules.

```bash
go get github.com/robfig/cron/v3
```

--------------------------------

### Create and Start River UI Handler

Source: https://riverqueue.com/docs/river-ui

Initializes River UI endpoints and creates an http.Handler. Starts the handler to initialize background processes and mounts it to an HTTP mux.

```go
endpoints := riverui.NewEndpoints(riverClient, nil)
// or, for Pro customers:
endpoints = riverproui.NewEndpoints(riverProClient, nil)


opts := &riverui.HandlerOpts{
    Endpoints: endpoints,
    Logger: slogLogger,
    Prefix: "/riverui", // mount the UI and its APIs under /riverui or another path
    // ...
}
handler, err := riverui.NewHandler(opts)
if err != nil {
    log.Fatal(err)
}
// Start the handler to initialize background processes for caching and periodic queries:
handler.Start(ctx)


mux := http.NewServeMux()
mux.Handle("/riverui/", handler)


// ... start and run your HTTP server

```

--------------------------------

### Implement Batching for a Job Kind

Source: https://riverqueue.com/docs/pro/batching

Implement `BatchOpts()` on your job args to enable batching and configure how batches are formed. This example shows the basic structure for enabling batching and defining `Work` and `WorkMany` methods.

```go
type MyBatchArgs struct{}


func (MyBatchArgs) Kind() string { return "my_batch" }


// Enable batching for this job kind. Customize options as needed.
func (MyBatchArgs) BatchOpts() riverpro.BatchOpts { return riverpro.BatchOpts{} }

type MyWorker struct {
    river.WorkerDefaults[MyBatchArgs]
}


func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyBatchArgs]) error {
    // Invoke the batch helper so this job can gather a batch and run WorkMany.
    return riverbatch.Work[MyBatchArgs, pgx.Tx](ctx, job, w, nil)
}


func (w *MyWorker) WorkMany(ctx context.Context, jobs []*river.Job[MyBatchArgs]) error {
    // Process the entire batch at once.
    //
    // Return nil to mark the entire batch as successful, a regular error to
    // fail the entire batch with the same error, or a MultiError to return
    // individualized errors per job.
    return nil
}
```

--------------------------------

### Get River Client from Context (database/sql)

Source: https://riverqueue.com/docs/context-client

When using the `database/sql` driver, the type parameter for `ClientFromContext` should be `*sql.Tx`.

```go
client := river.ClientFromContext[*sql.Tx](ctx)
```

--------------------------------

### Configure River Pro client with encryption

Source: https://riverqueue.com/docs/pro/changelog

Example of initializing a River Pro client with an encryption hook using riversecretbox.

```go
var key [32]byte
rand.Reader.Read(key[:])


riverClient, err := riverpro.NewClient(riverpropgxv5.New(dbPool), &riverpro.Config{
    Config: river.Config{
        Hooks: []rivertype.Hook{
            riverencrypt.NewEncryptHook(riversecretbox.NewEncryptor(key)),
        },
    },
})
```

--------------------------------

### Install plugins in River Config

Source: https://riverqueue.com/docs/plugins

Register plugins globally by adding them to the Plugins slice in the river.Config. The order of plugins in the list determines the execution order of hooks and middleware.

```go
riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    // Order is significant.
    Plugins: []rivertype.Plugin{
        &logPlugin{},
        &timingPlugin{},
    },
})
```

--------------------------------

### Get River Pro Client from Context

Source: https://riverqueue.com/docs/context-client

For River Pro customers, the `riverpro.Client` can be accessed from the context using `riverpro.ClientFromContext`.

```go
riverproClient := riverpro.ClientFromContext[pgx.Tx](ctx)
```

--------------------------------

### Define Initial Periodic Job Schedule

Source: https://riverqueue.com/docs/pro/durable-periodic-jobs

Sets up a periodic job with an initial schedule and a unique ID. This is the starting point before any modifications.

```go
PeriodicJobs: []*river.PeriodicJob{
    river.NewPeriodicJob(
        river.PeriodicInterval(365*24*time.Hour),
        ...
        &river.PeriodicJobOpts{
            ID: "my_periodic_job",
        },
    ),
},

```

--------------------------------

### Configure Unique Job Properties

Source: https://riverqueue.com/docs/unique-jobs

Implement `JobArgsWithInsertOpts` and populate `UniqueOpts` to define uniqueness constraints for a job type. This example enforces uniqueness by arguments and a 24-hour period.

```go
type ReconcileAccountArgs struct {
    AccountID int `json:"account_id"`
}


func (ReconcileAccountArgs) Kind() string { return "reconcile_account" }


// InsertOpts returns custom insert options that every job of this type will
// inherit, including unique options.
func (ReconcileAccountArgs) InsertOpts() river.InsertOpts {
    return river.InsertOpts{
        UniqueOpts: river.UniqueOpts{
            ByArgs:   true,
            ByPeriod: 24 * time.Hour,
        },
    }
}


...
```

--------------------------------

### Partition Concurrency Limits by Job Arguments

Source: https://riverqueue.com/docs/pro/concurrency-limits

Partition concurrency limits by job arguments to enforce limits independently for different subsets of jobs. This example partitions globally by 'customer_id' with a limit of 2 jobs per customer.

```go
&riverpro.Config{
    ProQueues: map[string]riverpro.QueueConfig{
        "expensive_actions": {
            Concurrency: riverpro.ConcurrencyConfig{
                GlobalLimit: 2,
                Partition:   riverpro.PartitionConfig{
                    ByArgs: []string{"customer_id"},
                },
            },
            MaxWorkers: 10,
        },
    },
}
```

--------------------------------

### Fetch River Pro modules using .netrc

Source: https://riverqueue.com/docs/pro/go-proxy

Fetch River Pro modules after configuring credentials in ~/.netrc. This example uses GOPROXY without an inline secret.

```shell
export GOPROXY=https://proxy.golang.org,https://riverqueue.com/goproxy,direct
export GONOSUMDB=riverqueue.com/riverpro,$GONOSUMDB

go get riverqueue.com/riverpro
go get riverqueue.com/riverpro/driver/riverpropgxv5
```

--------------------------------

### Configure Go environment variables with go env

Source: https://riverqueue.com/docs/pro/go-proxy

Use `go env -w` to persistently configure GOPROXY and GONOSUMDB. This example shows configuration with and without an inline secret.

```shell
go env -w GOPROXY=https://proxy.golang.org,https://riverqueue.com/goproxy,direct
go env -w GONOSUMDB=riverqueue.com/riverpro
```

```shell
# or, with a secret inline instead of being read from `~/.netrc`
go env -w GOPROXY=https://proxy.golang.org,https://river:river_secret_...@riverqueue.com/goproxy,direct
```

```shell

go get riverqueue.com/riverpro
go get riverqueue.com/riverpro/driver/riverpropgxv5
```

--------------------------------

### Migrate River Pro database using CLI

Source: https://riverqueue.com/docs/pro/changelog

Install the latest version of the River Pro CLI and run the migration to update the database schema to version 4.

```bash
go install riverqueue.com/riverpro/cmd/riverpro@latest
riverpro migrate-up --database-url "$DATABASE_URL" --line pro
```

--------------------------------

### Enable encryption with EncryptHook

Source: https://riverqueue.com/docs/pro/encrypted-jobs

Install riverencrypt.EncryptHook on a River client to enable job argument encryption. This example demonstrates decoding a base64 key and configuring the client.

```go
import (
    "encoding/base64"


    "riverqueue.com/riverpro/riverencrypt"
    "riverqueue.com/riverpro/riverencrypt/riversecretbox"
)


decodedBytes, err := base64.StdEncoding.DecodeString(
    "iRmwTuVGl2BAwTUPRTJbP/iA2EKpTrzXpEcNIXG2BI0=",
)
if err != nil {
    panic(err)
}


var key [32]byte
if copy(key[:], decodedBytes) != 32 {
    panic("expected to copy exactly 32 bytes")
}


riverClient, err := riverpro.NewClient(riverpropgxv5.New(dbPool), &riverpro.Config{
    Config: river.Config{
        Plugins: []rivertype.Plugin{
            riverencrypt.NewEncryptHook(riversecretbox.NewEncryptor(key)),
        },
    },
})
```

--------------------------------

### Initialize River Client with libSQL

Source: https://riverqueue.com/docs/sqlite

Set up the database connection and River client using the libsql driver. Ensure the database pool is configured for single connection to manage concurrency.

```go
dbPool, err := sql.Open("libsql", "file:./river.libsql")
if err != nil {
    panic(err)
}
deferr dbPool.Close()


dbPool.SetMaxOpenConns(1)

workers := river.NewWorkers()
river.AddWorker(workers, &SortWorker{})


riverClient, err := river.NewClient(riversqlite.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 100},
    },
    Workers:  workers,
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Initialize Sequel client with Postgres

Source: https://riverqueue.com/docs/ruby

Connect to the database and initialize the River client using the Sequel driver for Postgres.

```ruby
db = Sequel.connect("postgres://localhost/my_app")
client = River::Client.new(River::Driver::Sequel.new(db))
```

--------------------------------

### Initialize River client with riverpgxv5

Source: https://riverqueue.com/docs/database-drivers

Initialize a River client using the recommended riverpgxv5 driver, which wraps a pgxpool.

```go
import "github.com/riverqueue/river"
import "github.com/riverqueue/river/riverdriver/riverpgxv5"


...


dbPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
    panic(err)
}
defer dbPool.Close()


riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    ...
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Initialize River Pro Client

Source: https://riverqueue.com/docs/pro/getting-started

Initialize a River Pro client using `riverpro.NewClient` with the `riverpropgxv5` driver and a database connection pool. Configure queues and workers as needed.

```go
import (
    ...

    "github.com/riverqueue/river"
    "river.com/riverpro"
    "river.com/riverpro/driver/riverpropgxv5"
)


riverClient, err := riverpro.NewClient(riverpropgxv5.New(dbPool), &riverpro.Config{
    Config: river.Config{
        Queues: map[string]river.QueueConfig{
            river.QueueDefault: {MaxWorkers: 100},
        },
        Workers: workers,
    }
})
if err != nil {
    // handle error
}


if err := riverClient.Start(ctx); err != nil {
    // handle error
}
```

--------------------------------

### Initialize River client with riverdatabasesql

Source: https://riverqueue.com/docs/database-drivers

Create a River client using the riverdatabasesql driver to adapt a standard *sql.DB pool.

```go
riverClient, err := river.NewClient(
    riverdatabasesql.New(sqlDB),
    &river.Config{
        Workers: workers,
    },
)
if err != nil {
    panic(err)
}
```

--------------------------------

### Configure Goose main.go binary

Source: https://riverqueue.com/docs/migrations

Initializes the database connection using the pgx driver for the custom Goose migration binary.

```go

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


...


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

```

--------------------------------

### Run database migrations

Source: https://riverqueue.com/docs

Execute the migration command using your DATABASE_URL to set up the required database tables.

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

--------------------------------

### Stuck worker example

Source: https://riverqueue.com/docs/stuck-jobs

An example of a worker that can become stuck because it waits on a channel receive without observing the job context.

```go
func (w *ReportWorker) Work(ctx context.Context, job *river.Job[ReportArgs]) error {
    result := <-w.resultChan
    return saveReport(ctx, result)
}
```

--------------------------------

### Define and Execute a Workflow

Source: https://riverqueue.com/docs/pro/workflows

Demonstrates creating a workflow with dependencies, adding tasks, and inserting them into the queue using the River Pro client.

```go
import (
    "riverqueue.com/riverpro"
)


// MyJobArgs is a sample River JobArgs struct
type MyJobArgs struct {
    // ...
}


func (MyJobArgs) Kind() string { return "my_job" }


func SampleWorkflow(client *riverpro.Client[pgx.Tx]) *riverpro.Workflow[pgx.Tx] {
    // Create a new workflow:
    workflow := client.NewWorkflow(&riverpro.WorkflowOpts{Name: "My first workflow"})


    // Add a first task to the workflow, named "a":
    taskA := workflow.Add("a", MyJobArgs{}, nil, nil)


    // Fan-out to tasks b1 and b2, which both depend on task a:
    taskB1 := workflow.Add("b1", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskA.Name}})
    taskB2 := workflow.Add("b2", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskA.Name}})


    // Fan-in to task c, which depends on both b1 and b2:
    taskC := workflow.Add("c", MyJobArgs{}, nil, &riverpro.WorkflowTaskOpts{Deps: []string{taskB1.Name, taskB2.Name}})


    var _ = taskC // avoids "declared and not used" error


    return workflow
}


func main() {
    ctx := context.Background()
    riverClient, err := riverpro.NewClient(riverpropgxv5.New(dbPool), &riverpro.Config{
        Config: river.Config{
            Queues: map[string]river.QueueConfig{
                river.QueueDefault: {MaxWorkers: 100},
            },
            Workers: workers,
        },
    })


    // Prepare the workflow for insertion and validate it:
    workflow := SampleWorkflow(riverClient)
    result, err := workflow.Prepare(ctx)
    if err != nil {
        panic(err)
    }


    // The result.Jobs field holds a slice of river.InsertManyParams. Enqueue
    // them with InsertMany or InsertManyTx:
    if _, err := riverClient.InsertMany(ctx, result.Jobs); err != nil {
        panic(err)
    }


    // continue execution, stop client, etc...
}
```

--------------------------------

### Migrate Up

Source: https://riverqueue.com/docs/migrations

Apply migrations to the target database using the DATABASE_URL environment variable.

```bash
river migrate-up --line main --database-url "$DATABASE_URL"
```

--------------------------------

### Create an Insert-Only River Client

Source: https://riverqueue.com/docs/insert-only-clients

Initialize a River client for inserting jobs. Ensure the client is not started with `Start` and that `Queues` and `Workers` fields are omitted from its config.

```go
insertOnlyClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{})
```

--------------------------------

### Initialize Sequel client with SQLite

Source: https://riverqueue.com/docs/ruby

Connect to the database and initialize the River client using the Sequel driver for SQLite.

```ruby
db = Sequel.connect("sqlite://storage/river.sqlite3", timeout: 5_000)
client = River::Client.new(River::Driver::Sequel.new(db))
```

--------------------------------

### Initialize SQLite Client with River

Source: https://riverqueue.com/docs/sqlite

Activate the SQLite driver by initializing a client with `riversqlite` instead of the default `riverpgxv5`. Ensure `modernc.org/sqlite` is imported.

```go
import (
    "database/sql"


    _ "modernc.org/sqlite"


    "github.com/riverqueue/river"
    "github.com/riverqueue/river/riverdriver/riversqlite"
)
```

```go
dbPool, err := sql.Open("sqlite", "file:./river.sqlite3")
if err != nil {
    panic(err)
}
deferr dbPool.Close()


dbPool.SetMaxOpenConns(1)


workers := river.NewWorkers()
river.AddWorker(workers, &SortWorker{})


riverClient, err := river.NewClient(riversqlite.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 100},
    },
    Workers:  workers,
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Add Periodic Job After Client Start

Source: https://riverqueue.com/docs/periodic-jobs

Dynamically add a periodic job using `riverClient.PeriodicJobs().Add()` after the client has started. This operation only has an effect if the client is the elected leader.

```go
riverClient, err := river.NewClient(...)
if err != nil {
    panic(err)
}


if err := riverClient.Start(ctx); err != nil {
    panic(err)
}


// Add a periodic job after client has already started.
periodicJobHandle := riverClient.PeriodicJobs().Add(
    river.NewPeriodicJob(
        river.PeriodicInterval(15*time.Minute),
        func() (river.JobArgs, *river.InsertOpts) {
            return MyPeriodicJobArgs{}, nil
        },
        nil,
    ),
)
```

--------------------------------

### Safely Get River Client from Context (pgx.Tx)

Source: https://riverqueue.com/docs/context-client

Use `ClientFromContextSafely` to get the River client without panicking if called on an invalid context. This version returns an error if the client cannot be retrieved.

```go
func (w *MyWorker) Work(ctx context.Context, job *river.Job[MyArgs]) error {
    client, err := river.ClientFromContextSafely[pgx.Tx](ctx)
    if err != nil {
        return fmt.Errorf("error getting client from context: %w", err)
    }


    ...
}
```

--------------------------------

### Example encoded key output

Source: https://riverqueue.com/docs/pro/encrypted-jobs

The output format of the generated base64 encoded key.

```text
encoded key: iRmwTuVGl2BAwTUPRTJbP/iA2EKpTrzXpEcNIXG2BI0=
```

--------------------------------

### Example failure output for RequireInsertedTx

Source: https://riverqueue.com/docs/testing

This shows the error message displayed when a job assertion fails.

```text
--- FAIL: TestRequireInsertedTx (0.00s)
    --- FAIL: TestRequireInsertedTx/FailsWithoutInsert (0.12s)
        rivertest.go:352:
                River assertion failure:
                No jobs found with kind: required
```

--------------------------------

### Custom Error Handler Implementation

Source: https://riverqueue.com/docs/error-handling

Example implementation of the ErrorHandler interface for custom error and panic handling.

```APIDOC
## Custom Error Handler Implementation

### Description

This example shows a basic implementation of the `ErrorHandler` interface. It prints messages to the console when a job errors or panics.

### Methods

- **HandleError(ctx context.Context, job *rivertype.JobRow, err error) *river.ErrorHandlerResult**
  - Prints the error message to standard output.
  - Returns `nil`, allowing the job to follow the retry schedule.

- **HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult**
  - Prints the panic value and stack trace to standard output.
  - Returns `nil`, allowing the job to follow the retry schedule.

### Code Example

```go
type CustomErrorHandler struct{}

func (*CustomErrorHandler) HandleError(ctx context.Context, job *rivertype.JobRow, err error) *river.ErrorHandlerResult {
    fmt.Printf("Job errored with: %s\n", err)
    return nil
}

func (*CustomErrorHandler) HandlePanic(ctx context.Context, job *rivertype.JobRow, panicVal any, trace string) *river.ErrorHandlerResult {
    fmt.Printf("Job panicked with: %v\n", panicVal)
    fmt.Printf("Stack trace: %s\n", trace)
    return nil
}
```
```

--------------------------------

### Initialize River client with ActiveRecord

Source: https://riverqueue.com/docs/ruby

Require the necessary gems, establish an ActiveRecord database connection, and initialize the River client using the ActiveRecord driver.

```ruby
require "riverqueue"
require "riverqueue-activerecord"


ActiveRecord::Base.establish_connection("postgres://...")
client = River::Client.new(River::Driver::ActiveRecord.new)
```

--------------------------------

### Run Pro Migrations from Go

Source: https://riverqueue.com/docs/pro/migrations

Initializes and runs Pro migrations from Go code using the `riverpropgxv5` driver and specifying the `pro` line in the configuration. Leaving the `Line` property empty defaults to the main River migration line.

```go
migrator := rivermigrate.New(riverpropgxv5.New(dbPool), &rivermigrate.Config{
    Line: "pro",
})


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

--------------------------------

### Cancellable worker example

Source: https://riverqueue.com/docs/stuck-jobs

A corrected version of the worker that uses a select block to listen for both the channel result and context cancellation.

```go
func (w *ReportWorker) Work(ctx context.Context, job *river.Job[ReportArgs]) error {
    select {
    case result := <-w.resultChan:
        return saveReport(ctx, result)
    case <-ctx.Done():
        return ctx.Err()
    }
}
```

--------------------------------

### Migrate Up with Explicit PG Environment Variables

Source: https://riverqueue.com/docs/migrations

Run migrations by explicitly setting host, port, and database environment variables.

```bash
PGHOST=localhost PGPORT=5432 PGDATABASE=river_dev river migrate-up --line main
```

--------------------------------

### Sharing a transaction with River

Source: https://riverqueue.com/docs/bun

Start a transaction using Bun and pass the underlying *sql.Tx to River's InsertTx method.

```go
tx, err := bunDB.BeginTx(ctx, &sql.TxOptions{})
if err != nil {
    return nil, err
}


_, err = riverClient.InsertTx(ctx, tx.Tx, SortArgs{ // tx.Tx is *sql.Tx
    Strings: []string{
        "whale", "tiger", "bear",
    },
}, nil)
if err != nil {
    return nil, err
}


if err := tx.Commit(); err != nil {
    return nil, err
}
```

--------------------------------

### Initialize River client

Source: https://riverqueue.com/docs/python

Initialize the River client using a SQLAlchemy engine and the riversqlalchemy driver.

```python
import riverqueue
from riverqueue.driver import riversqlalchemy


engine = sqlalchemy.create_engine("postgresql://...")
client = riverqueue.Client(riversqlalchemy.Driver(engine))
```

--------------------------------

### Instantiating the Go migrator

Source: https://riverqueue.com/docs/migrations

Initialize the migrator using the riverpgxv5 driver and a database pool.

```go
migrator, err := rivermigrate.New(riverpgxv5.New(dbPool), nil)
if err != nil {
    panic(err)
}
```

--------------------------------

### Implement SequenceOpts for Basic Sequencing

Source: https://riverqueue.com/docs/pro/sequences

Implement the SequenceOpts() interface on your JobArgs struct to enable basic sequencing. By default, all jobs of the same kind run in a single sequence.

```go
type MyJobArgs struct {
    CustomerID string `json:"customer_id"`
}


func (MyJobArgs) Kind() string { return "my_job" }


func (MyJobArgs) SequenceOpts() riverpro.SequenceOpts {
    // Use the default sequence partitioning based solely on the job kind.
    return riverpro.SequenceOpts{}
}
```

--------------------------------

### Run River Pro Migrations

Source: https://riverqueue.com/docs/pro/getting-started

Execute database migrations using the `riverpro` CLI. First, apply standard River migrations, then apply Pro-specific migration lines.

```bash
# first, install the standard River migrations:
riverpro migrate-up --database-url "$DATABASE_URL"
```

```bash
# then add the pro migration line:
riverpro migrate-up --database-url "$DATABASE_URL" --line pro
```

--------------------------------

### Define Wait with Signal and Timer Terms

Source: https://riverqueue.com/docs/pro/workflows

Configures a wait specification using a signal term and a relative timer that starts after the wait becomes active.

```go
Wait: &riverworkflow.WaitSpec{
    Expr: "risk_hold_received || hold_timeout",
    Terms: []riverworkflow.WaitTermSpec{
        riverworkflow.WaitTermSignal("risk_hold_received", "risk_hold", `payload.active == true`),
        riverworkflow.WaitTermTimer(riverworkflow.TimerAfterWaitStarted("hold_timeout", 15*time.Minute)),
    },
},
```

--------------------------------

### Cancel start context for graceful shutdown

Source: https://riverqueue.com/docs/graceful-shutdown

Cancel the context passed to Client.Start to initiate a soft stop. Wait on Client.Stopped to ensure the client has fully finished.

```go
if err := riverClient.Start(ctx); err != nil {
    panic(err)
}


// ... later, cancel ctx to initiate stop ...


// wait for all jobs to complete
<-riverClient.Stopped()
```

--------------------------------

### Configure original encryption key

Source: https://riverqueue.com/docs/pro/encrypted-jobs

Step 0: Initialize the River Pro client with the original encryption key.

```go
keyOld := mustDecodeBase64EncodedKey("fdnQ7+v/5Pb28rYqpynRSdzWfqs1gD6/J/0I9IUh65s=")


_, err := riverpro.NewClient(riverpropgxv5.New(nil), &riverpro.Config{
    Config: river.Config{
        Plugins: []rivertype.Plugin{
            riverencrypt.NewEncryptHook(riversecretbox.NewEncryptor(
                keyOld,
            )),
        },
    },
})
```

--------------------------------

### Migrate Up with PGDATABASE

Source: https://riverqueue.com/docs/migrations

Run migrations using libpq environment variables instead of a database URL.

```bash
PGDATABASE=river_dev river migrate-up --line main
```

--------------------------------

### Enable Immediate Transactions in SQLite

Source: https://riverqueue.com/docs/sqlite

Configure the database connection to use `BEGIN IMMEDIATE` transactions by appending `?_txlock=immediate` to the database URL. This prevents SQLITE_BUSY errors during transaction upgrades.

```go
dbPool, err := sql.Open("sqlite",
    "file:./river.sqlite3?_txlock=immediate")
```

--------------------------------

### Fetch River Pro modules with secret

Source: https://riverqueue.com/docs/pro/go-proxy

Configure GOPROXY and GONOSUMDB to fetch River Pro modules, including setting a River Pro secret and installing the module and driver.

```shell
export RIVER_PRO_SECRET=river_secret_...
export GOPROXY=https://proxy.golang.org,https://river:$RIVER_PRO_SECRET@riverqueue.com/goproxy,direct
export GONOSUMDB=riverqueue.com/riverpro,$GONOSUMDB

go get riverqueue.com/riverpro
go get riverqueue.com/riverpro/driver/riverpropgxv5
```

```shell
# install riverpro CLI:
go install riverqueue.com/riverpro/cmd/riverpro@latest
```

--------------------------------

### Implement timingPlugin with WorkerMiddleware

Source: https://riverqueue.com/docs/plugins

An example of a middleware plugin that measures job duration using the Work method. Middleware must call doInner to ensure the underlying operation proceeds.

```go
type timingPlugin struct {
    river.PluginDefaults
}


func (*timingPlugin) Work(
    ctx context.Context,
    job *rivertype.JobRow,
    doInner func(context.Context) error,
) error {
    start := time.Now()
    err := doInner(ctx)
    slog.InfoContext(ctx, "job worked",
        "kind", job.Kind,
        "duration", time.Since(start),
    )
    return err
}
```

--------------------------------

### Login to River Pro Docker Registry and Run Image

Source: https://riverqueue.com/docs/river-ui

Logs into the riverqueue.com Docker registry using a River Pro license key and then pulls and runs the Pro variant of the River UI Docker image.

```bash
# assuming RIVER_PRO_SECRET is set to a valid River Pro license key:
echo $RIVER_PRO_SECRET | docker login riverqueue.com -u river --password-stdin
docker pull riverqueue.com/riverproui:latest
docker run -p 8080:8080 --env DATABASE_URL riverqueue.com/riverproui:latest
```

--------------------------------

### Initialize in-memory River client

Source: https://riverqueue.com/docs/in-memory-queue

Configures a database connection, runs migrations, and initializes the River client. Note that SetMaxOpenConns(1) is required for SQLite concurrency constraints.

```go
dbPool, err := sql.Open("sqlite", ":memory:")
if err != nil {
    panic(err)
}
defer dbPool.Close()


dbPool.SetMaxOpenConns(1)


migrator, err := rivermigrate.New(riversqlite.New(dbPool), nil)
if err != nil {
    panic(err)
}


_, err = migrator.Migrate(ctx, rivermigrate.DirectionUp, nil)
if err != nil {
    panic(err)
}


workers := river.NewWorkers()
river.AddWorker(workers, &MyWorker{})


riverClient, err := river.NewClient(riversqlite.New(dbPool), &river.Config{
    Queues: map[string]river.QueueConfig{
        river.QueueDefault: {MaxWorkers: 100},
    },
    Workers: workers,
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Configure Custom Logging Middleware

Source: https://riverqueue.com/docs/job-logging

Use NewMiddlewareCustomContext to inject a custom logger into the job context. This example demonstrates storing a standard library logger, but can be adapted for Logrus or Zap.

```go
type customContextKey struct{}


riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{
    Plugins: []rivertype.Plugin{
        riverlog.NewMiddlewareCustomContext(func(ctx context.Context, w io.Writer) context.Context {
            // For demonstration purposes we show the use of a built-in
            // non-slog logger, but this could be anything like Logrus or
            // Zap. Even the raw writer could be stored if so desired.
            logger := log.New(w, "", 0)
            return context.WithValue(ctx, customContextKey{}, logger)
        }, nil),
    },
})
if err != nil {
    panic(err)
}
```

--------------------------------

### Custom Worker Retry Delay

Source: https://riverqueue.com/docs/job-retries

Override the default retry behavior for a specific worker type by implementing the `NextRetry` method. This example sets a constant 10-second delay for all retries.

```go
type ConstantRetryTimeWorker struct {
    river.WorkerDefaults[MyJobArgs]
}


func (w *ConstantRetryTimeWorker) Work(job *Job[MyJobArgs]) error {
    // ...
}


// NextRetry always schedules the next retry for 10 seconds from now.
func (w *ConstantRetryTimeWorker) NextRetry(job *Job[MyJobArgs]) time.Time {
    return time.Now().Add(10*time.Second)
}

```

--------------------------------

### Configure riverdatabasesql with a Pgx listener

Source: https://riverqueue.com/docs/database-drivers

Use NewWithPgxListener to enable LISTEN/NOTIFY support when using the riverdatabasesql driver.

```go
listenerPool, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
if err != nil {
    panic(err)
}
defer listenerPool.Close()


riverClient, err := river.NewClient(
    riverdatabasesql.NewWithPgxListener(sqlDB, listenerPool),
    &river.Config{
        Workers: workers,
    },
)
if err != nil {
    panic(err)
}
```
Tokens:
Raw