jobs

package

v0.0.0-...-51ec7a6 Latest Latest Go to latest Published: Sep 23, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/ruby-dev/soro

Links

Open Source Insights

Documentation ¶

Overview ¶

Package jobs wraps River with Soro transactions, options, and telemetry.

Index ¶

func Perform[T river.JobArgs](ctx context.Context, args T, handler Handler[T]) error
func Register[T river.JobArgs](client *Client, handler Handler[T]) error
type Client
- func New(db *database.DB, observer *observability.Provider, logger *slog.Logger, ...) (*Client, error)
type Config
- func (config Config) Validate() error
type Handler
type Option
type Result
type UniqueConfig

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func Perform ¶

func Perform[T river.JobArgs](ctx context.Context, args T, handler Handler[T]) error

Perform executes a typed handler synchronously without inserting a River job. It is intended for focused handler tests; queue semantics remain the job client's responsibility.

func Register ¶

func Register[T river.JobArgs](client *Client, handler Handler[T]) error

Types ¶

type Client ¶

type Client struct {
	// contains filtered or unexported fields
}

func New ¶

func New(db *database.DB, observer *observability.Provider, logger *slog.Logger, config Config) (*Client, error)

func (*Client) Cancel ¶

func (client *Client) Cancel(ctx context.Context, id int64) (*rivertype.JobRow, error)

Cancel permanently stops a queued job and signals a running worker to cancel its context. Authorization remains the host application's concern.

func (*Client) Enabled ¶

func (client *Client) Enabled() bool

func (*Client) Enqueue ¶

func (client *Client) Enqueue(ctx context.Context, args river.JobArgs, options ...Option) (*Result, error)

func (*Client) EnqueueTx ¶

func (client *Client) EnqueueTx(ctx context.Context, args river.JobArgs, options ...Option) (*Result, error)

func (*Client) Get ¶

func (client *Client) Get(ctx context.Context, id int64) (*rivertype.JobRow, error)

Get returns the persisted queue record for a job. Hosts should authorize access to the record before exposing it outside their application boundary.

func (*Client) Migrate ¶

func (client *Client) Migrate(ctx context.Context) error

func (*Client) Queues ¶

func (client *Client) Queues() map[string]int

Queues returns the configured per-queue worker caps without exposing the mutable map retained by the River client.

func (*Client) Retry ¶

func (client *Client) Retry(ctx context.Context, id int64) (*rivertype.JobRow, error)

Retry makes a non-running job immediately available for another attempt. Authorization remains the host application's concern.

func (*Client) Start ¶

func (client *Client) Start(ctx context.Context) error

func (*Client) Stop ¶

func (client *Client) Stop(ctx context.Context) error

type Config ¶

type Config struct {
	WorkersEnabled  bool
	DefaultQueue    string
	Queues          map[string]int
	Schema          string
	ShutdownTimeout time.Duration
}

func (Config) Validate ¶

func (config Config) Validate() error

type Handler ¶

type Handler[T river.JobArgs] func(context.Context, T) error

type Option ¶

type Option func(*insertSettings) error

func Delay ¶

func Delay(duration time.Duration) Option

func MaxAttempts ¶

func MaxAttempts(attempts int) Option

func Priority ¶

func Priority(priority int) Option

func Queue ¶

func Queue(name string) Option

func Unique ¶

func Unique(config UniqueConfig) Option

func UniqueByArgs ¶

func UniqueByArgs() Option

type Result ¶

type Result struct {
	ID        int64
	Duplicate bool
}

type UniqueConfig ¶

type UniqueConfig struct {
	ByArgs   bool
	ByQueue  bool
	ByPeriod time.Duration
}

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL