driver

type AttemptError struct {
	// At is the database time at which the failure was recorded.
	At      time.Time `json:"at"`
	Attempt int       `json:"attempt"`
	Error   string    `json:"error"`
	// Trace is the stack trace when the attempt panicked.
	Trace string `json:"trace,omitempty"`
}

type BatchInsertParams struct {
	Total int
	// Callbacks are inserted when the batch completes: OnSuccess if no job
	// was cancelled or discarded, OnFailure otherwise, and OnComplete
	// either way. Each callback's Metadata gains a "batch_id".
	OnSuccess, OnFailure, OnComplete *JobInsertParams
	Metadata                         json.RawMessage
}

type BatchRow struct {
	ID JobID
	// Name is set for workflows.
	Name    string
	Pending int
	Failed  int
	Total   int
	// CompletedAt is set once every job has been finalized.
	CompletedAt time.Time
	CreatedAt   time.Time
	Metadata    json.RawMessage
}

type Capabilities struct {
	// Copy reports whether JobInsertCopy is implemented (COPY on Postgres).
	Copy bool
	// Listen reports whether Listener is implemented (LISTEN/NOTIFY on
	// Postgres).
	Listen bool
}

type ClientRegisterParams struct {
	Hostname string
	TTL      time.Duration
	Info     json.RawMessage
}

type ClientRenewParams struct {
	ClientID int64
	TTL      time.Duration
}

type ClientRenewResult struct {
	Renewed bool
	// CancelRequested lists the client's running jobs that have a pending
	// cancel request.
	CancelRequested []JobID
	// PausedQueues lists the queues currently paused.
	PausedQueues []string
	// LimitedQueues lists the queues with a global, rate or partition
	// limit, which must be claimed through the limited path.
	LimitedQueues []string
}

type ClientRow struct {
	ID        int64
	Hostname  string
	StartedAt time.Time
	ExpiresAt time.Time
	Info      json.RawMessage
}

type ConflictAction int

type DependencyFailure string

type Driver[TTx any] interface {
	// Executor returns the pool-scoped executor.
	Executor() Executor
	// UnwrapTx returns an executor that runs inside the caller's transaction.
	UnwrapTx(tx TTx) Executor
	// Capabilities reports which optional strategies the engine supports.
	Capabilities() Capabilities
	// Migrator returns the engine's schema migrator.
	Migrator() Migrator
	// Listener opens a connection for push notifications. Engines without
	// them return ErrNotSupported, and the core polls instead.
	Listener(ctx context.Context) (Listener, error)
}

