workflow

type BatchBuilder interface {
	// Name sets a display name for the batch.
	Name(name string) BatchBuilder
	// OnQueue applies a default queue to batch jobs that do not set one.
	OnQueue(queue string) BatchBuilder
	// AllowFailures keeps the batch running when individual jobs fail.
	AllowFailures() BatchBuilder
	// Progress registers a callback invoked as jobs complete.
	Progress(fn func(ctx context.Context, st BatchState) error) BatchBuilder
	// Then registers a callback invoked once when batch succeeds.
	Then(fn func(ctx context.Context, st BatchState) error) BatchBuilder
	// Catch registers a callback invoked when batch encounters a failure.
	Catch(fn func(ctx context.Context, st BatchState, err error) error) BatchBuilder
	// Finally registers a callback invoked once when batch reaches terminal state.
	Finally(fn func(ctx context.Context, st BatchState) error) BatchBuilder
	// Dispatch creates and starts the batch workflow.
	Dispatch(ctx context.Context) (string, error)
}

type BatchJob struct {
	JobID string
	Job   StoredJob
}

type BatchJobOutcome string

type BatchRecord struct {
	BatchID     string
	DispatchID  string
	Name        string
	Queue       string
	AllowFailed bool
	Jobs        []BatchJob
	CreatedAt   time.Time
	// contains filtered or unexported fields
}

type BatchState struct {
	BatchID     string
	DispatchID  string
	Name        string
	Queue       string
	AllowFailed bool
	Total       int
	Pending     int
	Processed   int
	Failed      int
	Cancelled   bool
	Completed   bool
	CreatedAt   time.Time
	UpdatedAt   time.Time
}

type ChainBuilder interface {
	// OnQueue applies a default queue to chain jobs that do not set one.
	OnQueue(queue string) ChainBuilder
	// Catch registers a callback invoked when chain execution fails.
	Catch(fn func(ctx context.Context, st ChainState, err error) error) ChainBuilder
	// Finally registers a callback invoked once when chain execution finishes.
	Finally(fn func(ctx context.Context, st ChainState) error) ChainBuilder
	// Dispatch creates and starts the chain workflow.
	Dispatch(ctx context.Context) (string, error)
}

type ChainNode struct {
	NodeID string
	Job    StoredJob
}

type ChainRecord struct {
	ChainID    string
	DispatchID string
	Queue      string
	Nodes      []ChainNode
	CreatedAt  time.Time
	// contains filtered or unexported fields
}

type ChainState struct {
	ChainID    string
	DispatchID string
	Queue      string
	Nodes      []ChainNode
	NextIndex  int
	Completed  bool
	Failed     bool
	Failure    string
	CreatedAt  time.Time
	UpdatedAt  time.Time
}

type Context struct {
	SchemaVersion int
	DispatchID    string
	JobID         string
	ChainID       string
	BatchID       string
	Attempt       int
	JobType       string
	// contains filtered or unexported fields
}

type ContinuationBinder interface {
	BindContinuations([]ContinuationDefinition) (any, error)
}

type ContinuationClaim = continuationClaim

type ContinuationDefinition struct {
	Phase string
	Job   StoredJob
}

type ContinuationReader interface {
	FindContinuations(context.Context, string) ([]busruntime.ContinuationOutcome, error)
}

type ContinuationStore interface {
	PrepareContinuations(context.Context) error
	ContinuationOutcomes(context.Context, string) ([]busruntime.ContinuationOutcome, error)
	ClaimContinuation(context.Context, string, string, time.Time, time.Duration) (continuationClaim, bool, error)
	ValidateContinuation(context.Context, continuationClaim) (bool, error)
	SettleContinuation(context.Context, continuationClaim, string, time.Time) error
	BeginContinuation(context.Context, continuationClaim, string, time.Time) (continuationClaim, bool, error)
	RenewContinuation(context.Context, continuationClaim, time.Time) error
}

type CoordinatorLifecycle interface {
	StartCoordinator(context.Context) error
	StopCoordinator(context.Context) error
	CoordinatorDone() <-chan struct{}
	CoordinatorError() error
}

type DispatchResult struct {
	DispatchID string
}

type Engine interface {
	// Chain creates a sequential workflow builder.
	Chain(jobs ...Job) ChainBuilder
	// Batch creates an aggregate workflow builder.
	Batch(jobs ...Job) BatchBuilder
	// FindBatch returns persisted aggregate workflow state.
	FindBatch(ctx context.Context, batchID string) (BatchState, error)
	// FindChain returns persisted sequential workflow state.
	FindChain(ctx context.Context, chainID string) (ChainState, error)
	// Prune removes terminal workflow state older than before.
	Prune(ctx context.Context, before time.Time) error
}

