queue

type Config struct {
	// Driver is where jobs are kept: sync (run at once), memory, database,
	// or one passed to ForApp (redis). QUEUE_DRIVER, default sync.
	Driver string `env:"QUEUE_DRIVER" default:"sync"`
	// Default is the queue jobs go on without [OnQueue], and the one
	// workers take jobs from without [Queues]. QUEUE_DEFAULT, default
	// "default".
	Default string `env:"QUEUE_DEFAULT" default:"default"`
	// Tries is how many times a job runs before it fails for good, unless
	// its type says otherwise ([Tries]). QUEUE_TRIES, default 3.
	Tries int `env:"QUEUE_TRIES" default:"3"`
	// Timeout is how long an attempt may run ([Timeout]). QUEUE_TIMEOUT,
	// default 1m.
	Timeout time.Duration `env:"QUEUE_TIMEOUT" default:"1m"`
	// Backoff is the wait before the first retry, doubling for each
	// later one ([Backoff]). QUEUE_BACKOFF, default 10s.
	Backoff time.Duration `env:"QUEUE_BACKOFF" default:"10s"`
	// MaxBackoff caps the doubling. QUEUE_BACKOFF_MAX, default 10m.
	MaxBackoff time.Duration `env:"QUEUE_BACKOFF_MAX" default:"10m"`
	// Poll is how long an idle worker waits before looking for jobs
	// again. QUEUE_POLL, default 1s.
	Poll time.Duration `env:"QUEUE_POLL" default:"1s"`
	// Table is the database driver's table of jobs. QUEUE_TABLE, default
	// jobs.
	Table string `env:"QUEUE_TABLE" default:"jobs"`
	// FailedTable is the database driver's table of failed jobs.
	// QUEUE_FAILED_TABLE, default failed_jobs.
	FailedTable string `env:"QUEUE_FAILED_TABLE" default:"failed_jobs"`
	// Prefix starts the keys of the Redis driver, so apps (and the cache
	// and sessions) can share a server. QUEUE_PREFIX, default APP_NAME
	// followed by ":queue:" ("blog:queue:").
	Prefix string `env:"QUEUE_PREFIX"`
}

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

type DispatchOption func(*dispatchOptions)

type Dispatched struct {
	// ID is the job's ID.
	ID string
	// Job is the job type's name ("SendWelcome", "mail:send").
	Job string
	// Queue is the queue it went on.
	Queue string
	// Delay is its delay ([Delay]).
	Delay time.Duration
	// Data is the job, or a function job's payload, as JSON.
	Data json.RawMessage
}

type Driver struct {
	// Name is the value of QUEUE_DRIVER that selects the driver.
	Name string
	// Open returns the store for the app. It may add providers to the
	// app, for example to check a server when the app boots. A nil store
	// runs jobs at once.
	Open func(app *anetos.App, cfg Config) (Store, error)
}

type FailedCounter interface {
	// CountFailed returns the number of failed jobs.
	CountFailed(ctx context.Context) (int64, error)
}

type FailedFinder interface {
	// FindFailed returns the failed job id, and whether there is one.
	FindFailed(ctx context.Context, id string) (FailedJob, bool, error)
}

type FailedJob struct {
	// ID is the job's ID.
	ID string
	// Queue is the queue it was on.
	Queue string
	// Payload is the job's encoding.
	Payload []byte
	// Error is the error of its last attempt.
	Error string
	// Attempts is the number of times it ran.
	Attempts int
	// FailedAt is when it failed, by the store's clock.
	FailedAt time.Time
}

type Failer interface {
	// Failed handles the job's failure, say by telling someone.
	Failed(ctx context.Context, err error)
}

type Info struct {
	// ID identifies the job across its attempts: use it to make the job
	// idempotent.
	ID string
	// Job is the job type's name.
	Job string
	// Queue is the queue it came from.
	Queue string
	// Attempt is this attempt's number, from 1.
	Attempt int
	// Tries is the number of attempts the job may have.
	Tries int
}

type Job interface {
	// Handle does the work. An error retries the job, after a backoff,
	// until it has used its tries; a [Permanent] error fails it at once.
	Handle(ctx context.Context) error
}

type JobOption func(*jobType) error

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

type Message struct {
	// ID identifies the job, also once it has failed: a UUID.
	ID string
	// Queue is the name of the queue.
	Queue string
	// Payload is the job's encoding.
	Payload []byte
}

type Option func(*Queue)

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

type Reservation struct {
	// ID is the job's ID.
	ID string
	// Queue is the name of its queue.
	Queue string
	// Payload is the job's encoding.
	Payload []byte
	// Attempts counts the reservations of the job, this one included.
	Attempts int
	// Token identifies this reservation to the store.
	Token string
}

type Store interface {
	// Push adds a job to m.Queue, available after delay (0 for now).
	Push(ctx context.Context, m Message, delay time.Duration) error
	// Reserve leases the next available job of queue for lease, adding
	// one to its attempts. It returns nil, nil when none is available.
	// Jobs become available in order of the time they became available,
	// as far as the store's clock allows.
	Reserve(ctx context.Context, queue string, lease time.Duration) (*Reservation, error)
	// Delete removes a reserved job, once it is done. It returns
	// [ErrLeaseLost] if the job isn't reserved with r's token anymore.
	Delete(ctx context.Context, r *Reservation) error
	// Release ends a reservation: the job becomes available again after
	// delay. With refund, the attempt doesn't count (the worker stopped
	// before the job could finish). It returns [ErrLeaseLost] if the job
	// isn't reserved with r's token anymore.
	Release(ctx context.Context, r *Reservation, delay time.Duration, refund bool) error
	// Fail removes a reserved job and keeps it as failed, with the error
	// message (replacing a failed job with the same ID). It returns
	// [ErrLeaseLost], and keeps nothing, if the job isn't reserved with
	// r's token anymore.
	Fail(ctx context.Context, r *Reservation, errMsg string) error
	// Size returns the number of jobs in queue, reserved ones included.
	Size(ctx context.Context, queue string) (int64, error)
	// Clear removes every job of queue, reserved ones included, and
	// returns how many there were.
	Clear(ctx context.Context, queue string) (int64, error)

	// Failed returns up to limit failed jobs, the latest first, after
	// skipping offset of them.
	Failed(ctx context.Context, offset, limit int) ([]FailedJob, error)
	// Retry pushes the failed job id back to its queue, with no attempts,
	// and stops keeping it as failed. It reports whether there was one.
	Retry(ctx context.Context, id string) (bool, error)
	// Forget removes the failed job id, and reports whether there was one.
	Forget(ctx context.Context, id string) (bool, error)
	// Flush removes every failed job, and returns how many there were.
	Flush(ctx context.Context) (int64, error)

	// Close releases the store's resources.
	Close() error
}

type WorkOption func(*workOptions) error