Help us shape the future of Go! Take the 2026 Go Developer Survey

postgresql

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 22, 2026 License: BSD-2-Clause Imports: 13 Imported by: 0

Documentation

Overview

Package postgresql provides PostgreSQL persistence implementation for workflows and triggers.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type EventDefinitions 

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

func (*EventDefinitions) ByTopic 

func (e *EventDefinitions) ByTopic(ctx context.Context, topic string) ([]*models.EventDefinition, error)

func (*EventDefinitions) Get 

func (e *EventDefinitions) Get(ctx context.Context, key string) (*models.EventDefinition, error)

func (*EventDefinitions) List 

func (e *EventDefinitions) List(ctx context.Context) ([]*models.EventDefinition, error)

func (*EventDefinitions) Save 

func (e *EventDefinitions) Save(ctx context.Context, definition *models.EventDefinition) error

type ExecutionContextRepository 

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

ExecutionContextRepository handles execution context-related database operations.

func NewExecutionContextRepository 

func NewExecutionContextRepository(db *sql.DB, logger *slog.Logger) *ExecutionContextRepository

NewExecutionContextRepository creates a new execution context repository.

func (*ExecutionContextRepository) GetExecutionContext 

func (ecr *ExecutionContextRepository) GetExecutionContext(ctx context.Context, executionID string) (*models.ExecutionContext, error)

GetExecutionContext retrieves an execution context by its ID from the database.

func (*ExecutionContextRepository) GetExecutionsByStatus 

func (ecr *ExecutionContextRepository) GetExecutionsByStatus(ctx context.Context, status models.ExecutionStatus) ([]*models.ExecutionContext, error)

GetExecutionsByStatus retrieves all execution contexts with a specific status.

func (*ExecutionContextRepository) GetExecutionsByWorkflow 

func (ecr *ExecutionContextRepository) GetExecutionsByWorkflow(ctx context.Context, workflowID string, limit int) ([]*models.ExecutionContext, error)

GetExecutionsByWorkflow retrieves all execution contexts for a specific workflow.

func (*ExecutionContextRepository) SaveExecutionContext 

func (ecr *ExecutionContextRepository) SaveExecutionContext(ctx context.Context, execCtx *models.ExecutionContext) error

SaveExecutionContext saves an execution context to the database.

func (*ExecutionContextRepository) UpdateExecutionContext 

func (ecr *ExecutionContextRepository) UpdateExecutionContext(ctx context.Context, execCtx *models.ExecutionContext) error

UpdateExecutionContext updates an existing execution context in the database.

type Jobs 

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

func (*Jobs) ProcessNext 

func (j *Jobs) ProcessNext(
	ctx context.Context,
	process func(ctx context.Context, job models.Job, execution *models.ExecutionContext) ([]models.Job, error),
) (bool, error)

func (*Jobs) Resume 

func (j *Jobs) Resume(ctx context.Context, executionID, nodeID string, data map[string]any) error

func (*Jobs) ResumeIf 

func (j *Jobs) ResumeIf(ctx context.Context, executionID, nodeID string, data map[string]any, allowed func(models.NodeResult) bool) error

func (*Jobs) Start 

func (j *Jobs) Start(ctx context.Context, execution *models.ExecutionContext, job models.Job) error

type Persistence 

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

Persistence implements the persistence layer for PostgreSQL.

func NewPersistence 

func NewPersistence(ctx context.Context, logger *slog.Logger, databaseURL string) (*Persistence, error)

NewPersistence creates a new PostgreSQL persistence layer.

func (*Persistence) Close 

func (p *Persistence) Close(ctx context.Context) error

Close closes the database connection.

func (*Persistence) DeleteFinishedExecutions added in v0.2.0 

func (p *Persistence) DeleteFinishedExecutions(ctx context.Context, finishedBefore time.Time, limit int) (int, error)

func (*Persistence) EventDefinitions 

func (p *Persistence) EventDefinitions() *EventDefinitions

func (*Persistence) EventTriggers 

func (p *Persistence) EventTriggers(ctx context.Context, eventKey, tenantID string) ([]models.TriggerNodeMatch, error)

func (*Persistence) ExecutionContextRepository 

func (p *Persistence) ExecutionContextRepository() persistence.ExecutionContextRepository

func (*Persistence) HealthCheck 

func (p *Persistence) HealthCheck(ctx context.Context) error