type Executor interface {
	// Now returns the database's clock.
	Now(ctx context.Context) (time.Time, error)
	// SchemaVersion returns the newest applied schema version, or 0 if the
	// schema has not been installed. It takes no lock, so it does not wait
	// for a migration in progress.
	SchemaVersion(ctx context.Context) (int, error)

	// JobInsertMany inserts jobs in one statement and returns one result per
	// input, in input order. An input whose unique key matches a live job is
	// not inserted; its result carries the existing job and Duplicate = true.
	// With ConflictReplace, the existing job's args, metadata, priority,
	// max attempts and schedule are replaced unless it is running. Inputs are
	// already deduplicated by unique key within the batch.
	JobInsertMany(ctx context.Context, params []JobInsertParams, opts JobInsertOpts) ([]JobInsertResult, error)
	// JobInsertCopy bulk-loads jobs without conflict handling. Callers use it
	// only for batches with no unique keys, and only if Capabilities().Copy.
	JobInsertCopy(ctx context.Context, params []JobInsertParams, opts JobInsertOpts) ([]JobInsertResult, error)
	// JobClaim atomically moves up to Limit claimable jobs in a queue to
	// running, owned by ClientID, and returns them in claim order. With
	// Limited, the claim runs inside a transaction that locks the queue row
	// and applies its global, rate and partition limits; the result then
	// says how long to wait when the rate limit allowed nothing.
	JobClaim(ctx context.Context, params JobClaimParams) (JobClaimResult, error)
	// JobAge raises the priority of jobs in a queue that have been claimable
	// for longer than after, one level per call, and returns how many.
	JobAge(ctx context.Context, queue string, after time.Duration) (int64, error)
	// JobFinalizeMany applies a batch of results in one statement. A result is
	// applied only if its job is still running and owned by the client that
	// produced it; the IDs that were applied are returned, and the caller
	// treats the rest as fenced.
	JobFinalizeMany(ctx context.Context, params JobFinalizeParams) ([]JobID, error)
	// JobGet returns a job by ID, live or from history.
	JobGet(ctx context.Context, id JobID) (*JobRow, error)
	// JobList returns jobs matching the filter in ID order, after the cursor.
	JobList(ctx context.Context, params JobListParams) ([]*JobRow, error)
	// JobCancel cancels a job. A waiting job moves to history as cancelled.
	// A running job gets a cancel request that its client picks up through
	// ChannelControl and on its next lease renewal. A finalized job is
	// returned unchanged. The job as it is afterwards is returned.
	JobCancel(ctx context.Context, id JobID) (*JobRow, error)
	// JobRetry makes a job available now: a cancelled or discarded job is
	// moved back from history (with one more attempt allowed if it had run
	// out), and a waiting job has its schedule brought forward. Either way
	// the job's TTL is cleared, as an operator overriding the wait. It returns
	// ErrJobRunning for a running job and ErrUniqueConflict if a live job
	// holds the same unique key.
	JobRetry(ctx context.Context, id JobID) (*JobRow, error)
	// JobDiscardExpired moves waiting jobs whose TTL has passed to history
	// as discarded, and returns them.
	JobDiscardExpired(ctx context.Context, limit int) ([]*JobRow, error)
	// JobRescueCandidates returns running jobs whose owning client has no
	// unexpired lease, plus, if StuckAfter is positive, running jobs claimed
	// longer ago than that. The caller finalizes them.
	JobRescueCandidates(ctx context.Context, params JobRescueParams) ([]*JobRow, error)

	// ClientRegister creates a lease row for a client process and returns its ID.
	ClientRegister(ctx context.Context, params ClientRegisterParams) (int64, error)
	// ClientRenew extends an unexpired lease. Renewed is false if the lease
	// had already expired, in which case the client has been fenced. The
	// result also carries control state the client re-reads on every
	// renewal, in case a notification was missed.
	ClientRenew(ctx context.Context, params ClientRenewParams) (ClientRenewResult, error)
	// ClientDelete removes a client's lease row.
	ClientDelete(ctx context.Context, clientID int64) error
	// ClientPruneExpired removes lease rows that have expired and returns
	// how many.
	ClientPruneExpired(ctx context.Context) (int64, error)
	// ClientList returns every client lease row, live or expired.
	ClientList(ctx context.Context) ([]*ClientRow, error)

	// Notify sends a notification per payload on a channel, delivered to
	// listeners when the surrounding transaction commits. Engines without
	// notifications return nil.
	Notify(ctx context.Context, channel string, payloads []string) error

	// LeaderAttempt takes or renews the leader lease for a client. It
	// returns true if the client is the leader afterwards.
	LeaderAttempt(ctx context.Context, params LeaderParams) (bool, error)
	// LeaderResign gives up leadership if the client holds it, and signals
	// other clients to elect a replacement.
	LeaderResign(ctx context.Context, clientID int64) error

	// QueueEnsure records queues a client works, so they are listed even
	// before any job is inserted.
	QueueEnsure(ctx context.Context, names []string) error
	// QueuePause stops every client from claiming from a queue, and
	// signals them through ChannelControl.
	QueuePause(ctx context.Context, name string) error
	// QueueResume undoes QueuePause.
	QueueResume(ctx context.Context, name string) error
	// QueueList returns every known queue.
	QueueList(ctx context.Context) ([]*QueueRow, error)
	// QueueSetLimits sets a queue's cluster-wide limits, creating the queue
	// if needed. Zero values remove the corresponding limit.
	QueueSetLimits(ctx context.Context, limits QueueLimits) error

	// BatchInsert creates a batch expecting Total jobs and returns its ID.
	// The jobs are inserted separately with their BatchID set.
	BatchInsert(ctx context.Context, params BatchInsertParams) (JobID, error)
	// BatchGet returns a batch's progress.
	BatchGet(ctx context.Context, id JobID) (*BatchRow, error)
	// WorkflowInsert inserts a batch, its jobs and the dependencies between
	// them atomically. Jobs with dependencies start pending. Results are in
	// input order.
	WorkflowInsert(ctx context.Context, params WorkflowInsertParams) (*WorkflowInsertResult, error)
	// WorkflowGet returns a workflow's progress, its jobs (live and
	// finished) and its edges.
	WorkflowGet(ctx context.Context, id JobID) (*WorkflowRow, error)

	// PeriodicInsert inserts the job for a periodic slot if that slot has
	// not been inserted yet, atomically, and reports whether it did. A slot
	// therefore produces at most one job across leaders.
	PeriodicInsert(ctx context.Context, params PeriodicInsertParams) (*JobRow, bool, error)
	// PeriodicLastSlots returns the last inserted slot per periodic job.
	PeriodicLastSlots(ctx context.Context) (map[string]time.Time, error)

	// Stats returns queue depths and cluster state.
	Stats(ctx context.Context) (*Stats, error)

	// SubscriptionUpsert records subscriptions, updating pattern, queue and
	// settings of existing ones by name.
	SubscriptionUpsert(ctx context.Context, subs []SubscriptionRow) error
	// SubscriptionList returns every subscription.
	SubscriptionList(ctx context.Context) ([]*SubscriptionRow, error)
	// MessagePublish inserts one delivery per subscription whose pattern
	// matches the topic, in one statement, and returns them. A delivery
	// whose dedup key matches a live one is reported as a duplicate.
	MessagePublish(ctx context.Context, params MessagePublishParams) ([]JobInsertResult, error)

	// StreamAppend appends one event to the stream log and returns it with
	// its position.
	StreamAppend(ctx context.Context, params StreamAppendParams) (*StreamEvent, error)
	// StreamRead pages through the committed events after a position whose
	// topic matches a pattern, in position order. It is for browsing; a
	// consumer is what reads the log without missing late commits.
	StreamRead(ctx context.Context, params StreamReadParams) ([]*StreamEvent, error)
	// StreamConsumerUpsert records consumers, updating pattern, queue and
	// settings of existing ones by name; a new consumer starts at Start.
	StreamConsumerUpsert(ctx context.Context, consumers []StreamConsumerRow) error
	// StreamConsumerList returns every consumer with its position.
	StreamConsumerList(ctx context.Context) ([]*StreamConsumerRow, error)
	// StreamConsumerSeek moves a consumer's position: events after it are
	// delivered (again).
	StreamConsumerSeek(ctx context.Context, name string, params StreamSeekParams) error
	// StreamPump turns up to limit events after a consumer's position into
	// delivery jobs and advances the position, atomically. Two leaders
	// pumping the same consumer serialize on its row.
	StreamPump(ctx context.Context, name string, limit int) (StreamPumpResult, error)

	// HistoryMaintain enforces retention on finalized jobs and prepares
	// storage for the near future, in whatever way suits the engine (dropping
	// time partitions on Postgres). It is idempotent and may run on two
	// leaders at once.
	HistoryMaintain(ctx context.Context, params HistoryMaintainParams) (HistoryMaintainResult, error)
	// JobsMaintain keeps the live table fit between the engine's own
	// maintenance passes: it reclaims the space and index entries of jobs
	// that have run and left once enough have accumulated (a hundred
	// thousand on Postgres), and refreshes planner statistics when the
	// planner's idea of the table's size is off by an order of magnitude. It
	// never waits for a lock, so it costs nothing while the engine's own
	// pass is running. The live table churns at the job rate, and on
	// Postgres every claim walks the claim index past the entries of jobs
	// that have left, so pickup latency climbs between vacuums; autovacuum
	// looks only every autovacuum_naptime (a minute by default). Engines
	// without such maintenance return zero values.
	JobsMaintain(ctx context.Context) (JobsMaintainResult, error)
}

