river

Dec 22, 2025

Changes in this version

type Client

+ func (c *Client[TTx]) JobUpdate(ctx context.Context, id int64, params *JobUpdateParams) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobUpdateTx(ctx context.Context, tx TTx, id int64, params *JobUpdateParams) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) Notify() *ClientNotifyBundle[TTx]

+ type ClientNotifyBundle struct

+ func (c *ClientNotifyBundle[TTx]) RequestResign(ctx context.Context) error

+ func (c *ClientNotifyBundle[TTx]) RequestResignTx(ctx context.Context, tx TTx) error

+ type HookPeriodicJobsStartFunc func(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error

+ func (f HookPeriodicJobsStartFunc) IsHook() bool

+ func (f HookPeriodicJobsStartFunc) Start(ctx context.Context, params *rivertype.HookPeriodicJobsStartParams) error

+ type JobUpdateParams struct

+ Output any

Nov 17, 2025

Changes in this version

type PeriodicJobBundle

+ func (b *PeriodicJobBundle) RemoveByID(id string) bool

+ func (b *PeriodicJobBundle) RemoveManyByID(ids []string)

Sep 15, 2025

Changes in this version

type JobDeleteManyParams

+ func (p *JobDeleteManyParams) UnsafeAll() *JobDeleteManyParams

Aug 16, 2025

Changes in this version

+ func AddWorkerArgs[T JobArgs](workers *Workers, jobArgs T, worker Worker[T])

type Client

+ func (c *Client[TTx]) JobDeleteMany(ctx context.Context, params *JobDeleteManyParams) (*JobDeleteManyResult, error)

+ func (c *Client[TTx]) JobDeleteManyTx(ctx context.Context, tx TTx, params *JobDeleteManyParams) (*JobDeleteManyResult, error)

+ func (c *Client[TTx]) Schema() string

+ type JobDeleteManyParams struct

+ func NewJobDeleteManyParams() *JobDeleteManyParams

+ func (p *JobDeleteManyParams) First(count int) *JobDeleteManyParams

+ func (p *JobDeleteManyParams) IDs(ids ...int64) *JobDeleteManyParams

+ func (p *JobDeleteManyParams) Kinds(kinds ...string) *JobDeleteManyParams

+ func (p *JobDeleteManyParams) Priorities(priorities ...int16) *JobDeleteManyParams

+ func (p *JobDeleteManyParams) Queues(queues ...string) *JobDeleteManyParams

+ func (p *JobDeleteManyParams) States(states ...rivertype.JobState) *JobDeleteManyParams

+ type JobDeleteManyResult struct

+ Jobs []*rivertype.JobRow

type QueueConfig

+ FetchCooldown time.Duration

+ FetchPollInterval time.Duration

Jun 4, 2025

Changes in this version

type Config

+ ReindexerTimeout time.Duration

type JobListParams

+ func (p *JobListParams) Where(sql string, namedArgsMany ...NamedArgs) *JobListParams

+ type NamedArgs map[string]any

type PeriodicJobBundle

+ func (b *PeriodicJobBundle) AddManySafely(periodicJobs []*PeriodicJob) ([]rivertype.PeriodicJobHandle, error)

+ func (b *PeriodicJobBundle) AddSafely(periodicJob *PeriodicJob) (rivertype.PeriodicJobHandle, error)

type PeriodicJobOpts

+ ID string

+ type QueueAlreadyAddedError struct

+ Name string

+ func (e *QueueAlreadyAddedError) Error() string

+ func (e *QueueAlreadyAddedError) Is(target error) bool

May 10, 2025

Changes in this version

type Config

+ SkipJobKindValidation bool

+ type JobArgsWithKindAliases interface

+ KindAliases func() []string

May 3, 2025

Changes in this version

type Config

+ Schema string

+ type HookWorkEndFunc func(ctx context.Context, err error) error

+ func (f HookWorkEndFunc) IsHook() bool

+ func (f HookWorkEndFunc) WorkEnd(ctx context.Context, err error) error

+ type JobInsertMiddlewareFunc func(ctx context.Context, manyParams []*rivertype.JobInsertParams, ...) ([]*rivertype.JobInsertResult, error)

+ func (f JobInsertMiddlewareFunc) InsertMany(ctx context.Context, manyParams []*rivertype.JobInsertParams, ...) ([]*rivertype.JobInsertResult, error)

+ func (f JobInsertMiddlewareFunc) IsMiddleware() bool

type JobListParams

+ func (p *JobListParams) IDs(ids ...int64) *JobListParams

+ func (p *JobListParams) Priorities(priorities ...int16) *JobListParams

+ type WorkerMiddlewareFunc func(ctx context.Context, job *rivertype.JobRow, ...) error

+ func (f WorkerMiddlewareFunc) IsMiddleware() bool

+ func (f WorkerMiddlewareFunc) Work(ctx context.Context, job *rivertype.JobRow, ...) error

Apr 9, 2025

Changes in this version

type Client

+ func (c *Client[TTx]) QueueUpdateTx(ctx context.Context, tx TTx, name string, params *QueueUpdateParams) (*rivertype.Queue, error)

Apr 4, 2025

Changes in this version

type Client

+ func (c *Client[TTx]) QueueUpdate(ctx context.Context, name string, params *QueueUpdateParams) (*rivertype.Queue, error)

+ type QueueUpdateParams struct

+ Metadata []byte

Mar 17, 2025

Changes in this version

type Config

+ Hooks []rivertype.Hook

+ Middleware []rivertype.Middleware

+ type HookDefaults struct

+ func (d *HookDefaults) IsHook() bool

+ type HookInsertBeginFunc func(ctx context.Context, params *rivertype.JobInsertParams) error

+ func (f HookInsertBeginFunc) InsertBegin(ctx context.Context, params *rivertype.JobInsertParams) error

+ func (f HookInsertBeginFunc) IsHook() bool

+ type HookWorkBeginFunc func(ctx context.Context, job *rivertype.JobRow) error

+ func (f HookWorkBeginFunc) IsHook() bool

+ func (f HookWorkBeginFunc) WorkBegin(ctx context.Context, job *rivertype.JobRow) error

+ type JobArgsWithHooks interface

+ Hooks func() []rivertype.Hook

+ type MiddlewareDefaults struct

+ func (d *MiddlewareDefaults) IsMiddleware() bool

type WorkerDefaults

+ func (w WorkerDefaults[T]) Hooks(*rivertype.JobRow) []rivertype.Hook

Feb 20, 2025

Changes in this version

+ func RecordOutput(ctx context.Context, output any) error

type TestConfig

+ DisableUniqueEnforcement bool

Feb 15, 2025

Changes in this version

type Config

+ Test TestConfig

+ func (c *Config) WithDefaults() *Config

+ type TestConfig struct

+ Time rivertype.TimeGenerator

Jan 28, 2025

Changes in this version

type Config

+ SkipUnknownJobCheck bool

type PeriodicSchedule

+ func NeverSchedule() PeriodicSchedule

Nov 3, 2024

Changes in this version

+ type JobCancelError struct

+ func (e *JobCancelError) Error() string

+ func (e *JobCancelError) Is(target error) bool

+ func (e *JobCancelError) Unwrap() error

+ type JobSnoozeError struct

+ func (e *JobSnoozeError) Error() string

+ func (e *JobSnoozeError) Is(target error) bool

Oct 8, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) Pilot() riverpilot.Pilot

