ctxengine

Documentation

Index

• type Lineage
  • func (l *Lineage) GetEndedAt() *time.Time
  • func (l *Lineage) GetError() string
  • func (l *Lineage) GetEvents() []eventbus.Event
  • func (l *Lineage) GetID() string
  • func (l *Lineage) GetName() string
  • func (l *Lineage) GetStartedAt() time.Time
  • func (l *Lineage) GetState() string
• type LineageModel
  • func (LineageModel) TableName() string
• type Module
  • func NewModule() *Module
  • func (m *Module) Events(ctx context.Context, obj *model.WorkflowLineage) ([]*mdk.Event, error)
  • func (m *Module) FieldResolvers() map[string]any
  • func (m *Module) GetWorkflowLineage(ctx context.Context, id string) (*model.WorkflowLineage, error)
  • func (m *Module) HealthMutation(ctx context.Context) (string, error)
  • func (m *Module) HealthQuery(ctx context.Context) (string, error)
  • func (m *Module) ID() string
  • func (m *Module) Init(ctx context.Context, rt mdk.Runtime) error
  • func (m *Module) ListLineages(ctx context.Context) ([]*model.WorkflowLineage, error)
  • func (m *Module) Models() []any
  • func (m *Module) Mutations() map[string]any
  • func (m *Module) Projector() mdk.Projector
  • func (m *Module) Queries() map[string]any
  • func (m *Module) RelatedLineages(ctx context.Context, obj *model.WorkflowLineage) ([]*model.WorkflowLineage, error)
  • func (m *Module) Routes() []mdk.Route
  • func (m *Module) Shutdown(ctx context.Context) error
• type Projector
  • func NewProjector(bus mdk.EventBus) *Projector
  • func (p *Projector) GetLineage(id string) (*Lineage, error)
  • func (p *Projector) GetRelatedLineages(ctx context.Context, id string) ([]*Lineage, error)
  • func (p *Projector) ListLineages() []registry.LineageData
  • func (p *Projector) QueryLineages(filter func(registry.LineageData) bool) []registry.LineageData
  • func (p *Projector) SetDB(database *gorm.DB)
  • func (p *Projector) Start(ctx context.Context) error
• type StepLineage

Types

type Lineage

type Lineage struct {
	ID        string           `json:"id"`
	Name      string           `json:"name"`
	Version   string           `json:"version"`
	State     string           `json:"state"`
	StartedAt time.Time        `json:"started_at"`
	EndedAt   *time.Time       `json:"ended_at,omitempty"`
	Steps     []*StepLineage   `json:"steps"`
	Events    []eventbus.Event `json:"events"`
	Error     string           `json:"error,omitempty"`
}

Lineage represents the execution history of a workflow.

func (*Lineage) GetEndedAt

func (l *Lineage) GetEndedAt() *time.Time

func (*Lineage) GetError

func (l *Lineage) GetError() string

func (*Lineage) GetEvents

func (l *Lineage) GetEvents() []eventbus.Event

func (*Lineage) GetID

func (l *Lineage) GetID() string

func (*Lineage) GetName

func (l *Lineage) GetName() string

func (*Lineage) GetStartedAt

func (l *Lineage) GetStartedAt() time.Time

func (*Lineage) GetState

func (l *Lineage) GetState() string

type LineageModel

type LineageModel struct {
	ID        string `gorm:"primaryKey"`
	Name      string
	Version   string
	State     string
	StartedAt time.Time
	EndedAt   *time.Time
	Error     string
	Steps     string `gorm:"type:text"` // Serialized JSON of []*StepLineage
	Events    string `gorm:"type:text"` // Serialized JSON of []eventbus.Event
	CreatedAt time.Time
	UpdatedAt time.Time
}

LineageModel GORM model for persisting lineages to database.

func (LineageModel) TableName

func (LineageModel) TableName() string

type Module

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

Module implements the mdk.Module interface.

func NewModule

func NewModule() *Module

func (*Module) Events

func (m *Module) Events(ctx context.Context, obj *model.WorkflowLineage) ([]*mdk.Event, error)

func (*Module) FieldResolvers

func (m *Module) FieldResolvers() map[string]any

func (*Module) GetWorkflowLineage

func (m *Module) GetWorkflowLineage(ctx context.Context, id string) (*model.WorkflowLineage, error)

func (*Module) HealthMutation

func (m *Module) HealthMutation(ctx context.Context) (string, error)

func (*Module) HealthQuery

func (m *Module) HealthQuery(ctx context.Context) (string, error)

func (*Module) ID

func (m *Module) ID() string

func (*Module) Init

func (m *Module) Init(ctx context.Context, rt mdk.Runtime) error

func (*Module) ListLineages

func (m *Module) ListLineages(ctx context.Context) ([]*model.WorkflowLineage, error)

func (*Module) Models

func (m *Module) Models() []any

func (*Module) Mutations

func (m *Module) Mutations() map[string]any

func (*Module) Projector

func (m *Module) Projector() mdk.Projector

func (*Module) Queries

func (m *Module) Queries() map[string]any

func (*Module) RelatedLineages

func (m *Module) RelatedLineages(ctx context.Context, obj *model.WorkflowLineage) ([]*model.WorkflowLineage, error)

func (*Module) Routes

func (m *Module) Routes() []mdk.Route

func (*Module) Shutdown

func (m *Module) Shutdown(ctx context.Context) error

type Projector

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

Projector listens to events and maintains execution lineage.

func NewProjector

func NewProjector(bus mdk.EventBus) *Projector

NewProjector creates a new Projector.

func (*Projector) GetLineage

func (p *Projector) GetLineage(id string) (*Lineage, error)

GetLineage returns the lineage for a workflow ID.

func (*Projector) GetRelatedLineages

func (p *Projector) GetRelatedLineages(ctx context.Context, id string) ([]*Lineage, error)

GetRelatedLineages returns all lineages that share metadata with the given workflow ID.

func (*Projector) ListLineages

func (p *Projector) ListLineages() []registry.LineageData

ListLineages returns all lineages as registry.LineageData.

func (*Projector) QueryLineages

func (p *Projector) QueryLineages(filter func(registry.LineageData) bool) []registry.LineageData

QueryLineages returns lineages that match the given filter.

func (*Projector) SetDB

func (p *Projector) SetDB(database *gorm.DB)

SetDB sets the database connection for the projector.

func (*Projector) Start

func (p *Projector) Start(ctx context.Context) error

Start begins listening for workflow events.

type StepLineage

type StepLineage struct {
	ID        string     `json:"id"`
	StepID    string     `json:"step_id"`
	State     string     `json:"state"`
	StartedAt time.Time  `json:"started_at"`
	EndedAt   *time.Time `json:"ended_at,omitempty"`
	Attempts  int        `json:"attempts"`
	Error     string     `json:"error,omitempty"`
}