type HistoryMaintainParams struct {
	// CompletedRetention is how long completed jobs are kept.
	CompletedRetention time.Duration
	// FailedRetention is how long cancelled and discarded jobs are kept.
	FailedRetention time.Duration
	// StreamRetention is how long stream events are kept.
	StreamRetention time.Duration
}

type HistoryMaintainResult struct {
	// Created and Dropped name storage units (partitions on Postgres).
	Created []string
	Dropped []string
	// Pruned counts rows deleted individually, outside partition drops.
	Pruned int64
}

type JobClaimParams struct {
	Queue    string
	ClientID int64
	Limit    int
	// Limited selects the path that locks the queue row and applies its
	// limits. Unlimited queues never touch the row.
	Limited bool
}

type JobClaimResult struct {
	Jobs []*JobRow
	// Wait, when positive, is how long the rate limit will allow nothing.
	Wait time.Duration
}

type JobDependency struct {
	Job, DependsOn int
	OnFailure      DependencyFailure
}

type JobEdge struct {
	Job, DependsOn JobID
}

type JobFinalize struct {
	ID JobID
	// AttemptedBy is the client that claimed the attempt. The engine applies
	// the result only if the job is still running and owned by this client.
	AttemptedBy int64
	// State is the state to move to: completed, cancelled or discarded (the
	// job moves to history) or retryable or scheduled (it stays live).
	State JobState
	// Delay is how long from database now() the job waits before it can be
	// claimed again. Only for retryable and scheduled.
	Delay time.Duration
	// Snooze gives the attempt back, so the retry does not count against
	// MaxAttempts.
	Snooze bool
	// SnoozeReason is stored as the job's SnoozeReason when Snooze is set.
	SnoozeReason string
	// Error is appended to the job's errors. Its At field is set by the
	// engine from database time.
	Error *AttemptError
	// Output is stored with the finalized job.
	Output json.RawMessage
	// Archive controls whether a terminal job is written to history. Cancelled
	// and discarded jobs are always archived; completed jobs may skip it.
	Archive bool
}