type Config

+ JobInsertMiddleware []rivertype.JobInsertMiddleware

+ WorkerMiddleware []rivertype.WorkerMiddleware

+ type JobInsertMiddlewareDefaults struct

+ func (d *JobInsertMiddlewareDefaults) InsertMany(ctx context.Context, manyParams []*rivertype.JobInsertParams, ...) ([]*rivertype.JobInsertResult, error)

type Worker

+ Middleware func(job *Job[T]) []rivertype.WorkerMiddleware

type WorkerDefaults

+ func (w WorkerDefaults[T]) Middleware(*Job[T]) []rivertype.WorkerMiddleware

+ type WorkerMiddlewareDefaults struct

+ func (d *WorkerMiddlewareDefaults) Work(ctx context.Context, job *rivertype.JobRow, ...) error

Sep 23, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) InsertManyFast(ctx context.Context, params []InsertManyParams) (int, error)

+ func (c *Client[TTx]) InsertManyFastTx(ctx context.Context, tx TTx, params []InsertManyParams) (int, error)

type Config

+ JobCleanerTimeout time.Duration

type UniqueOpts

+ ExcludeKind bool

Aug 3, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) Driver() riverdriver.Driver[TTx]

Jul 19, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) Queues() *QueueBundle

+ type QueueBundle struct

