retrypool

type BlockingConfig[T any] struct {

	// Task callbacks
	OnTaskSubmitted func(task T)
	OnTaskStarted   func(task T)
	OnTaskCompleted func(task T)
	OnTaskFailed    func(task T, err error)
	// Group/Pool callbacks
	OnGroupCreated   func(groupID any)
	OnGroupCompleted func(groupID any, tasks []T)
	OnGroupFailed    func(groupID any, tasks []T)
	OnGroupRemoved   func(groupID any, tasks []T)
	OnPoolCreated    func(groupID any)
	OnPoolDestroyed  func(groupID any)
	OnWorkerAdded    func(groupID any, workerID int)
	OnWorkerRemoved  func(groupID any, workerID int)
	OnPoolClosed     func()
	// contains filtered or unexported fields
}

type BlockingDependentTask[GID comparable, TID comparable] interface {
	GetGroupID() GID
	GetTaskID() TID
}

type BlockingMetricsSnapshot[T any, GID comparable] struct {
	TotalTasksSubmitted int64
	TotalTasksProcessed int64
	TotalTasksSucceeded int64
	TotalTasksFailed    int64
	TotalDeadTasks      int64
	Metrics             []GroupBlockingMetricSnapshot[T, GID]
}

type BlockingPool[T any, GID comparable, TID comparable] struct {
	// contains filtered or unexported fields
}

type BlockingPoolOption[T any] func(*BlockingConfig[T])

type BlockingRequestResponse[T any, R any, GID comparable, TID comparable] struct {
	// contains filtered or unexported fields
}

type CircularQueue[T any] struct {
	// contains filtered or unexported fields
}

type Config[T any] struct {
	// contains filtered or unexported fields
}

type DeadTask[T any] struct {
	Data          T
	Retries       int
	TotalDuration time.Duration
	Errors        []error
	Reason        string
	StateHistory  []*TaskStateTransition[T]
}

type DelayFunc[T any] func(retries int, err error, config *Config[T]) time.Duration

type DependencyGraph[TID comparable] struct {
	Nodes map[TID]*Node[TID]
	Order []TID // Topologically sorted execution order
}

type DependencyMode int

type ExponentialBackoffRetryPolicy[T any] struct {
	BaseDelay time.Duration
	MaxDelay  time.Duration
	MaxJitter time.Duration
}

type FixedDelayRetryPolicy[T any] struct {
	Delay     time.Duration
	MaxJitter time.Duration
}

type GroupBlockingMetricSnapshot[T any, GID comparable] struct {
	GroupID GID
	MetricsSnapshot[T]
}

type GroupMetricSnapshot[T any, GID comparable] struct {
	GroupID         GID
	PoolID          uint
	HasPool         bool
	IsPending       bool
	TasksPending    int
	MetricsSnapshot MetricsSnapshot[T]
}

type GroupMetricsSnapshot[T any, GID comparable] struct {
	// These totals are GLOBAL and never reset
	TotalTasksSubmitted int64
	TotalTasksProcessed int64
	TotalTasksSucceeded int64
	TotalTasksFailed    int64
	TotalDeadTasks      int64

	TotalPools        int
	ActivePools       int
	TotalPendingTasks int
	GroupsWithPending int

	Pools        []PoolMetricsSnapshot[T]
	PendingTasks []PendingTasksSnapshot[GID]
	Metrics      []GroupMetricSnapshot[T, GID]
}

type GroupPool[T GroupTask[GID], GID comparable] struct {
	// contains filtered or unexported fields
}

type GroupPoolConfig[T GroupTask[GID], GID comparable] struct {
	Logger            Logger
	WorkerFactory     WorkerFactory[T]
	MaxActivePools    int
	MaxWorkersPerPool int
	UseFreeWorkerOnly bool
	GroupMustSucceed  bool // By default a group can have deadtask/failure, if true the pending tasks will be discarded

	OnTaskSuccess func(gid GID, poolID uint, data T, metadata map[string]any)
	OnTaskFailure func(gid GID, poolID uint, data T, metadata map[string]any, err error) TaskAction

	// GroupPool specific callback handled either OnTaskSuccess or OnTaskFailure
	OnTaskExecuted func(gid GID, poolID uint, data T, metadata map[string]any, err error)

	OnSnapshot func(snapshot GroupMetricsSnapshot[T, GID])

	OnGroupStarts func(gid GID, poolID uint)
	OnGroupEnds   func(gid GID, poolID uint)
	OnGroupFails  func(gid GID, poolID uint, pendingTasks []T) // Only if GroupMustSucceed is enabled
	// contains filtered or unexported fields
}

