queue

type DBTask struct {
	DirectOwner *ent.User
	Task        *ent.Task
	// contains filtered or unexported fields
}

type Dep interface {
	ForkWithLogger(ctx context.Context, l logging.Logger) context.Context
}

type InMemoryTask struct {
	*DBTask
}

type Metric interface {
	IncBusyWorker()
	DecBusyWorker()
	BusyWorkers() uint64
	SuccessTasks() uint64
	FailureTasks() uint64
	SubmittedTasks() uint64
	IncSuccessTask()
	IncFailureTask()
	IncSubmittedTask()
}

type Option interface {
	// contains filtered or unexported methods
}

type OptionFunc func(*options)

type Progress struct {
	Total      int64  `json:"total"`
	Current    int64  `json:"current"`
	Identifier string `json:"identifier"`
}

type Progresses map[string]*Progress

type Queue interface {
	// Start resume tasks and starts all workers.
	Start()
	// Shutdown stops all workers.
	Shutdown()
	// SubmitTask submits a Task to the queue.
	QueueTask(ctx context.Context, t Task) error
	// BusyWorkers returns the numbers of workers in the running process.
	BusyWorkers() int
	// BusyWorkers returns the numbers of success tasks.
	SuccessTasks() int
	// FailureTasks returns the numbers of failure tasks.
	FailureTasks() int
	// SubmittedTasks returns the numbers of submitted tasks.
	SubmittedTasks() int
	// SuspendingTasks returns the numbers of suspending tasks.
	SuspendingTasks() int
}

type ResumableTaskFactory func(model *ent.Task) Task

type Scheduler interface {
	// Queue add a new Task into the queue
	Queue(task Task) error
	// Request get a new Task from the queue
	Request() (Task, error)
	// Shutdown stop all worker
	Shutdown() error
}

type Summary struct {
	NodeID int            `json:"-"`
	Phase  string         `json:"phase,omitempty"`
	Props  map[string]any `json:"props,omitempty"`
}

type Task interface {
	Do(ctx context.Context) (task.Status, error)

	// ID returns the Task ID
	ID() int
	// Type returns the Task type
	Type() string
	// Status returns the Task status
	Status() task.Status
	// Owner returns the Task owner
	Owner() *ent.User
	// State returns the internal Task state
	State() string
	// ShouldPersist returns true if the Task should be persisted into DB
	ShouldPersist() bool
	// Persisted returns true if the Task is persisted in DB
	Persisted() bool
	// Executed returns the duration of the Task execution
	Executed() time.Duration
	// Retried returns the number of times the Task has been retried
	Retried() int
	// Error returns the error of the Task
	Error() error
	// ErrorHistory returns the error history of the Task
	ErrorHistory() []error
	// Model returns the ent model of the Task
	Model() *ent.Task
	// CorrelationID returns the correlation ID of the Task
	CorrelationID() uuid.UUID
	// ResumeTime returns the time when the Task is resumed
	ResumeTime() int64
	// ResumeAfter sets the time when the Task should be resumed
	ResumeAfter(next time.Duration)
	Progress(ctx context.Context) Progresses
	// Summarize returns the Task summary for UI display
	Summarize(hasher hashid.Encoder) *Summary
	// OnSuspend is called when queue decides to suspend the Task
	OnSuspend(time int64)
	// OnPersisted is called when the Task is persisted or updated in DB
	OnPersisted(task *ent.Task)
	// OnError is called when the Task encounters an error
	OnError(err error, d time.Duration)
	// OnRetry is called when the iteration returns error and before retry
	OnRetry(err error)
	// OnIterationComplete is called when the one iteration is completed
	OnIterationComplete(executed time.Duration)
	// OnStatusTransition is called when the Task status is changed
	OnStatusTransition(newStatus task.Status)

	// Cleanup is called when the Task is done or error.
	Cleanup(ctx context.Context) error

	Lock()
	Unlock()
}

type TaskRegistry interface {
	// NextID returns the next available Task ID.
	NextID() int
	// Get returns the Task by ID.
	Get(id int) (Task, bool)
	// Set sets the Task by ID.
	Set(id int, t Task)
	// Delete deletes the Task by ID.
	Delete(id int)
}