repository

Documentation

Index

• type ClaimAndGetPipelineParkedRunsDueParams
• type ClaimAndGetPipelineParkedRunsDueRow
• type DBTX
• type GetPipelineParkedRunsDueForResumeRow
• type InsertPipelineRunParams
• type InsertPipelineRunStageParams
• type PipelineParkedRun
• type PipelineRetryAttempt
• type PipelineRun
• type PipelineRunStage
• type Queries
  • func New(db DBTX) *Queries
  • func (q *Queries) ClaimAndGetPipelineParkedRunsDue(ctx context.Context, arg ClaimAndGetPipelineParkedRunsDueParams) ([]ClaimAndGetPipelineParkedRunsDueRow, error)
  • func (q *Queries) DeletePipelineParkedRun(ctx context.Context, runID string) error
  • func (q *Queries) GetPipelineParkedRunsDueForResume(ctx context.Context) ([]GetPipelineParkedRunsDueForResumeRow, error)
  • func (q *Queries) GetRetryAttempt(ctx context.Context, runID string) (int32, error)
  • func (q *Queries) InsertPipelineRun(ctx context.Context, arg InsertPipelineRunParams) error
  • func (q *Queries) InsertPipelineRunStage(ctx context.Context, arg InsertPipelineRunStageParams) error
  • func (q *Queries) UpdatePipelineRunComplete(ctx context.Context, arg UpdatePipelineRunCompleteParams) error
  • func (q *Queries) UpdatePipelineRunStage(ctx context.Context, arg UpdatePipelineRunStageParams) error
  • func (q *Queries) UpsertPipelineParkedRun(ctx context.Context, arg UpsertPipelineParkedRunParams) error
  • func (q *Queries) UpsertPipelineRun(ctx context.Context, arg UpsertPipelineRunParams) error
  • func (q *Queries) UpsertRetryAttempt(ctx context.Context, arg UpsertRetryAttemptParams) error
  • func (q *Queries) WithTx(tx pgx.Tx) *Queries
• type UpdatePipelineRunCompleteParams
• type UpdatePipelineRunStageParams
• type UpsertPipelineParkedRunParams
• type UpsertPipelineRunParams
• type UpsertRetryAttemptParams

Types

type ClaimAndGetPipelineParkedRunsDueParams

type ClaimAndGetPipelineParkedRunsDueParams struct {
	ClaimedBy pgtype.Text `json:"claimed_by"`
	Limit     int32       `json:"limit"`
}

type ClaimAndGetPipelineParkedRunsDueRow

type ClaimAndGetPipelineParkedRunsDueRow struct {
	RunID             string    `json:"run_id"`
	PipelineName      string    `json:"pipeline_name"`
	NextStageIndex    int32     `json:"next_stage_index"`
	InputForNextStage []byte    `json:"input_for_next_stage"`
	ResumeAt          time.Time `json:"resume_at"`
}

type DBTX

type DBTX interface {
	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
	QueryRow(context.Context, string, ...interface{}) pgx.Row
}

type GetPipelineParkedRunsDueForResumeRow

type GetPipelineParkedRunsDueForResumeRow struct {
	RunID             string    `json:"run_id"`
	PipelineName      string    `json:"pipeline_name"`
	NextStageIndex    int32     `json:"next_stage_index"`
	InputForNextStage []byte    `json:"input_for_next_stage"`
	ResumeAt          time.Time `json:"resume_at"`
}

type InsertPipelineRunParams

type InsertPipelineRunParams struct {
	RunID   string `json:"run_id"`
	Name    string `json:"name"`
	Payload []byte `json:"payload"`
}

type InsertPipelineRunStageParams

type InsertPipelineRunStageParams struct {
	PipelineRunID string `json:"pipeline_run_id"`
	StageIndex    int32  `json:"stage_index"`
	InputJson     []byte `json:"input_json"`
}

type PipelineParkedRun

type PipelineParkedRun struct {
	RunID             string             `json:"run_id"`
	PipelineName      string             `json:"pipeline_name"`
	NextStageIndex    int32              `json:"next_stage_index"`
	InputForNextStage []byte             `json:"input_for_next_stage"`
	ResumeAt          time.Time          `json:"resume_at"`
	ClaimedBy         pgtype.Text        `json:"claimed_by"`
	ClaimedAt         pgtype.Timestamptz `json:"claimed_at"`
	CreatedAt         time.Time          `json:"created_at"`
	UpdatedAt         time.Time          `json:"updated_at"`
}

type PipelineRetryAttempt

type PipelineRetryAttempt struct {
	RunID        string    `json:"run_id"`
	AttemptCount int32     `json:"attempt_count"`
	UpdatedAt    time.Time `json:"updated_at"`
}