type GroupPoolOption[T GroupTask[GID], GID comparable] func(*GroupPoolConfig[T, GID])

type GroupRequestResponse[T any, R any, GID comparable] struct {
	// contains filtered or unexported fields
}

type GroupTask[GID comparable] interface {
	GetGroupID() GID
}

type GroupTaskOption[T any, GID comparable] func(*groupSubmitConfig)

type GrowingCircularQueue[T any] struct {
	// contains filtered or unexported fields
}

type GrowingRingBufferQueue[T any] struct {
	// contains filtered or unexported fields
}

type IndependentConfig[T any] struct {

	// Task callbacks
	OnTaskSubmitted func(task T)
	OnTaskStarted   func(task T)
	OnTaskCompleted func(task T)
	OnTaskFailed    func(task T, err error)
	// Group callbacks
	OnGroupCreated   func(groupID any)
	OnGroupCompleted func(groupID any)
	OnGroupRemoved   func(groupID any, tasks []T)
	// Pool events
	OnWorkerAdded   func(workerID int)
	OnWorkerRemoved func(workerID int)
	OnPoolClosed    func()
	// contains filtered or unexported fields
}

type IndependentDependentTask[GID comparable, TID comparable] interface {
	GetDependencies() []TID
	GetGroupID() GID
	GetTaskID() TID
}

type IndependentPool[T any, GID comparable, TID comparable] struct {
	// contains filtered or unexported fields
}

type IndependentPoolOption[T any] func(*IndependentConfig[T])

type LogFormat string

type LogLevel = slog.Level

type LogOption func(*loggerConfig)

type Logger interface {
	Debug(ctx context.Context, msg string, keysAndValues ...any)
	Info(ctx context.Context, msg string, keysAndValues ...any)
	Warn(ctx context.Context, msg string, keysAndValues ...any)
	Error(ctx context.Context, msg string, keysAndValues ...any)
	WithFields(fields map[string]any) Logger
	Enable()
	Disable()
}

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

type Metrics struct {
	TasksSubmitted atomic.Int64
	TasksProcessed atomic.Int64
	TasksSucceeded atomic.Int64
	TasksFailed    atomic.Int64
	DeadTasks      atomic.Int64
}

type MetricsSnapshot[T any] struct {
	TasksSubmitted int64
	TasksProcessed int64
	TasksSucceeded int64
	TasksFailed    int64
	DeadTasks      int64
	TaskQueues     map[int]int
	Workers        map[int]WorkerSnapshot[T]
}

type NoWorkerPolicy int

type Node[TID comparable] struct {
	ID           TID
	Dependencies []TID
	Dependents   []TID
	Visited      bool
	InProgress   bool // Used for cycle detection
}

type Option[T any] func(*Pool[T])

type PendingTasksSnapshot[GID comparable] struct {
	GroupID      GID
	TasksPending int
}

type Pool[T any] struct {
	// contains filtered or unexported fields
}

type PoolMetricsSnapshot[T any] struct {
	PoolID        uint
	AssignedGroup any
	IsActive      bool
	QueueSize     int64
	Workers       map[int]WorkerSnapshot[T]
}

