sparkv1

type Bindable interface {
	Bind(a any) error
	GetValue() ([]byte, error)
	GetMimeType() string
	String() string
}

type BindableConfig interface {
	Bind(a any) error
	Raw() ([]byte, error)
}

type BindableMap map[string]Bindable

type BindableValue Value

type Builder interface {
	NewChain(name string) BuilderChain
	ChainFinalizer
}

type BuilderChain interface {
	ChainNode
}

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

type ChainCancelled interface {
	Cancelled(newNode Chain) ChainComplete
}

type ChainCancelledOrComplete interface {
	ChainCancelled
	ChainComplete
}

type ChainCompensate interface {
	Compensate(newNode Chain) ChainCancelledOrComplete
}

type ChainComplete interface {
	Complete(completeDefinitionFn CompleteDefinitionFn, options ...StageOption) Chain
}

type ChainFinalizer interface {
	BuildChain() *SparkChain
}

type ChainNode interface {
	ChainStage // must have at least 1 Stage
}

type ChainReport struct {
	Errors   []error
	StageMap map[string]ChainReportStage
	NodeMap  map[string]ChainReportNode
}

type ChainReportNode struct {
	Name          string
	CanCompensate bool
	CanCancel     bool
	// contains filtered or unexported fields
}

type ChainReportStage struct {
	Name  string
	Crumb string
}

type ChainStage interface {
	Stage(name string, stageDefinitionFn StageDefinitionFn, options ...StageOption) ChainStageAny
}

type ChainStageAny interface {
	ChainStage
	ChainCompensate
	ChainCancelled
	ChainComplete
}

type CompleteContext interface {
	StageContext
	Output(variables ...*Var) error
	Name() string
}

type CompleteDefinitionFn = func(ctx CompleteContext) StageError

type CompleteStage struct {
	Node *Node
	Name string
	// contains filtered or unexported fields
}

type Config struct {
	Id                     string        `yaml:"id"`
	Name                   string        `yaml:"Name"`
	NatsRequestSubject     string        `yaml:"nats_request_subject"`
	NatsResponseSubject    string        `yaml:"nats_response_subject"`
	NatsRequestStreamName  string        `yaml:"nats_request_stream_name"`
	NatsResponseStreamName string        `yaml:"nats_response_stream_name"`
	NatsBucket             string        `yaml:"nats_bucket"`
	RetryCount             uint          `yaml:"retry_count"`
	RetryBackoff           time.Duration `yaml:"retry_backoff"`
	RetryBackoffMultiplier uint          `yaml:"retry_backoff_multiplier"`
	Timeout                time.Duration `yaml:"timeout"`
	Health                 *configHealth `yaml:"health"`
	Server                 *configServer `yaml:"plugin"`
	Log                    *configLog    `yaml:"logging"`
	App                    *configApp    `yaml:"app"`
	Nats                   *configNats   `yaml:"nats"`
}

type ConfigType string

type Context interface {
	context.Context
	JobKey() string
	CorrelationID() string
	TransactionID() string
}

type ErrorCode string

type ErrorOption = func(err *stageError) *stageError

type ExecuteSparkError struct {
	StageName    string           `json:"stage_name"`
	ErrorCode    ErrorCode        `json:"error_code"`
	ErrorMessage string           `json:"error_message,omitempty"`
	Metadata     map[string]any   `json:"metadata,omitempty"`
	StackTrace   []StackTraceItem `json:"stack_trace"`
}

type ExecuteSparkInputs map[string]*BindableValue

type ExecuteSparkOutput struct {
	Error         *ExecuteSparkError `json:"error,omitempty"`
	JobPid        *JobPid            `json:"job_pid,omitempty"`
	VariablesKey  string             `json:"variables_key,omitempty"`
	JobKey        string             `json:"job_key,omitempty"`
	CorrelationId string             `json:"correlation_id,omitempty"`
	TransactionId string             `json:"transaction_id,omitempty"`
	Model         string             `json:"model,omitempty"`
	Outputs       BindableMap        `json:"outputs,omitempty"`
}

type ExecuteStageRequest struct {
	StageName     string
	TransactionId string
	CorrelationId string
	JobKey        string
	Inputs        map[string]Bindable
}

type ExecuteStageResponse struct {
	Outputs BindableMap        `json:"outputs,omitempty"`
	Error   *ExecuteSparkError `json:"error,omitempty"`
}

type Gettable interface {
	Get(name string) Bindable
}

type IOState interface {
	GetVar(varName string) any
}

type InitContext interface {
	Config() BindableConfig
}

type Input interface {
	Bindable
}