type PipelineRun

type PipelineRun struct {
	RunID     string      `json:"run_id"`
	Name      string      `json:"name"`
	Payload   []byte      `json:"payload"`
	Status    string      `json:"status"`
	Result    []byte      `json:"result"`
	Error     pgtype.Text `json:"error"`
	CreatedAt time.Time   `json:"created_at"`
	UpdatedAt time.Time   `json:"updated_at"`
}

type PipelineRunStage

type PipelineRunStage struct {
	PipelineRunID string      `json:"pipeline_run_id"`
	StageIndex    int32       `json:"stage_index"`
	InputJson     []byte      `json:"input_json"`
	OutputJson    []byte      `json:"output_json"`
	Status        string      `json:"status"`
	Error         pgtype.Text `json:"error"`
	DurationMs    pgtype.Int8 `json:"duration_ms"`
	CreatedAt     time.Time   `json:"created_at"`
	UpdatedAt     time.Time   `json:"updated_at"`
}

type Queries

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

func New

func New(db DBTX) *Queries

func (*Queries) ClaimAndGetPipelineParkedRunsDue

func (q *Queries) ClaimAndGetPipelineParkedRunsDue(ctx context.Context, arg ClaimAndGetPipelineParkedRunsDueParams) ([]ClaimAndGetPipelineParkedRunsDueRow, error)

Claims up to :limit rows (FOR UPDATE SKIP LOCKED) so only one resumer processes each run. Stale claims (claimed_at older than 5 minutes) are treated as unclaimed.

func (*Queries) DeletePipelineParkedRun

func (q *Queries) DeletePipelineParkedRun(ctx context.Context, runID string) error

func (*Queries) GetPipelineParkedRunsDueForResume

func (q *Queries) GetPipelineParkedRunsDueForResume(ctx context.Context) ([]GetPipelineParkedRunsDueForResumeRow, error)

func (*Queries) GetRetryAttempt

func (q *Queries) GetRetryAttempt(ctx context.Context, runID string) (int32, error)

func (*Queries) InsertPipelineRun

func (q *Queries) InsertPipelineRun(ctx context.Context, arg InsertPipelineRunParams) error

func (*Queries) InsertPipelineRunStage

func (q *Queries) InsertPipelineRunStage(ctx context.Context, arg InsertPipelineRunStageParams) error

func (*Queries) UpdatePipelineRunComplete

func (q *Queries) UpdatePipelineRunComplete(ctx context.Context, arg UpdatePipelineRunCompleteParams) error

func (*Queries) UpdatePipelineRunStage

func (q *Queries) UpdatePipelineRunStage(ctx context.Context, arg UpdatePipelineRunStageParams) error

func (*Queries) UpsertPipelineParkedRun

func (q *Queries) UpsertPipelineParkedRun(ctx context.Context, arg UpsertPipelineParkedRunParams) error

func (*Queries) UpsertPipelineRun

func (q *Queries) UpsertPipelineRun(ctx context.Context, arg UpsertPipelineRunParams) error

func (*Queries) UpsertRetryAttempt

func (q *Queries) UpsertRetryAttempt(ctx context.Context, arg UpsertRetryAttemptParams) error

func (*Queries) WithTx

func (q *Queries) WithTx(tx pgx.Tx) *Queries

type UpdatePipelineRunCompleteParams

type UpdatePipelineRunCompleteParams struct {
	RunID  string      `json:"run_id"`
	Status string      `json:"status"`
	Result []byte      `json:"result"`
	Error  pgtype.Text `json:"error"`
}

type UpdatePipelineRunStageParams

type UpdatePipelineRunStageParams struct {
	PipelineRunID string      `json:"pipeline_run_id"`
	StageIndex    int32       `json:"stage_index"`
	OutputJson    []byte      `json:"output_json"`
	Status        string      `json:"status"`
	Error         pgtype.Text `json:"error"`
	DurationMs    pgtype.Int8 `json:"duration_ms"`
}

type UpsertPipelineParkedRunParams

type UpsertPipelineParkedRunParams struct {
	RunID             string    `json:"run_id"`
	PipelineName      string    `json:"pipeline_name"`
	NextStageIndex    int32     `json:"next_stage_index"`
	InputForNextStage []byte    `json:"input_for_next_stage"`
	ResumeAt          time.Time `json:"resume_at"`
}

type UpsertPipelineRunParams

type UpsertPipelineRunParams struct {
	RunID   string `json:"run_id"`
	Name    string `json:"name"`
	Payload []byte `json:"payload"`
}

type UpsertRetryAttemptParams

type UpsertRetryAttemptParams struct {
	RunID        string `json:"run_id"`
	AttemptCount int32  `json:"attempt_count"`
}