+ func (b *QueueBundle) Add(queueName string, queueConfig QueueConfig) error

Jul 4, 2024

Changes in this version

type Config

+ TestOnly bool

Jun 26, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) QueueGetTx(ctx context.Context, tx TTx, name string) (*rivertype.Queue, error)

+ func (c *Client[TTx]) QueueListTx(ctx context.Context, tx TTx, params *QueueListParams) (*QueueListResult, error)

+ func (c *Client[TTx]) QueuePauseTx(ctx context.Context, tx TTx, name string, opts *QueuePauseOpts) error

+ func (c *Client[TTx]) QueueResumeTx(ctx context.Context, tx TTx, name string, opts *QueuePauseOpts) error

Jun 14, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) JobDelete(ctx context.Context, id int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobDeleteTx(ctx context.Context, tx TTx, id int64) (*rivertype.JobRow, error)

type Config

+ MaxAttempts int

May 3, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) QueueGet(ctx context.Context, name string) (*rivertype.Queue, error)

+ func (c *Client[TTx]) QueueList(ctx context.Context, params *QueueListParams) (*QueueListResult, error)

+ func (c *Client[TTx]) QueuePause(ctx context.Context, name string, opts *QueuePauseOpts) error

+ func (c *Client[TTx]) QueueResume(ctx context.Context, name string, opts *QueuePauseOpts) error

type Event

+ Queue *rivertype.Queue

type EventKind

+ const EventKindQueuePaused

+ const EventKindQueueResumed

type InsertOpts

+ Pending bool

+ type QueueListParams struct

+ func NewQueueListParams() *QueueListParams

+ func (p *QueueListParams) First(count int) *QueueListParams

+ type QueueListResult struct

+ Queues []*rivertype.Queue

+ type QueuePauseOpts struct

Apr 21, 2024

Changes in this version

type JobListOrderByField

+ const JobListOrderByFinalizedAt

+ const JobListOrderByID

+ const JobListOrderByScheduledAt

type JobListParams

+ func (p *JobListParams) States(states ...rivertype.JobState) *JobListParams

+ type JobListResult struct

+ Jobs []*rivertype.JobRow

+ LastCursor *JobListCursor

Apr 16, 2024

Changes in this version

type Config

+ PollOnly bool

Mar 29, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) PeriodicJobs() *PeriodicJobBundle

+ type PeriodicJobBundle struct

+ func (b *PeriodicJobBundle) Add(periodicJob *PeriodicJob) rivertype.PeriodicJobHandle

+ func (b *PeriodicJobBundle) AddMany(periodicJobs []*PeriodicJob) []rivertype.PeriodicJobHandle

+ func (b *PeriodicJobBundle) Clear()

+ func (b *PeriodicJobBundle) Remove(periodicJobHandle rivertype.PeriodicJobHandle)

+ func (b *PeriodicJobBundle) RemoveMany(periodicJobHandles []rivertype.PeriodicJobHandle)

Mar 17, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) SubscribeConfig(config *SubscribeConfig) (<-chan *Event, func())

+ type SubscribeConfig struct

+ ChanSize int

+ Kinds []EventKind

Feb 29, 2024

Changes in this version

type JobListParams

+ func (p *JobListParams) Kinds(kinds ...string) *JobListParams

Feb 19, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) ID() string

type Config

+ ID string

Feb 11, 2024

Changes in this version

type Client

+ func (c *Client[TTx]) JobGet(ctx context.Context, jobID int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobGetTx(ctx context.Context, tx TTx, jobID int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobRetry(ctx context.Context, jobID int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobRetryTx(ctx context.Context, tx TTx, jobID int64) (*rivertype.JobRow, error)

Jan 23, 2024

Changes in this version

+ var ErrJobCancelledRemotely = JobCancel(errors.New("job cancelled remotely"))

+ var ErrNotFound = errors.New("not found")

type Client

+ func ClientFromContextSafely[TTx any](ctx context.Context) (*Client[TTx], error)

+ func ClientFromContext[TTx any](ctx context.Context) *Client[TTx]

+ func (c *Client[TTx]) JobCancel(ctx context.Context, jobID int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobCancelTx(ctx context.Context, tx TTx, jobID int64) (*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobList(ctx context.Context, params *JobListParams) ([]*rivertype.JobRow, error)

+ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListParams) ([]*rivertype.JobRow, error)

+ type JobListCursor struct

+ func JobListCursorFromJob(job *rivertype.JobRow) *JobListCursor

+ func (c *JobListCursor) UnmarshalText(text []byte) error

+ func (c JobListCursor) MarshalText() ([]byte, error)

+ type JobListOrderByField int

+ const JobListOrderByTime

+ type JobListParams struct

+ func NewJobListParams() *JobListParams

+ func (p *JobListParams) After(cursor *JobListCursor) *JobListParams

+ func (p *JobListParams) First(count int) *JobListParams

+ func (p *JobListParams) Metadata(json string) *JobListParams

+ func (p *JobListParams) OrderBy(field JobListOrderByField, direction SortOrder) *JobListParams

+ func (p *JobListParams) Queues(queues ...string) *JobListParams

+ func (p *JobListParams) State(state rivertype.JobState) *JobListParams

+ type SortOrder int

+ const SortOrderAsc

+ const SortOrderDesc

Jan 6, 2024

Changes in this version

type InsertOpts

+ Metadata []byte

Dec 2, 2023

Changes in this version

type Client

+ func (c *Client[TTx]) Stopped() <-chan struct{}

Nov 20, 2023

Changes in this version

+ const FetchCooldownDefault

+ const FetchCooldownMin

+ const FetchPollIntervalDefault

+ const FetchPollIntervalMin

+ const JobTimeoutDefault

+ const MaxAttemptsDefault

+ const PriorityDefault

+ const QueueDefault

+ const QueueNumWorkersMax

Nov 13, 2023

Changes in this version

+ const DefaultFetchCooldown

+ const DefaultFetchPollInterval

+ const DefaultJobTimeout

+ const DefaultMaxAttempts

+ const DefaultPriority

+ const DefaultQueue

+ const MaxQueueNumWorkers

+ const MinFetchCooldown

+ const MinFetchPollInterval

+ func AddWorkerSafely[T JobArgs](workers *Workers, worker Worker[T]) error

+ func AddWorker[T JobArgs](workers *Workers, worker Worker[T])

+ func JobCancel(err error) error

+ func JobSnooze(duration time.Duration) error

+ type AttemptError struct

+ At time.Time

+ Error string

+ Num int

+ Trace string

+ type Client struct

+ func NewClient[TTx any](driver riverdriver.Driver[TTx], config *Config) (*Client[TTx], error)

+ func (c *Client[TTx]) Insert(ctx context.Context, args JobArgs, opts *InsertOpts) (*JobRow, error)

+ func (c *Client[TTx]) InsertMany(ctx context.Context, params []InsertManyParams) (int64, error)

+ func (c *Client[TTx]) InsertManyTx(ctx context.Context, tx TTx, params []InsertManyParams) (int64, error)

+ func (c *Client[TTx]) InsertTx(ctx context.Context, tx TTx, args JobArgs, opts *InsertOpts) (*JobRow, error)

+ func (c *Client[TTx]) Start(ctx context.Context) error

+ func (c *Client[TTx]) Stop(ctx context.Context) error

+ func (c *Client[TTx]) StopAndCancel(ctx context.Context) error

+ func (c *Client[TTx]) Subscribe(kinds ...EventKind) (<-chan *Event, func())

+ type ClientRetryPolicy interface

+ NextRetry func(job *JobRow) time.Time

+ type Config struct

+ AdvisoryLockPrefix int32

+ CancelledJobRetentionPeriod time.Duration

+ CompletedJobRetentionPeriod time.Duration

+ DiscardedJobRetentionPeriod time.Duration

+ ErrorHandler ErrorHandler

+ FetchCooldown time.Duration

+ FetchPollInterval time.Duration

+ JobTimeout time.Duration

+ Logger *slog.Logger

+ PeriodicJobs []*PeriodicJob

+ Queues map[string]QueueConfig

+ ReindexerSchedule PeriodicSchedule

+ RescueStuckJobsAfter time.Duration

+ RetryPolicy ClientRetryPolicy

+ Workers *Workers

+ type DefaultClientRetryPolicy struct

+ func (p *DefaultClientRetryPolicy) NextRetry(job *JobRow) time.Time

+ type ErrorHandler interface

+ HandleError func(ctx context.Context, job *JobRow, err error) *ErrorHandlerResult

+ HandlePanic func(ctx context.Context, job *JobRow, panicVal any) *ErrorHandlerResult

+ type ErrorHandlerResult struct

+ SetCancelled bool

+ type Event struct

+ Job *JobRow

+ JobStats *JobStatistics

+ Kind EventKind

+ type EventKind string

+ const EventKindJobCancelled

+ const EventKindJobCompleted

+ const EventKindJobFailed

+ const EventKindJobSnoozed

+ type InsertManyParams struct

+ Args JobArgs

+ InsertOpts *InsertOpts

+ type InsertOpts struct

+ MaxAttempts int

+ Priority int

+ Queue string

+ ScheduledAt time.Time

+ Tags []string

+ UniqueOpts UniqueOpts

+ type Job struct

+ Args T

+ func JobCompleteTx[TDriver riverdriver.Driver[TTx], TTx any, TArgs JobArgs](ctx context.Context, tx TTx, job *Job[TArgs]) (*Job[TArgs], error)

+ type JobArgs interface

+ Kind func() string

+ type JobArgsWithInsertOpts interface

+ InsertOpts func() InsertOpts

+ type JobRow struct

+ Attempt int

+ AttemptedAt *time.Time

+ AttemptedBy []string

+ CreatedAt time.Time

+ EncodedArgs []byte

+ Errors []AttemptError

+ FinalizedAt *time.Time

+ ID int64

+ Kind string

+ MaxAttempts int

+ Priority int

+ Queue string

+ ScheduledAt time.Time

+ State JobState

+ Tags []string

+ type JobState string

+ const JobStateAvailable

+ const JobStateCancelled

+ const JobStateCompleted

+ const JobStateDiscarded

+ const JobStateRetryable

+ const JobStateRunning

+ const JobStateScheduled

+ type JobStatistics struct

+ CompleteDuration time.Duration

+ QueueWaitDuration time.Duration

+ RunDuration time.Duration

+ type PeriodicJob struct

+ func NewPeriodicJob(scheduleFunc PeriodicSchedule, constructorFunc PeriodicJobConstructor, ...) *PeriodicJob

+ type PeriodicJobConstructor func() (JobArgs, *InsertOpts)

+ type PeriodicJobOpts struct

+ RunOnStart bool

+ type PeriodicSchedule interface

+ Next func(time.Time) time.Time

+ func PeriodicInterval(interval time.Duration) PeriodicSchedule

+ type QueueConfig struct

+ MaxWorkers int

+ type UniqueOpts struct

+ ByArgs bool

+ ByPeriod time.Duration

+ ByQueue bool

+ ByState []JobState

+ type UnknownJobKindError struct

+ Kind string

+ func (e *UnknownJobKindError) Error() string

+ func (e *UnknownJobKindError) Is(target error) bool

+ type Worker interface

+ NextRetry func(job *Job[T]) time.Time

+ Timeout func(job *Job[T]) time.Duration

+ Work func(ctx context.Context, job *Job[T]) error

+ func WorkFunc[T JobArgs](f func(context.Context, *Job[T]) error) Worker[T]

+ type WorkerDefaults struct

+ func (w WorkerDefaults[T]) NextRetry(*Job[T]) time.Time

+ func (w WorkerDefaults[T]) Timeout(*Job[T]) time.Duration

+ type Workers struct

+ func NewWorkers() *Workers