riverx

type Config ¶

type Config struct {
	// Queues maps a queue to the number of jobs it works at once. Capacity is an
	// environment parameter: production and staging differ in it, the jobs do not.
	Queues map[string]int

	// Work turns job processing on. With it off the instance only inserts jobs, which
	// is how an API instance and a worker instance of one binary split the load.
	Work bool

	JobTimeout time.Duration // how long one job may run

	// How long finished jobs stay in the database before River deletes them.
	CompletedRetention time.Duration
	CancelledRetention time.Duration
	DiscardedRetention time.Duration
}

Config holds module settings. Load fills it from the environment; the platform generator writes the Load call into the project's config.gen.go.

func Load(l *confx.Loader) Config

type ConfigurableArgs ¶

type ConfigurableArgs interface {
	river.JobArgs
	InsertOpts() *river.InsertOpts
}

ConfigurableArgs is a job that carries its own insert options: queue, priority, attempts, uniqueness. It is the convention of taply — the options live next to the job kind, so every place that inserts the job gets the same ones.

func (OrderCleanupArgs) InsertOpts() *river.InsertOpts {
	return &river.InsertOpts{Queue: "maintenance", MaxAttempts: 1}
}

type Module ¶

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

Module implements platform.Module.

func New(cfg Config) *Module

func (m *Module) Health(context.Context) error

func (m *Module) Init(ctx context.Context, app *platform.App) error

func (m *Module) Name() string

func (m *Module) Scheduled() map[string]string

func (m *Module) Start(ctx context.Context) error

func (m *Module) Stop(ctx context.Context) error

type PeriodicJob ¶

type PeriodicJob struct {
	Name       string                   // unique name, used for the schedule and in logs
	Schedule   func() string            // cron expression, @daily, @every 15m
	Enabled    func() bool              // nil means always enabled
	Args       func() river.JobArgs     // the job to insert each time
	Opts       func() *river.InsertOpts // nil means the options the job carries
	RunOnStart bool
}

PeriodicJob is a job River inserts on a schedule. Schedule and Enabled are functions, so they can come from business settings: a changed value takes effect without a restart.

riverx.Periodic(app, riverx.PeriodicJob{
	Name:     "orders.cleanup",
	Schedule: s.OrdersCleanup().Schedule,
	Enabled:  s.OrdersCleanup().Enabled,
	Args:     func() river.JobArgs { return jobs.CleanupArgs{} },
})

type Queue ¶

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

Queue inserts jobs. It is available from wireDomain on, before the client starts, so domain services can keep it; inserting before Start returns an error.

func QueueFrom(app *platform.App) *Queue

func (q *Queue) Client() *river.Client[pgx.Tx]

func (q *Queue) Insert(ctx context.Context, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)

func (q *Queue) InsertAt(ctx context.Context, args river.JobArgs, at time.Time) (*rivertype.JobInsertResult, error)

func (q *Queue) InsertMany(ctx context.Context, args ...river.JobArgs) ([]*rivertype.JobInsertResult, error)

func (q *Queue) InsertTx(ctx context.Context, tx pgx.Tx, args river.JobArgs, opts *river.InsertOpts) (*rivertype.JobInsertResult, error)

type Registry ¶

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

Registry holds what the project declares. It is filled from wireDomain.