type Inputs interface {
	Get(name string) Bindable
}

type InternalStageTracker interface {
	SetStageResult(name string, value Bindable)
	SetStageStatus(name string, status StageStatus)
}

type JobContext struct {
	context.Context
	Metadata *JobMetadata
}

type JobMetadata struct {
	SparkId                string             `json:"spark_id"` // id of the spark to execute
	JobKeyValue            string             `json:"job_key"`
	CorrelationIdValue     string             `json:"correlation_id"`
	TransactionIdValue     string             `json:"transaction_id"`
	RetryCount             uint               `json:"retry_count"`
	RetryBackoff           time.Duration      `json:"retry_backoff"`
	RetryBackoffMultiplier uint               `json:"retry_backoff_multiplier"`
	JobPid                 *JobPid            `json:"job_pid,omitempty"`
	VariablesBucket        string             `json:"variables_bucket"`
	VariablesKey           string             `json:"variables_key"`
	Model                  string             `json:"model,omitempty"`
	Inputs                 ExecuteSparkInputs `json:"-"`
}

type JobPid struct {
	Address   string `json:"Address"`
	Id        string `json:"Id"`
	RequestId uint32 `json:"request_id"`
}

type JobState struct {
	JobContext   *JobMetadata
	StageResults map[string]Bindable
}

type JobWorkflow interface {
	Run(msg jetstream.Msg)
	ExecuteStageActivity(ctx context.Context, req *ExecuteStageRequest, io SparkDataIO) (Bindable, StageError)
	ExecuteCompleteActivity(ctx context.Context, req *ExecuteStageRequest, io SparkDataIO) (*ExecuteStageResponse, StageError)
}

type Logger interface {
	Info(format string, v ...any)
	Warn(format string, v ...any)
	Debug(format string, v ...any)
	Error(err error, format string, v ...any)
	AddFields(k string, v any) Logger
}

type Node struct {
	Stages     []*Stage
	Complete   *CompleteStage
	Cancel     *Node
	Compensate *Node
	Name       string
	// contains filtered or unexported fields
}

type Option = func(je *SparkOpts) *SparkOpts

type RetryConfig struct {
	Times             uint          `json:"times" yaml:"times"`
	FirstBackoffWait  time.Duration `json:"first_backoff_wait" yaml:"first_backoff_wait"`
	BackoffMultiplier uint          `json:"backoff_multiplier" yaml:"backoff_multiplier"`
}

type Spark interface {
	BuildChain(b Builder) Chain
	Init(ctx InitContext) error
	Stop()
}

type SparkChain struct {
	RootNode *Node
	// contains filtered or unexported fields
}

type SparkDataIO interface {
	NewInput(name, stageName string, value *BindableValue) Bindable
	NewOutput(stageName string, value *BindableValue) (Bindable, error)
	GetStageResult(stageName string) (Bindable, error)
	PutStageResult(stageName string, stageValue []byte) (Bindable, error)
	LoadVariables(key string) error
	GetInputValue(name string) (*BindableValue, bool)
	SetInitialInputs(inputs ExecuteSparkInputs)
}

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

type StackTraceItem struct {
	Type     string `json:"type"`
	Filepath string `json:"filepath"`
}

type Stage struct {
	Node *Node
	Name string
	// contains filtered or unexported fields
}

type StageContext interface {
	Context
	Input(names string) Input
	StageResult(name string) Bindable
	Log() Logger
	Name() string
}

type StageDefinitionFn = func(ctx StageContext) (any, StageError)

type StageError interface {
	ErrorCode() ErrorCode
	StageName() string
	Error() string
	Metadata() map[string]any
	GetRetryConfig() *RetryConfig
	// contains filtered or unexported methods
}

type StageOption = func(StageOptionParams) StageError

type StageOptionParams interface {
	StageName() string
	Context() Context
}

type StageStatus string

type StageTracker interface {
	GetStageResult(name string) (data any, mime codec.MimeType, err StageError)
	AssertStageCompleted(stageName string)
	AssertStageStarted(stageName string)
	AssertStageSkipped(stageName string)
	AssertStageCancelled(stageName string)
	AssertStageFailed(stageName string)
	AssertStageResult(stageName string, expectedStageResult any)
	AssertStageOrder(stageNames ...string)
}

type Value struct {
	Value    []byte `json:"value"`
	MimeType string `json:"mime_type"`
}

type Var struct {
	Name     string
	MimeType codec.MimeType
	Value    []byte
}

type Worker interface {
	Run()
}

type WorkflowOption = func(je *workflowOpts) *workflowOpts