pipeline

type AnyTask struct {
	BaseTask `mapstructure:",squash"`
}

type BaseTask struct {
	Index   int32         `mapstructure:"index" json:"-" `
	Timeout time.Duration `mapstructure:"timeout"`
	// contains filtered or unexported fields
}

type BridgeTask struct {
	BaseTask `mapstructure:",squash"`

	Name        string          `json:"name"`
	RequestData HttpRequestData `json:"requestData"`
	// contains filtered or unexported fields
}

type Config interface {
	BridgeResponseURL() *url.URL
	DatabaseMaximumTxDuration() time.Duration
	DatabaseURL() string
	DefaultHTTPLimit() int64
	DefaultHTTPTimeout() models.Duration
	DefaultMaxHTTPAttempts() uint
	DefaultHTTPAllowUnrestrictedNetworkAccess() bool
	TriggerFallbackDBPollInterval() time.Duration
	JobPipelineMaxTaskDuration() time.Duration
	JobPipelineMaxRunDuration() time.Duration
	JobPipelineParallelism() uint8
	JobPipelineReaperInterval() time.Duration
	JobPipelineReaperThreshold() time.Duration
}

type FinalErrors []null.String

type FinalResult struct {
	Values []interface{}
	Errors []error
}

type HTTPTask struct {
	BaseTask                       `mapstructure:",squash"`
	Method                         string
	URL                            models.WebURL
	RequestData                    HttpRequestData `json:"requestData"`
	AllowUnrestrictedNetworkAccess MaybeBool
	// contains filtered or unexported fields
}

type HttpRequestData map[string]interface{}

type JSONParseTask struct {
	BaseTask `mapstructure:",squash"`
	Path     JSONPath `json:"path"`
	// Lax when disabled will return an error if the path does not exist
	// Lax when enabled will return nil with no error if the path does not exist
	Lax bool
}

type JSONPath []string

type JSONSerializable struct {
	Val  interface{}
	Null bool
}

type MaybeBool string

type MedianTask struct {
	BaseTask      `mapstructure:",squash"`
	AllowedFaults uint64 `json:"allowedFaults"`
}

type MultiplyTask struct {
	BaseTask `mapstructure:",squash"`
	Times    decimal.Decimal `json:"times"`
}

type ORM interface {
	CreateSpec(ctx context.Context, db *gorm.DB, taskDAG TaskDAG, maxTaskTimeout models.Interval) (int32, error)
	CreateRun(ctx context.Context, jobID int32, meta map[string]interface{}) (int64, error)
	ProcessNextUnfinishedRun(ctx context.Context, fn ProcessRunFunc) (bool, error)
	ListenForNewRuns() (postgres.Subscription, error)
	InsertFinishedRunWithResults(ctx context.Context, run Run, trrs []TaskRunResult) (runID int64, err error)
	AwaitRun(ctx context.Context, runID int64) error
	RunFinished(runID int64) (bool, error)
	ResultsForRun(ctx context.Context, runID int64) ([]Result, error)
	DeleteRunsOlderThan(threshold time.Duration) error

	FindBridge(name models.TaskType) (models.BridgeType, error)

	DB() *gorm.DB
}

type PossibleErrorResponses struct {
	Error        string `json:"error"`
	ErrorMessage string `json:"errorMessage"`
}

type ProcessRunFunc func(ctx context.Context, txdb *gorm.DB, pRun Run, l logger.Logger) (TaskRunResults, error)

type Result struct {
	Value interface{}
	Error error
}

type ResultTask struct {
	BaseTask `mapstructure:",squash"`
}

type Run struct {
	ID               int64            `json:"-" gorm:"primary_key"`
	PipelineSpecID   int32            `json:"-"`
	PipelineSpec     Spec             `json:"pipelineSpec"`
	Meta             JSONSerializable `json:"meta"`
	Errors           JSONSerializable `json:"errors" gorm:"type:jsonb"`
	Outputs          JSONSerializable `json:"outputs" gorm:"type:jsonb"`
	CreatedAt        time.Time        `json:"createdAt"`
	FinishedAt       *time.Time       `json:"finishedAt"`
	PipelineTaskRuns []TaskRun        `json:"taskRuns" gorm:"foreignkey:PipelineRunID;->"`
}

type Runner interface {
	Start() error
	Close() error
	CreateRun(ctx context.Context, jobID int32, meta map[string]interface{}) (runID int64, err error)
	ExecuteRun(ctx context.Context, run Run, l logger.Logger) (trrs TaskRunResults, err error)
	ExecuteAndInsertNewRun(ctx context.Context, spec Spec, l logger.Logger) (finalResult FinalResult, err error)
	AwaitRun(ctx context.Context, runID int64) error
	ResultsForRun(ctx context.Context, runID int64) ([]Result, error)
}

type Spec struct {
	ID                int32           `gorm:"primary_key"`
	DotDagSource      string          `json:"dotDagSource"`
	CreatedAt         time.Time       `json:"-"`
	MaxTaskDuration   models.Interval `json:"-"`
	PipelineTaskSpecs []TaskSpec      `json:"-" gorm:"foreignkey:PipelineSpecID;->"`
}

type Task interface {
	Type() TaskType
	DotID() string
	Run(ctx context.Context, taskRun TaskRun, inputs []Result) Result
	OutputTask() Task
	SetOutputTask(task Task)
	OutputIndex() int32
	TaskTimeout() (time.Duration, bool)
	SetDefaults(inputValues map[string]string, g TaskDAG, self taskDAGNode) error
}

type TaskDAG struct {
	*simple.DirectedGraph
	DOTSource string
}

type TaskRun struct {
	ID                 int64             `json:"-" gorm:"primary_key"`
	Type               TaskType          `json:"type"`
	PipelineRun        Run               `json:"-"`
	PipelineRunID      int64             `json:"-"`
	Output             *JSONSerializable `json:"output" gorm:"type:jsonb"`
	Error              null.String       `json:"error"`
	PipelineTaskSpecID int32             `json:"-"`
	PipelineTaskSpec   TaskSpec          `json:"taskSpec" gorm:"foreignkey:PipelineTaskSpecID;->"`
	CreatedAt          time.Time         `json:"createdAt"`
	FinishedAt         *time.Time        `json:"finishedAt"`
}

type TaskRunResult struct {
	ID         int64
	TaskSpecID int32
	Result     Result
	FinishedAt time.Time
	IsTerminal bool
}

type TaskRunResults []TaskRunResult

type TaskSpec struct {
	ID             int32            `json:"-" gorm:"primary_key"`
	DotID          string           `json:"dotId"`
	PipelineSpecID int32            `json:"-"`
	PipelineSpec   Spec             `json:"-"`
	Type           TaskType         `json:"-"`
	JSON           JSONSerializable `json:"-" gorm:"type:jsonb"`
	Index          int32            `json:"-"`
	SuccessorID    null.Int         `json:"-"`
	CreatedAt      time.Time        `json:"-"`
}

type TaskType string