pipeline

type BaseTask struct {
	Index int32 `mapstructure:"index" json:"-" `
	// 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
	JobPipelineDBPollInterval() time.Duration
	JobPipelineMaxTaskDuration() time.Duration
	JobPipelineParallelism() uint8
	JobPipelineReaperInterval() time.Duration
	JobPipelineReaperThreshold() time.Duration
}

type FinalErrors []null.String

type HTTPTask struct {
	BaseTask    `mapstructure:",squash"`
	Method      string
	URL         models.WebURL
	RequestData HttpRequestData `json:"requestData"`
	// 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{}
}

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

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

type ORM interface {
	CreateSpec(ctx context.Context, taskDAG TaskDAG) (int32, error)
	CreateRun(ctx context.Context, jobID int32, meta map[string]interface{}) (int64, error)
	ProcessNextUnclaimedTaskRun(ctx context.Context, fn ProcessTaskRunFunc) (bool, error)
	ListenForNewRuns() (postgres.Subscription, 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)
}

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

type ProcessTaskRunFunc func(ctx context.Context, txdb *gorm.DB, jobID int32, ptRun TaskRun, predecessors []TaskRun) Result

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;association_autoupdate:false;association_autocreate:false"`
}

type Runner interface {
	Start()
	Stop()
	CreateRun(ctx context.Context, jobID int32, meta map[string]interface{}) (int64, 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
	CreatedAt    time.Time
}

type Task interface {
	Type() TaskType
	DotID() string
	Run(ctx context.Context, taskRun TaskRun, inputs []Result) Result
	OutputTask() Task
	SetOutputTask(task Task)
	OutputIndex() int32
}

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;association_autoupdate:false;association_autocreate:false"`
	CreatedAt          time.Time         `json:"createdAt"`
	FinishedAt         *time.Time        `json:"finishedAt"`
}

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

type TaskType string