type JobFinalizeParams struct {
	Jobs []JobFinalize
}

type JobID [16]byte

type JobInsertOpts struct {
	// Notify wakes workers on the inserted queues once the insert commits.
	Notify bool
	// OnConflict applies to inputs with a unique key.
	OnConflict ConflictAction
}

type JobInsertParams struct {
	Kind        string
	Queue       string
	Priority    int
	MaxAttempts int
	// ScheduledAt is zero to run as soon as possible.
	ScheduledAt time.Time
	Args        json.RawMessage
	Metadata    json.RawMessage
	// UniqueKey is empty for non-unique jobs.
	UniqueKey string
	// OrderingKey, if set, serializes the job with others of the same key
	// in its queue: at most one runs at a time, oldest first.
	OrderingKey string
	// PartitionKey groups the job for partitioned limits.
	PartitionKey string
	// BatchID, if set, counts the job toward a batch.
	BatchID JobID
	// TTL, if positive, sets ExpiresAt to this long after database now().
	// A job past ExpiresAt is not claimed while it waits (available,
	// scheduled or retryable) and JobDiscardExpired discards it.
	TTL time.Duration
	// Await marks the job so that its finalize is announced on ChannelDone.
	Await bool
}

type JobInsertResult struct {
	Job *JobRow
	// Duplicate is true when a live job with the same unique key already
	// existed. Job is then that existing job.
	Duplicate bool
}

type JobListParams struct {
	Queue  string
	Kinds  []string
	States []JobState
	// After is the cursor: only jobs with a greater ID are returned.
	After JobID
	Limit int
}

type JobRescueParams struct {
	// StuckAfter, if positive, also selects running jobs claimed longer ago
	// than this, whatever the state of their client's lease.
	StuckAfter time.Duration
	Limit      int
}

