taskservice

type ActiveRoutine interface {
	// Resume resumes the go routine of the daemon task.
	Resume() error
	// Pause pauses the go routine of the daemon task.
	Pause() error
	// Cancel cancels the go routine of the daemon task.
	Cancel() error
	// Restart restart the go routine of the daemon task.
	Restart() error
}

type CDCTaskKey struct {
	AccountId uint64
	TaskId    string
}

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

type Condition func(*conditions)

type Getter func() (TaskService, bool)

type Op int

type PauseTaskCompletedHook func(context.Context, task.DaemonTask) error

type RunnerOption func(*taskRunner)

type SQLTask struct {
	TaskID         uint64
	TaskName       string
	AccountID      uint32
	DatabaseName   string
	CronExpr       string
	Timezone       string
	SQLBody        string
	GateCondition  string
	RetryLimit     int
	TimeoutSeconds int
	Enabled        bool
	NextFireTime   int64
	TriggerCount   uint64
	Creator        string
	CreatorUserID  uint32
	CreatorRoleID  uint32
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

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

type SQLTaskRun struct {
	RunID           uint64
	TaskID          uint64
	TaskName        string
	AccountID       uint32
	ScheduledAt     time.Time
	StartedAt       time.Time
	FinishedAt      time.Time
	DurationSeconds float64
	Status          string
	TriggerType     string
	AttemptNumber   int
	RowsAffected    int64
	ErrorCode       int
	ErrorMessage    string
	GateResult      bool
	RunnerCN        string
}

type SqlExecutor interface {
	PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
	ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
}

type TaskExecutor func(ctx context.Context, task task.Task) error

type TaskHandler interface {
	Handle(ctx context.Context) error
}

type TaskRunner interface {
	// ID returns the TaskRunner ID
	ID() string
	// Start start the runner, after runner starts it will start to periodically load the tasks assigned to
	// the current executor, as well as periodically send heartbeats.
	Start() error
	// Stop stop the runner, all running tasks will be terminated
	Stop() error
	// Parallelism maximum number of concurrently executing Tasks
	Parallelism() int
	// RegisterExecutor register the task executor
	RegisterExecutor(code task.TaskCode, executor TaskExecutor)
	// GetExecutor returns the task executor
	GetExecutor(code task.TaskCode) TaskExecutor
	// Attach attaches the active go-routine to the daemon task.
	Attach(ctx context.Context, taskID uint64, routine ActiveRoutine) error
}

type TaskService interface {
	// Close close the task service
	Close() error

	// CreateAsyncTask Creates an asynchronous task that executes a single time, this method is idempotent, the
	// same task is not created repeatedly based on multiple calls.
	CreateAsyncTask(context.Context, task.TaskMetadata) error
	// CreateBatch is similar to Create, but with a batch task list
	CreateBatch(context.Context, []task.TaskMetadata) error
	// CreateCronTask is similar to Create, but create a task that runs periodically, with the period
	// described using a Cron expression.
	CreateCronTask(ctx context.Context, task task.TaskMetadata, cronExpr string) error
	// Allocate allocate task runner fot spec task.
	Allocate(ctx context.Context, value task.AsyncTask, taskRunner string) error
	// Complete task completed. The result used to indicate whether the execution succeeded or failed
	Complete(ctx context.Context, taskRunner string, task task.AsyncTask, result task.ExecuteResult) error
	// Heartbeat sending a heartbeat tells the scheduler that the specified task is running normally.
	// If the scheduler does not receive the heartbeat for a long time, it will reassign the task executor
	// to execute the task. Returning `ErrInvalidTask` means that the Task has been reassigned or has
	// ended, and the Task execution needs to be terminated immediately。
	Heartbeat(ctx context.Context, task task.AsyncTask) error
	// QueryAsyncTask query tasks by conditions
	QueryAsyncTask(context.Context, ...Condition) ([]task.AsyncTask, error)
	// QueryCronTask query cron tasks by conditions
	QueryCronTask(context.Context, ...Condition) ([]task.CronTask, error)

	// CreateDaemonTask creates a daemon task that will run in background for long time.
	CreateDaemonTask(ctx context.Context, value task.TaskMetadata, details *task.Details) error
	// QueryDaemonTask returns all daemon tasks which match the conditions.
	QueryDaemonTask(ctx context.Context, conds ...Condition) ([]task.DaemonTask, error)
	// UpdateDaemonTask updates the daemon task record.
	UpdateDaemonTask(ctx context.Context, tasks []task.DaemonTask, cond ...Condition) (int, error)
	// HeartbeatDaemonTask sends heartbeat to daemon task.
	HeartbeatDaemonTask(ctx context.Context, task task.DaemonTask) error

	// StartScheduleCronTask start schedule cron tasks. A timer will be started to pull the latest CronTask
	// from the TaskStore at regular intervals, and a timer will be maintained in memory for all Cron's to be
	// triggered at regular intervals.
	StartScheduleCronTask()
	// StopScheduleCronTask stop schedule cron tasks.
	StopScheduleCronTask()
	// StartScheduleSQLTask starts SQL task scheduling.
	StartScheduleSQLTask()
	// StopScheduleSQLTask stops SQL task scheduling.
	StopScheduleSQLTask()

	TruncateCompletedTasks(ctx context.Context) error

	// GetStorage returns the task storage
	GetStorage() TaskStorage

	// AddCDCTask Update cdc task in one transaction
	AddCDCTask(
		context.Context,
		task.TaskMetadata,
		*task.Details,
		func(context.Context, SqlExecutor) (int, error),
	) (int, error)

	// UpdateCDCTask Update cdc task in one transaction
	UpdateCDCTask(
		context.Context,
		task.TaskStatus,
		func(
			context.Context,
			task.TaskStatus,
			map[CDCTaskKey]struct{},
			SqlExecutor,
		) (int, error),
		...Condition,
	) (int, error)
}

type TaskServiceHolder interface {
	// Close close the holder
	Close() error
	// Get returns the taskservice
	Get() (TaskService, bool)
	// Create create the taskservice
	Create(command logservicepb.CreateTaskService) error
}

type TaskStorage interface {
	// Close close the task storage
	Close() error
	// PingContext ping the db with context
	PingContext(context.Context) error

	// AddAsyncTask adds async tasks and returns number of successful added
	AddAsyncTask(context.Context, ...task.AsyncTask) (int, error)
	// UpdateAsyncTask updates async tasks and returns number of successful updated
	UpdateAsyncTask(context.Context, []task.AsyncTask, ...Condition) (int, error)
	// DeleteAsyncTask deletes tasks and returns number of successful deleted
	DeleteAsyncTask(context.Context, ...Condition) (int, error)
	// QueryAsyncTask queries tasks by conditions
	QueryAsyncTask(context.Context, ...Condition) ([]task.AsyncTask, error)

	// AddCronTask add cron task and returns number of successful added
	AddCronTask(context.Context, ...task.CronTask) (int, error)
	// QueryCronTask query all cron tasks
	QueryCronTask(context.Context, ...Condition) ([]task.CronTask, error)
	// UpdateCronTask crontask generates tasks periodically, and this update
	// needs to be in a transaction. Update cron task and insert a new task.
	// This update must be transactional and needs to be done conditionally
	// using CronTask.TriggerTimes and the task.Metadata.ID field.
	UpdateCronTask(context.Context, task.CronTask, task.AsyncTask) (int, error)

	// AddDaemonTask adds daemon tasks and returns number of successful added.
	AddDaemonTask(ctx context.Context, tasks ...task.DaemonTask) (int, error)
	// UpdateDaemonTask updates daemon tasks and returns number of successful updated.
	UpdateDaemonTask(ctx context.Context, tasks []task.DaemonTask, conds ...Condition) (int, error)
	// DeleteDaemonTask deletes daemon tasks and returns number of successful deleted.
	DeleteDaemonTask(ctx context.Context, condition ...Condition) (int, error)
	// QueryDaemonTask queries daemon tasks by conditions.
	QueryDaemonTask(ctx context.Context, condition ...Condition) ([]task.DaemonTask, error)
	// HeartbeatDaemonTask update the last heartbeat field of the task.
	HeartbeatDaemonTask(ctx context.Context, task []task.DaemonTask) (int, error)
	// AddCDCTask insert cdcTask and daemonTask
	AddCDCTask(context.Context, task.DaemonTask, func(context.Context, SqlExecutor) (int, error)) (int, error)
	// UpdateCDCTask Update cdc task in one transaction
	UpdateCDCTask(
		context.Context,
		task.TaskStatus,
		func(
			context.Context,
			task.TaskStatus,
			map[CDCTaskKey]struct{},
			SqlExecutor,
		) (int, error),
		...Condition,
	) (int, error)

	// AddSQLTask adds sql tasks and returns number of successful added.
	AddSQLTask(ctx context.Context, tasks ...SQLTask) (int, error)
	// UpdateSQLTask updates sql tasks and returns number of successful updated.
	UpdateSQLTask(ctx context.Context, tasks []SQLTask, conds ...Condition) (int, error)
	// DeleteSQLTask deletes sql tasks and returns number of successful deleted.
	DeleteSQLTask(ctx context.Context, conds ...Condition) (int, error)
	// QuerySQLTask queries sql tasks by conditions.
	QuerySQLTask(ctx context.Context, conds ...Condition) ([]SQLTask, error)

	// AddSQLTaskRun adds sql task run records.
	AddSQLTaskRun(ctx context.Context, runs ...SQLTaskRun) (int, error)
	// UpdateSQLTaskRun updates sql task run records.
	UpdateSQLTaskRun(ctx context.Context, runs []SQLTaskRun, conds ...Condition) (int, error)
	// QuerySQLTaskRun queries sql task run records.
	QuerySQLTaskRun(ctx context.Context, conds ...Condition) ([]SQLTaskRun, error)
	// QueryLatestSQLTaskRun returns the newest run for each task in taskIDs under the given account.
	QueryLatestSQLTaskRun(ctx context.Context, accountID uint32, taskIDs []uint64) ([]SQLTaskRun, error)
	// AcquireSQLTaskRun atomically acquires execution ownership and creates a RUNNING run record.
	AcquireSQLTaskRun(ctx context.Context, sqlTask SQLTask, run SQLTaskRun) (uint64, error)
	// CompleteSQLTaskRun updates the terminal state of a sql task run.
	CompleteSQLTaskRun(ctx context.Context, run SQLTaskRun) (int, error)

	// TriggerSQLTask updates persisted trigger state and inserts an async task in one transaction.
	TriggerSQLTask(ctx context.Context, sqlTask SQLTask, asyncTask task.AsyncTask) (int, error)
}

type TaskStorageFactory interface {
	Create(address string) (TaskStorage, error)
}