service

func (*CicadaBuiltinService) RunScript ¶

func (s *CicadaBuiltinService) RunScript(req model.RunScriptReq) (*model.ScriptResult, error)

func (*CicadaBuiltinService) SleepTime ¶

func (s *CicadaBuiltinService) SleepTime(req model.SleepTimeReq) (*model.SimpleMsg, error)

func (*ConnectionService) Create ¶

func (s *ConnectionService) Create(conn *model.Connection) (*model.Connection, error)

func (*ConnectionService) Delete ¶

func (s *ConnectionService) Delete(id string) error

func (*ConnectionService) Get ¶

func (s *ConnectionService) Get(id string) (*model.Connection, error)

func (*ConnectionService) List ¶

func (s *ConnectionService) List(page, row int, orderBy string) ([]model.Connection, error)

func (*ConnectionService) Update ¶

func (s *ConnectionService) Update(id string, conn *model.Connection) (*model.Connection, error)

func (*TaskComponentService) Create ¶

func (s *TaskComponentService) Create(req model.CreateTaskComponentReq) (*model.TaskComponent, error)

func (*TaskComponentService) Delete ¶

func (s *TaskComponentService) Delete(id string) error

func (*TaskComponentService) Get ¶

func (s *TaskComponentService) Get(id string) (*model.TaskComponent, error)

func (*TaskComponentService) GetByName ¶

func (s *TaskComponentService) GetByName(name string) (*model.TaskComponent, error)

func (*TaskComponentService) List ¶

func (s *TaskComponentService) List(page, row int) (*[]model.TaskComponent, error)

func (*TaskComponentService) Update ¶

func (s *TaskComponentService) Update(id string, req model.CreateTaskComponentReq) (*model.TaskComponent, error)

func (*TaskTypeService) Get ¶

func (s *TaskTypeService) Get(id string) (*catalog.TaskTypeDef, error)

Get returns the catalog entry for the given id, or an error if not found.

func (*TaskTypeService) List ¶

func (s *TaskTypeService) List() []catalog.TaskTypeDef

List returns all task type definitions in catalog file order.

func (*WorkflowRuntimeService) ClearTaskInstances ¶

func (s *WorkflowRuntimeService) ClearTaskInstances(wfID, wfRunID string, option model.TaskClearOption) ([]model.TaskInstanceReference, error)

func (*WorkflowRuntimeService) GetEventLogs ¶

func (s *WorkflowRuntimeService) GetEventLogs(wfID, wfRunID, taskID string) ([]model.EventLog, error)

func (*WorkflowRuntimeService) GetImportErrors ¶

func (s *WorkflowRuntimeService) GetImportErrors() (model.ImportErrors, error)

func (*WorkflowRuntimeService) GetTaskInstances ¶

func (s *WorkflowRuntimeService) GetTaskInstances(wfID, wfRunID string) ([]model.TaskInstance, error)

func (*WorkflowRuntimeService) GetTaskLogDownload ¶

func (s *WorkflowRuntimeService) GetTaskLogDownload(wfID, wfRunID, taskID string, taskTryNum int) (string, []byte, error)

func (*WorkflowRuntimeService) GetTaskLogs ¶

func (s *WorkflowRuntimeService) GetTaskLogs(wfID, wfRunID, taskID string, taskTryNum int) (*model.TaskLog, error)

func (*WorkflowRuntimeService) GetWorkflowRuns ¶

func (s *WorkflowRuntimeService) GetWorkflowRuns(wfID string) ([]model.WorkflowRun, error)

func (*WorkflowRuntimeService) GetWorkflowStatus ¶

func (s *WorkflowRuntimeService) GetWorkflowStatus(wfID string) ([]model.WorkflowStatus, error)

func (*WorkflowScheduleService) Cancel ¶

func (s *WorkflowScheduleService) Cancel(workflowID string) (*model.WorkflowSchedule, error)

Cancel marks the workflow's active schedule as canceled and rewrites the DAG metadata so the schedule line is dropped (DAG becomes manual-trigger only again). Since at most one active schedule per workflow is allowed, identifying the target by workflow id alone is unambiguous.

func (*WorkflowScheduleService) GetLatest ¶

func (s *WorkflowScheduleService) GetLatest(workflowID string) (*model.WorkflowSchedule, error)

GetLatest returns the workflow's most recently created schedule row regardless of status (active / executed / canceled), or (nil, nil) when there's no schedule history. Callers branch on .status to interpret the result.

func (*WorkflowScheduleService) Schedule ¶

func (s *WorkflowScheduleService) Schedule(workflowID string, req model.CreateWorkflowScheduleReq) (*model.WorkflowSchedule, error)

Schedule registers a scheduled execution. Exactly one of req.RunAt / req.Cron must be set:

• RunAt: one-shot future execution (Airflow schedule="@once" + start_date).
• Cron: recurring execution (Airflow schedule=<cron>).

Only one active schedule per workflow is allowed regardless of type; callers must Cancel the existing one before registering another. After persisting the row the DAG metadata is rewritten so Airflow picks up the new schedule on its next parse cycle.

func (*WorkflowService) CloneWorkflow ¶

func (s *WorkflowService) CloneWorkflow(srcWfID string) (*model.Workflow, error)

CloneWorkflow creates a brand-new Workflow by deep-copying an existing one looked up by srcWfID. The new workflow's name is auto-generated as "<source name>_copy" (with _copy_2, _copy_3 ... suffix on collision). task_groups / tasks are pulled from the source and all IDs are re-issued. Source's runs and snapshots are not copied. The new workflow's first snapshot records source_type="clone" and source_template_id=<source workflow ID> so the lineage is traceable.

func (*WorkflowService) CreateWorkflow ¶

func (s *WorkflowService) CreateWorkflow(req model.CreateWorkflowReq) (*model.Workflow, error)

func (*WorkflowService) DeleteWorkflow ¶

func (s *WorkflowService) DeleteWorkflow(wfId string) error

func (*WorkflowService) GetWorkflow ¶

func (s *WorkflowService) GetWorkflow(wfId string, includeDeleted bool) (*model.Workflow, error)

func (*WorkflowService) GetWorkflowByName ¶

func (s *WorkflowService) GetWorkflowByName(wfName string, includeDeleted bool) (*model.Workflow, error)

func (*WorkflowService) GetWorkflowVersion ¶

func (s *WorkflowService) GetWorkflowVersion(wfID, versionID string) (*model.WorkflowVersion, error)

func (*WorkflowService) ListWorkflow ¶

func (s *WorkflowService) ListWorkflow(name string, includeDeleted bool, page int, row int) (*[]model.Workflow, error)

func (*WorkflowService) ListWorkflowVersions ¶

func (s *WorkflowService) ListWorkflowVersions(wfID string, page, row int) (*[]model.WorkflowVersion, error)

func (*WorkflowService) RollbackWorkflow ¶

func (s *WorkflowService) RollbackWorkflow(wfID string, versionNo int) (*model.Workflow, error)

RollbackWorkflow restores the workflow's task graph + metadata from the snapshot identified by versionNo (1-based, scoped to the workflow). The rollback is recorded as a fresh WorkflowVersion entry with action="rollback" and source_template_id=<source version id>, so the lineage stays queryable.

The diff against the current workflow is computed via mapper.BuildWorkflowGraphDiff — same path UpdateWorkflow uses — so tasks that exist by name in both current and target keep their IDs (and Airflow task history continuity), tasks only in the target get fresh IDs, and tasks only in the current get soft-deleted with a "rollback_drop" snapshot for forensics. workflow_schedules rows are left untouched; schedule lifecycle is orthogonal to workflow definition.

Refuses to roll back to action="delete" snapshots (no meaningful state to restore). Deleted workflows are not supported in this first cut — restore them first via a future RestoreWorkflow path.

func (*WorkflowService) RunWorkflow ¶

func (s *WorkflowService) RunWorkflow(wfId string) error

func (*WorkflowService) UpdateWorkflow ¶

func (s *WorkflowService) UpdateWorkflow(wfId string, req model.CreateWorkflowReq) (*model.Workflow, error)

func (*WorkflowTaskService) GetTask ¶

func (s *WorkflowTaskService) GetTask(wfID string, taskID string, includeDeleted bool) (*model.Task, error)

func (*WorkflowTaskService) GetTaskDirectly ¶

func (s *WorkflowTaskService) GetTaskDirectly(taskID string, includeDeleted bool) (*model.TaskDirectly, error)

func (*WorkflowTaskService) GetTaskFromTaskGroup ¶

func (s *WorkflowTaskService) GetTaskFromTaskGroup(wfID string, tgID string, taskID string, includeDeleted bool) (*model.Task, error)

func (*WorkflowTaskService) GetTaskGroup ¶

func (s *WorkflowTaskService) GetTaskGroup(wfID string, tgID string, includeDeleted bool) (*model.TaskGroup, error)

func (*WorkflowTaskService) GetTaskGroupDirectly ¶

func (s *WorkflowTaskService) GetTaskGroupDirectly(tgID string, includeDeleted bool) (*model.TaskGroupDirectly, error)

func (*WorkflowTaskService) ListTask ¶

func (s *WorkflowTaskService) ListTask(wfID string, includeDeleted bool) ([]model.Task, error)

func (*WorkflowTaskService) ListTaskFromTaskGroup ¶

func (s *WorkflowTaskService) ListTaskFromTaskGroup(wfID string, tgID string, includeDeleted bool) ([]model.Task, error)

func (*WorkflowTaskService) ListTaskGroup ¶

func (s *WorkflowTaskService) ListTaskGroup(wfID string, includeDeleted bool) ([]model.TaskGroup, error)

func (*WorkflowTemplateService) Get ¶

func (s *WorkflowTemplateService) Get(id string) (*model.GetWorkflowTemplate, error)

func (*WorkflowTemplateService) GetByName ¶

func (s *WorkflowTemplateService) GetByName(name string) (*model.GetWorkflowTemplate, error)

func (*WorkflowTemplateService) List ¶

func (s *WorkflowTemplateService) List(name string, page, row int) (*[]model.WorkflowTemplate, error)