type JobRow struct {
	ID          JobID
	Kind        string
	Queue       string
	State       JobState
	Priority    int
	Attempt     int
	MaxAttempts int
	ScheduledAt time.Time
	// AttemptedAt is the time of the latest claim, or zero if never claimed.
	AttemptedAt time.Time
	// AttemptedBy is the client ID holding the claim, or zero.
	AttemptedBy  int64
	Args         json.RawMessage
	Metadata     json.RawMessage
	Errors       []AttemptError
	UniqueKey    string
	OrderingKey  string
	PartitionKey string
	// BatchID is the batch the job belongs to, or zero.
	BatchID JobID
	// ExpiresAt is the TTL deadline, or zero. A job still waiting to run
	// at this time, for its first attempt or a later one, is discarded.
	ExpiresAt time.Time
	// CancelRequestedAt is set when JobCancel is called on a running job.
	CancelRequestedAt time.Time
	Await             bool
	CreatedAt         time.Time
	// SnoozeReason is the reason the worker gave when it last snoozed the
	// job, while the job waits out that snooze. It is empty in any state but
	// scheduled: the claim, any other finalize, a retry and a replacing
	// insert clear it, and a row in another state never reports one.
	SnoozeReason string
	// FinalizedAt is set for jobs in history.
	FinalizedAt time.Time
	// Output is the value recorded by the worker, for jobs in history.
	Output json.RawMessage
}

type JobState string

type JobsMaintainResult struct {
	Vacuumed bool
	Analyzed bool
}

type LeaderParams struct {
	ClientID int64
	TTL      time.Duration
}

type Listener interface {
	// Listen subscribes to channels.
	Listen(ctx context.Context, channels ...string) error
	// Next blocks until a notification arrives or ctx is done.
	Next(ctx context.Context) (Notification, error)
	// Close releases the connection.
	Close(ctx context.Context) error
}

type MessagePublishParams struct {
	Topic   string
	Payload json.RawMessage
	// Headers are stored in each delivery's metadata.
	Headers map[string]string
	// OrderingKey serializes deliveries with the same key per queue.
	OrderingKey string
	// DedupKey makes the publish idempotent while an earlier delivery is
	// live: it becomes each delivery's unique key.
	DedupKey string
	// Delay schedules the deliveries for later.
	Delay time.Duration
	// TTL sets each delivery's ExpiresAt, as JobInsertParams.TTL does.
	TTL      time.Duration
	Priority int
	Await    bool
	// Notify wakes workers on the delivery queues once the publish commits.
	Notify bool
}

type MigrationExecutor interface {
	// Versions returns the applied schema versions in ascending order, or
	// nil if the schema has never been installed.
	Versions(ctx context.Context) ([]int, error)
	// Apply runs a migration script and records (up) or removes (down) its
	// version, atomically.
	Apply(ctx context.Context, version int, script string, up bool) error
	// Close releases the lock and the connection.
	Close(ctx context.Context) error
}

type Migrator interface {
	// Lock takes the engine's cross-process migration lock on a dedicated
	// connection (never a pooled one, so that processes waiting for the lock
	// cannot starve the migration of connections) and returns the executor to
	// migrate with. The caller must Close it to release the lock.
	Lock(ctx context.Context) (MigrationExecutor, error)
}

type Notification struct {
	Channel string
	Payload string
}

type PeriodicInsertParams struct {
	Name string
	Slot time.Time
	Job  JobInsertParams
}

type QueueLimits struct {
	Name string
	// GlobalLimit caps running jobs across all clients.
	GlobalLimit int
	// RatePerSec caps claims per second across all clients, with a burst
	// of RateBurst (defaulting to one second's worth).
	RatePerSec float64
	RateBurst  int
	// PartitionLimit caps running jobs per partition key.
	PartitionLimit int
	// Aging raises a waiting job's priority by one level each time it has
	// waited this long, so low priorities cannot starve. Zero disables it.
	Aging time.Duration
}