type Event struct {
	SchemaVersion int
	EventID       string
	Kind          EventKind
	DispatchID    string
	JobID         string
	ChainID       string
	BatchID       string
	Attempt       int
	JobType       string
	JobKey        string
	Queue         string
	Duration      time.Duration
	Time          time.Time
	Err           error
}

type EventKind string

type Handler func(ctx context.Context, j Context) error

type Job struct {
	Type    string
	Payload any
	Options JobOptions
}

type JobConfig struct {
	TranslateError func(error) error
	PanicError     func(any) error
	Clock          func() time.Time
	Observer       Observer
	Middlewares    []Middleware
}

type JobOptions struct {
	Queue     string
	Delay     time.Duration
	Timeout   time.Duration
	Retry     int
	Backoff   time.Duration
	UniqueFor time.Duration
}

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

type Middleware interface {
	// Handle wraps the remaining middleware and handler chain.
	Handle(ctx context.Context, jc Context, next Next) error
}

type Next func(ctx context.Context, jc Context) error

type Observer interface {
	// Observe consumes one best-effort workflow fact.
	Observe(ctx context.Context, event Event)
}

type ObserverFunc func(ctx context.Context, event Event)

type Option func(*runtime)

type ResolvedDelivery struct {
	// JobType is the logical application type when a supported envelope provides one, otherwise the physical type.
	JobType string
	// Payload is the logical application payload when a supported envelope provides one, otherwise the physical payload.
	Payload []byte
	// DispatchID correlates deliveries created by the same application dispatch.
	DispatchID string
	// JobID identifies the logical workflow job represented by this delivery.
	JobID string
	// ChainID identifies the owning chain when the delivery belongs to one.
	ChainID string
	// BatchID identifies the owning batch when the delivery belongs to one.
	BatchID string
}

type Runtime interface {
	Engine
	// Register binds a logical job type to its handler.
	Register(jobType string, handler Handler)
	// Dispatch submits a legacy envelope delivery.
	Dispatch(ctx context.Context, job Job) (DispatchResult, error)
	// DispatchDirect submits a canonical ordinary delivery.
	DispatchDirect(ctx context.Context, job StoredJob) (DispatchResult, error)
	// StartWorkers starts the selected physical runtime.
	StartWorkers(ctx context.Context) error
	// Shutdown drains the selected physical runtime.
	Shutdown(ctx context.Context) error
}

type SQLStoreConfig struct {
	DB         *sql.DB
	DriverName string
	DSN        string
	// AutoMigrate is retained for compatibility with the established config
	// shape. NewSQLStore keeps startup schema creation enabled by default; use
	// NewSQLStoreWithManagedSchema when deployment tooling owns the schema.
	AutoMigrate bool
}

type Store interface {
	// CreateChain persists a newly accepted chain.
	CreateChain(ctx context.Context, rec ChainRecord) error
	// AdvanceChain atomically claims completedNode and returns the current successor.
	// Repeating the same (chainID, completedNode) claim must not advance again.
	// When done is true, GetChain must immediately expose Completed or Failed state.
	AdvanceChain(ctx context.Context, chainID string, completedNode string) (next *ChainNode, done bool, err error)
	// FailChain commits terminal failure without replacing completed state.
	FailChain(ctx context.Context, chainID string, cause error) error
	// GetChain returns current chain state.
	GetChain(ctx context.Context, chainID string) (ChainState, error)

	// CreateBatch persists a newly accepted batch.
	CreateBatch(ctx context.Context, rec BatchRecord) error
	// MarkBatchJobStarted records that one batch member began execution.
	MarkBatchJobStarted(ctx context.Context, batchID, jobID string) error
	// MarkBatchJobSucceeded commits the first outcome for (batchID, jobID).
	// Duplicate outcomes must return current state without changing counters.
	MarkBatchJobSucceeded(ctx context.Context, batchID, jobID string) (BatchState, bool, error)
	// MarkBatchJobFailed commits the first outcome for (batchID, jobID).
	// Duplicate outcomes must return current state without changing counters.
	MarkBatchJobFailed(ctx context.Context, batchID, jobID string, cause error) (BatchState, bool, error)
	// CancelBatch commits aggregate batch cancellation.
	CancelBatch(ctx context.Context, batchID string) error
	// GetBatch returns current batch state.
	GetBatch(ctx context.Context, batchID string) (BatchState, error)

	// MarkCallbackInvoked atomically claims one callback idempotency key.
	MarkCallbackInvoked(ctx context.Context, key string) (bool, error)
	// Prune removes terminal workflow state older than before.
	Prune(ctx context.Context, before time.Time) error
}

type StoredJob struct {
	Type    string     `json:"type"`
	Payload []byte     `json:"payload"`
	Options JobOptions `json:"options"`
}