HealthCheck verifies the database connection is healthy.

func (*Persistence) Jobs 

func (p *Persistence) Jobs() *Jobs

func (*Persistence) KafkaTopics 

func (p *Persistence) KafkaTopics(ctx context.Context) ([]string, error)

func (*Persistence) KafkaTriggers 

func (p *Persistence) KafkaTriggers(ctx context.Context, topic string) ([]models.TriggerNodeMatch, error)

func (*Persistence) RemoteNodeTypes 

func (p *Persistence) RemoteNodeTypes() *RemoteNodeTypes

func (*Persistence) Schedules 

func (p *Persistence) Schedules() *Schedules

func (*Persistence) WebhookTriggers 

func (p *Persistence) WebhookTriggers(ctx context.Context, tenantID, path string) ([]models.TriggerNodeMatch, error)

func (*Persistence) WorkflowRepository 

func (p *Persistence) WorkflowRepository() persistence.WorkflowRepository

WorkflowRepository returns the workflow repository implementation.

type RemoteNodeTypes 

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

func (*RemoteNodeTypes) Get 

func (r *RemoteNodeTypes) Get(ctx context.Context, id string) (*models.RemoteNodeType, error)

func (*RemoteNodeTypes) Save 

func (r *RemoteNodeTypes) Save(ctx context.Context, nodeType *models.RemoteNodeType) error

type Schedules 

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

func (*Schedules) ProcessDue 

func (s *Schedules) ProcessDue(ctx context.Context, fire func(context.Context, models.Schedule) (models.Firing, error)) (bool, error)

func (*Schedules) Sync 

func (s *Schedules) Sync(ctx context.Context, first func(models.Schedule) (time.Time, error)) error

type WorkflowRepository 

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

WorkflowRepository handles workflow-related database operations.

func NewWorkflowRepository 

func NewWorkflowRepository(db *sql.DB, logger *slog.Logger) *WorkflowRepository

NewWorkflowRepository creates a new workflow repository.

func (*WorkflowRepository) CreateDraftFromPublished 

func (r *WorkflowRepository) CreateDraftFromPublished(ctx context.Context, workflowGroupID string) (*models.Workflow, error)

CreateDraftFromPublished creates a draft copy from published version.

func (*WorkflowRepository) Delete 

func (r *WorkflowRepository) Delete(ctx context.Context, id string) error

Delete soft deletes a workflow by setting deleted_at timestamp.

func (*WorkflowRepository) GetAll 

func (r *WorkflowRepository) GetAll(ctx context.Context, tenantID string) ([]*models.Workflow, error)

GetAll returns all workflows from the database.

func (*WorkflowRepository) GetByID 

func (r *WorkflowRepository) GetByID(ctx context.Context, id string) (*models.Workflow, error)

func (*WorkflowRepository) GetCurrentWorkflow 

func (r *WorkflowRepository) GetCurrentWorkflow(ctx context.Context, workflowGroupID string) (*models.Workflow, error)

GetCurrentWorkflow returns the current version (published if exists, otherwise draft).

func (*WorkflowRepository) GetDraftWorkflow 

func (r *WorkflowRepository) GetDraftWorkflow(ctx context.Context, workflowGroupID string) (*models.Workflow, error)

GetDraftWorkflow returns the draft version of a workflow group.

func (*WorkflowRepository) GetPublishedWorkflow 

func (r *WorkflowRepository) GetPublishedWorkflow(ctx context.Context, workflowGroupID string) (*models.Workflow, error)

GetPublishedWorkflow returns the published version of a workflow group.

func (*WorkflowRepository) GetWorkflowVersions 

func (r *WorkflowRepository) GetWorkflowVersions(ctx context.Context, workflowGroupID string) ([]*models.Workflow, error)

GetWorkflowVersions returns all versions of a workflow by group ID.

func (*WorkflowRepository) PublishWorkflow 

func (r *WorkflowRepository) PublishWorkflow(ctx context.Context, workflowID string) error

PublishWorkflow handles the publish operation.

func (*WorkflowRepository) Save 

func (r *WorkflowRepository) Save(ctx context.Context, workflow *models.Workflow) error

Save saves a workflow to the database.

func (*WorkflowRepository) UnpublishWorkflow added in v0.2.0 

func (r *WorkflowRepository) UnpublishWorkflow(ctx context.Context, workflowID string) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.