dlq

type CountOpts struct {
	Queue        string
	Replayed     *bool
	FailedBefore time.Time
}

type Enqueuer func(ctx context.Context, j *job.Job) error

type Entry struct {
	ID         id.DLQID   `json:"id"`
	JobID      id.JobID   `json:"job_id"`
	JobName    string     `json:"job_name"`
	Queue      string     `json:"queue"`
	Payload    []byte     `json:"payload"`
	Error      string     `json:"error"`
	RetryCount int        `json:"retry_count"`
	MaxRetries int        `json:"max_retries"`
	ScopeAppID string     `json:"scope_app_id,omitempty"`
	ScopeOrgID string     `json:"scope_org_id,omitempty"`
	FailedAt   time.Time  `json:"failed_at"`
	ReplayedAt *time.Time `json:"replayed_at,omitempty"`
	CreatedAt  time.Time  `json:"created_at"`

	// ReplayedJobID is the job a replay created, or for a retry the
	// retried job itself. ClaimReplay sets it together with ReplayedAt,
	// and it is nil while the entry is unreplayed. An entry marked by the
	// older ReplayDLQ has ReplayedAt set and this nil.
	ReplayedJobID *id.JobID `json:"replayed_job_id,omitempty"`

	// Priority is the claim ordering the job was enqueued with.
	Priority int `json:"priority,omitempty"`

	// Timeout is how long the handler was allowed to run.
	Timeout time.Duration `json:"timeout,omitempty"`

	// LeaseTTL is how long each renewal extended the job's lease. Losing
	// it is the reason this block exists: a six-hour job replayed without
	// it falls back to the pool default, which is measured in seconds, so
	// its lease lapses mid-run and it is reclaimed and restarted over and
	// over without ever finishing. That is the exact failure long-running
	// jobs carry a per-job TTL to avoid.
	LeaseTTL time.Duration `json:"lease_ttl,omitempty"`

	// ArtifactBindings is the encoded map of declared input names to
	// artifacts, carried verbatim. A handler that declares inputs cannot
	// run without them.
	ArtifactBindings []byte `json:"artifact_bindings,omitempty"`

	// Resources and ResourceLimits are what the job asked for and what it
	// was capped at. Without them a replayed job looks free to schedule
	// and can be claimed by a worker that cannot actually host it.
	Resources      resource.Set `json:"resources,omitempty"`
	ResourceLimits resource.Set `json:"resource_limits,omitempty"`

	// ResourceClass is the named class the job was placed in.
	ResourceClass string `json:"resource_class,omitempty"`

	// InputBytes and PrimaryInputHash are derived from the bindings by the
	// engine at enqueue. Replay does not run that derivation, so they are
	// carried rather than recomputed, and stay consistent with the
	// bindings above.
	InputBytes       int64  `json:"input_bytes,omitempty"`
	PrimaryInputHash string `json:"primary_input_hash,omitempty"`
}

type ListOpts struct {
	// Limit is the maximum number of entries to return. Zero means no limit.
	Limit int
	// Offset is the number of entries to skip.
	Offset int
	// Queue filters by queue name. Empty means all queues.
	Queue string
}

type Page struct {
	Entries    []*Entry
	NextCursor string
	Complete   bool
}

type PageLister interface {
	ListDLQPage(ctx context.Context, opts PageOpts) (Page, error)
	CountDLQEntries(ctx context.Context, opts CountOpts) (int64, error)
}

type PageOpts struct {
	Queue      string
	NamePrefix string
	ScopeAppID string
	ScopeOrgID string
	Replayed   *bool
	Cursor     string
	Limit      int
}

type ReplayClaimer interface {
	// ClaimReplay sets replayed_at = now and replayed_job_id = jobID only
	// if replayed_at is null. It returns dispatch.ErrDLQAlreadyReplayed
	// when replayed_at was already set, and dispatch.ErrDLQNotFound when
	// the entry does not exist. The check and the write are one atomic
	// step on every backend: of two concurrent claims, exactly one wins.
	ClaimReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

	// ReleaseReplay clears replayed_at and replayed_job_id, but only if
	// replayed_job_id still equals jobID, so a release never undoes
	// somebody else's claim. When it does not match, including on an
	// entry nobody claimed, it is a no-op and returns nil. It returns
	// dispatch.ErrDLQNotFound when the entry does not exist. Like the
	// claim, the comparison and the write are one atomic step.
	ReleaseReplay(ctx context.Context, entryID id.DLQID, jobID id.JobID) error

	// GetDLQByJobID returns the newest entry for a failed job, newest by
	// entry ID (the order ListDLQPage uses), or dispatch.ErrDLQNotFound
	// when the job has none.
	GetDLQByJobID(ctx context.Context, jobID id.JobID) (*Entry, error)

	// DeleteDLQ hard-deletes one entry. It returns dispatch.ErrDLQNotFound
	// when the entry does not exist.
	DeleteDLQ(ctx context.Context, entryID id.DLQID) error
}

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

type ServiceOption func(*Service)

type Store interface {
	ReplayClaimer

	// PushDLQ adds a failed job entry to the dead letter queue.
	PushDLQ(ctx context.Context, entry *Entry) error

	// ListDLQ returns DLQ entries matching the given options.
	ListDLQ(ctx context.Context, opts ListOpts) ([]*Entry, error)

	// GetDLQ retrieves a DLQ entry by ID.
	GetDLQ(ctx context.Context, entryID id.DLQID) (*Entry, error)

	// ReplayDLQ marks a DLQ entry as replayed. The actual re-enqueue is
	// handled at the service layer.
	ReplayDLQ(ctx context.Context, entryID id.DLQID) error

	// PurgeDLQ removes DLQ entries with FailedAt before the given time.
	// Returns the number of entries removed.
	PurgeDLQ(ctx context.Context, before time.Time) (int64, error)

	// CountDLQ returns the total number of entries in the dead letter queue.
	CountDLQ(ctx context.Context) (int64, error)
}