type Pooler[T any] interface {
	// SetOnTaskSuccess sets the handler that will be called when a task succeeds.
	SetOnTaskSuccess(handler func(data T, metadata map[string]any))

	// SetOnTaskFailure sets the handler that will be called when a task fails.
	// The handler should return a TaskAction indicating how the pool should proceed.
	SetOnTaskFailure(handler func(data T, metadata map[string]any, err error) TaskAction)

	SetOnTaskAttempt(handler func(task *Task[T], workerID int))

	// Add adds a new worker to the pool. If a queue is not provided, a new one will be created.
	Add(worker Worker[T], queue TaskQueue[T]) error

	// Remove removes a worker by its ID, redistributing its tasks to other workers.
	Remove(id int) error

	// Pause pauses a worker by its ID, redistributing its tasks and preventing it from processing further tasks.
	Pause(id int) error

	// Resume resumes a previously paused worker, allowing it to process tasks again.
	Resume(id int) error

	// Workers returns the list of worker IDs currently managed by the pool.
	Workers() ([]int, error)

	// GetFreeWorkers returns a list of worker IDs that have no tasks in their queue
	GetFreeWorkers() []int

	// SubmitToFreeWorker attempts to submit a task to a free worker
	SubmitToFreeWorker(taskData T, options ...TaskOption[T]) error

	// Submit allows submitting data directly as a task without pre-allocation. Optional TaskOptions can modify the task's behavior.
	Submit(data T, options ...TaskOption[T]) error

	// WaitWithCallback waits until the provided callback returns false, periodically invoking it at the given interval.
	// The callback receives the current queue size, processing count, and dead task count, and should return false to stop waiting.
	WaitWithCallback(ctx context.Context, callback func(queueSize, processingCount, deadTaskCount int) bool, interval time.Duration) error

	// Close gracefully shuts down the pool, stopping all workers and redistributing or discarding tasks as needed.
	Close() error

	// QueueSize returns the total number of tasks currently queued.
	QueueSize() int64

	// ProcessingCount returns the number of tasks currently being processed by workers.
	ProcessingCount() int64

	// DeadTaskCount returns the number of dead tasks that have permanently failed.
	DeadTaskCount() int64

	// RangeDeadTasks iterates over all dead tasks. If the callback returns false, iteration stops.
	RangeDeadTasks(fn func(*DeadTask[T]) bool)

	// PullDeadTask removes and returns a dead task at the specified index.
	PullDeadTask(idx int) (*DeadTask[T], error)

	// PullRangeDeadTasks removes and returns a range of dead tasks [from, to).
	PullRangeDeadTasks(from int, to int) ([]*DeadTask[T], error)

	// RangeWorkerQueues iterates over each worker's queue size. If the callback returns false, iteration stops.
	RangeWorkerQueues(f func(workerID int, queueSize int64) bool)

	// RangeTaskQueues iterates over each worker's TaskQueue. If the callback returns false, iteration stops.
	RangeTaskQueues(f func(workerID int, queue TaskQueue[T]) bool)

	// RangeWorkers iterates over each worker. If the callback returns false, iteration stops.
	RangeWorkers(f func(workerID int, state WorkerSnapshot[T]) bool)
}

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

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

type RequestResponse[T any, R any] struct {
	// contains filtered or unexported fields
}

type RetryPolicy[T any] interface {
	ComputeDelay(retries int, err error, config *Config[T]) time.Duration
}

type RingBufferQueue[T any] struct {
	// contains filtered or unexported fields
}

type SubmitOption[T any] func(*submitConfig)

type Task[T any] struct {
	// contains filtered or unexported fields
}

type TaskAction int

type TaskMetadataSetter func(metadata map[string]any)

type TaskOption[T any] func(*Task[T])

type TaskQueue[T any] interface {
	Enqueue(task *Task[T])
	PriorityEnqueue(task *Task[T]) error // For immediate retry
	Dequeue() (*Task[T], bool)
	Length() int
	Clear()
	Drain() []*Task[T]
}

type TaskQueueType int

type TaskState int

type TaskStateTransition[T any] struct {
	FromState TaskState
	ToState   TaskState
	Task      *Task[T]
	Reason    string
	Timestamp time.Time
}

type Worker[T any] interface {
	Run(ctx context.Context, data T) error
}

type WorkerFactory[T any] func() Worker[T]

type WorkerSnapshot[T any] struct {
	ID       int
	Paused   bool
	Removed  bool
	HasTask  bool
	Metadata map[string]any
	State    WorkerState
}

type WorkerState string