type QueueRow struct {
	Name string
	// PausedAt is zero when the queue is not paused.
	PausedAt  time.Time
	UpdatedAt time.Time
	Limits    QueueLimits
}

type QueueStats struct {
	Available int
	Scheduled int
	Retryable int
	Running   int
	// OldestAvailable is the age of the oldest job that could be claimed
	// now, or zero.
	OldestAvailable time.Duration
	// CompletedLastMinute counts jobs completed in the last minute.
	CompletedLastMinute int
	Paused              bool
}

type Stats struct {
	Queues map[string]*QueueStats
	// LiveClients counts clients with an unexpired lease.
	LiveClients int
	// Leader is the leader's client ID, or 0 if the lease is free or
	// expired.
	Leader int64
	// RunningByClient counts running jobs per client ID.
	RunningByClient map[int64]int
}

type StreamAppendParams struct {
	Topic   string
	Key     string
	Payload json.RawMessage
	Headers map[string]string
	// Notify signals ChannelStream from inside the statement.
	Notify bool
}

type StreamConsumerRow struct {
	Name    string
	Pattern string
	Kind    string
	Queue   string
	// MaxAttempts is the retry budget of deliveries, or 0 for the default
	// (25, or 10 for events with a key).
	MaxAttempts int
	// Metadata is merged into every delivery's metadata.
	Metadata json.RawMessage
	// Start applies when the consumer is created; it defaults to latest.
	Start StreamStart
	// Snapshot is the engine's record of what the consumer has read to:
	// on Postgres, the pg_snapshot every delivered transaction is visible
	// in.
	Snapshot string
	// Position is the last event delivered of the delta being read, or
	// zero between deltas.
	Position    StreamPosition
	DeliveredAt time.Time
	CreatedAt   time.Time
}

type StreamEvent struct {
	Position StreamPosition
	Topic    string
	// Key, if set, becomes the ordering key of the event's deliveries.
	Key       string
	Payload   json.RawMessage
	Headers   map[string]string
	MessageID string
	CreatedAt time.Time
}

type StreamPosition struct {
	Xid uint64
	Seq int64
}

type StreamPumpResult struct {
	Delivered int
	// Position is the consumer's position afterwards.
	Position StreamPosition
	// Queues received deliveries.
	Queues []string
}

type StreamReadParams struct {
	// Pattern is an AMQP topic pattern; empty matches every topic.
	Pattern string
	After   StreamPosition
	Limit   int
}

type StreamSeekParams struct {
	// Position, if set, becomes the consumer's position: the next event
	// delivered is the one after it.
	Position StreamPosition
	// Time moves the consumer to just before the first event appended at
	// or after it.
	Time time.Time
	// Earliest moves the consumer before every retained event.
	Earliest bool
	// Latest moves the consumer past every event appended so far.
	Latest bool
}

type StreamStart string

type SubscriptionRow struct {
	Name    string
	Pattern string
	Kind    string
	Queue   string
	// MaxAttempts is the retry budget of deliveries, or 0 for the default
	// (25, or 10 for deliveries with an ordering key).
	MaxAttempts int
	// Metadata is merged into every delivery's metadata.
	Metadata  json.RawMessage
	CreatedAt time.Time
}

type WorkflowInsertParams struct {
	Name string
	// Jobs must have no unique keys; a skipped job would leave its
	// dependents waiting forever.
	Jobs []JobInsertParams
	Deps []JobDependency
	// Callbacks and Metadata are the batch's; see BatchInsertParams.
	OnSuccess, OnFailure, OnComplete *JobInsertParams
	Metadata                         json.RawMessage
	// Notify sends insert notifications for the jobs that start
	// available from inside the statement.
	Notify bool
}

type WorkflowInsertResult struct {
	ID   JobID
	Jobs []JobInsertResult
}

type WorkflowRow struct {
	Batch BatchRow
	// Jobs are in insertion order, live and finished alike.
	Jobs  []*JobRow
	Edges []JobEdge
}