sqlike

Documentation

Overview

Package sqlike provides a consistent and unique API for SQL-like integrations.

This package could have been designed as an integration. But we would have lost the possibility for a SQL-like database to easily leverage other specifications such as Analytics or Infrastructure-as-Code.

Index

• Constants
• type Config
  • func (config *Config) Validate() error
• type Handler
  • func New(ctx context.Context, from integration.Integration, config Config, ...) (*Handler, error)
  • func (h *Handler) ActivityOperationsExecute(ctx context.Context, input InputOperations, step dslops.Step) (OutputOperations, error)
  • func (h *Handler) ActivityOperationsValidation(ctx context.Context, input InputOperations) (OutputOperations, error)
  • func (h *Handler) ActivityQueriesExecution(ctx context.Context, input InputQueries) (OutputQueries, error)
  • func (h *Handler) ActivityQueriesValidation(ctx context.Context, input InputQueries) (OutputQueries, error)
  • func (h *Handler) ChildWorkflowOperations(ctx workflow.Context, input InputOperations, step dslops.Step) (OutputOperations, error)
  • func (h *Handler) Close() error
  • func (h *Handler) ConfigMap() map[string]any
  • func (h *Handler) Init() error
  • func (h *Handler) IsReady() bool
  • func (h *Handler) ListActivities() []string
  • func (h *Handler) ListWorkflows() []string
  • func (h *Handler) Register(w worker.Worker) error
  • func (h *Handler) String() string
  • func (h *Handler) WorkflowOperations(ctx workflow.Context, input InputOperations) (OutputOperations, error)
  • func (h *Handler) WorkflowQueries(ctx workflow.Context, input InputQueries) (OutputQueries, error)
• type InputOperations
  • func (input *InputOperations) Validate(config *Config) error
• type InputQueries
  • func (input *InputQueries) Validate(config *Config) error
• type OutputOperations
• type OutputQueries
• type Policies
• type SQLike
• type With
  • func WithDB(db *sql.DB) With

Constants

const Specification string = "sqlike"

Specification is the string representation of the SQL-like specification.

Types

type Config

type Config struct {

	// IsolationLevel is the transaction isolation level used in transactions.
	IsolationLevel sql.IsolationLevel

	// Temporal is the Temporal client used within this specification. It is needed
	// to handle signals across workflows to locking and releasing the mutex.
	//
	// TODO(mutex): Is there a way to have access to the Temporal client without
	// asking it in the Config?
	//
	// Required.
	Temporal client.Client `json:"-"`

	// Policies allows to set activity policies, such as timeouts and retries.
	Policies Policies `json:"policies"`
	// contains filtered or unexported fields
}

Config allows the end-user to configure the specification for an integration.

func (*Config) Validate

func (config *Config) Validate() error

Validate validates the config passed by the end-user when registering the specification for the overlying integration. It returns an error if anything critical occured.

type Handler

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

Handler handles the SQL-like specification for an integration.

func New

func New(ctx context.Context, from integration.Integration, config Config, attachments ...With) (*Handler, error)

New returns a new SQL-like Handler for the given integration. It applies the configuration passed by the end-user (forwarded by the integration). It also applies additional attachments an integration might leverage.

Example:

func New(myintegration, config, WithDB(db))

func (*Handler) ActivityOperationsExecute

func (h *Handler) ActivityOperationsExecute(ctx context.Context, input InputOperations, step dslops.Step) (OutputOperations, error)

ActivityOperationsRun is the activity executing the set of operations in the SQL database.

Note that the operations are not wrapped inside a transaction! Some operations (such as concurrent indices) can't run inside a transaction. It's up to the end-user to explicitly set the transaction in the operation file.

func (*Handler) ActivityOperationsValidation

func (h *Handler) ActivityOperationsValidation(ctx context.Context, input InputOperations) (OutputOperations, error)

ActivityOperationsValidation is the activity validating the input.

func (*Handler) ActivityQueriesExecution

func (h *Handler) ActivityQueriesExecution(ctx context.Context, input InputQueries) (OutputQueries, error)

ActivityQueriesExecution is the activity executing the queries in the SQL database.

Note: Queries are all wrapped inside a transaction to ensure atomicity.

func (*Handler) ActivityQueriesValidation

func (h *Handler) ActivityQueriesValidation(ctx context.Context, input InputQueries) (OutputQueries, error)

ActivityQueriesValidation is the activity validating the input.

func (*Handler) ChildWorkflowOperations

func (h *Handler) ChildWorkflowOperations(ctx workflow.Context, input InputOperations, step dslops.Step) (OutputOperations, error)

ChildWorkflowOperations is the child workflow executing the DSL workflow. It's the same worflow for "running" and "rollingback".

func (*Handler) Close

func (h *Handler) Close() error

Close tries to properly close the specification. An error is returned in case the Handler has already been closed.

func (*Handler) ConfigMap

func (h *Handler) ConfigMap() map[string]any

ConfigMap transforms the configuration to a map, including a "from" key with the configuration of the overlying integration.

func (*Handler) Init

func (h *Handler) Init() error

Init initializes the specification. An error is returned in case the Handler has already been initialized.

func (*Handler) IsReady

func (h *Handler) IsReady() bool

IsReady indicates if the specification is ready to be consumed by the overlying integration. The specification must be initialized, must not be closed, and must have registered its workflows and activities in the Temporal worker.

func (*Handler) ListActivities

func (h *Handler) ListActivities() []string

ListActivities returns a sorted list of activities' name registered by the specification for the overlying integration.

func (*Handler) ListWorkflows

func (h *Handler) ListWorkflows() []string

ListWorkflows returns a sorted list of workflows' name registered by the specification for the overlying integration.

func (*Handler) Register

func (h *Handler) Register(w worker.Worker) error

Register registers the specification's workflows and activities in the given Temporal worker. An error is returned in case the registration has already been made.

func (*Handler) String

func (h *Handler) String() string

String returns the string representation of the overlying integration.

func (*Handler) WorkflowOperations

func (h *Handler) WorkflowOperations(ctx workflow.Context, input InputOperations) (OutputOperations, error)

WorkflowOperations is a high-level opiniated Temporal workflow. Before executing the DSL workflow sent by the client, it executes the following activities:

1. ActivityOperationsValidation: Validates the input (local activity).

2. ActivityMutex: Start and signal a workflow used as a mutex since only one workflow for a given integration within a namespace must run at the same time. If mutex is acquired, the following activites are executed:

func (*Handler) WorkflowQueries

func (h *Handler) WorkflowQueries(ctx workflow.Context, input InputQueries) (OutputQueries, error)

WorkflowQueries is a high-level opiniated Temporal workflow. It executes the following activities:

1. ActivityQueriesValidation: Validates the input (local activity).

2. ActivityQueriesExecution: Executes the SQL queries inside a transaction in the SQL database.

type InputOperations

type InputOperations struct {

	// Policies passed in the input can override the ones set when creating the
	// specification. It will only apply if allowed in the integration's Config
	// via AllowPoliciesOverride.
	Policies *Policies `json:"policies"`

	// Context represents the event's context, shared across every specifications
	// and integrations of this ecosystem.
	Context *event.Context `json:"context,omitempty"`

	// Version is the version number holding the timestamp of the operations set.
	// It's a 14 length character, formatted with YYYYMMDDHHMISS.
	//
	// Example: "20220722140548"
	//
	// Required.
	Version time.Time `json:"version"`

	// Name is the slugified title of the operations set.
	//
	// Example: "add_users_pkey"
	//
	// Required.
	Name string `json:"name"`

	// Direction indicates if want to apply ("up") or roll back ("down") the
	// operations set. It's one of "up", "down".
	//
	// Example: "down"
	//
	// Required.
	Direction string `json:"direction"`

	// Discards indicates if the operations set should be discarded in addition to
	// being rolled back. This can only be true if Direction is "down".
	Discard bool `json:"discard"`

	// Workflow is the DSL workflow to execute.
	//
	// Required.
	Workflow dslops.Workflow `json:"workflow"`
}

InputOperations is the input for the "Operations" workflow and activities.

func (*InputOperations) Validate

func (input *InputOperations) Validate(config *Config) error

Validate can be used to validate the workflow/activity's input. It's the validation function used in the local activity ActivityOperationsValidation.

type InputQueries

type InputQueries struct {

	// Policies passed in the input can override the ones set when creating the
	// specification. It will only apply if allowed in the integration's Config
	// via AllowPoliciesOverride.
	Policies *Policies `json:"policies"`

	// Context represents the event's context, shared across every specifications
	// and integrations of this ecosystem.
	Context *event.Context `json:"context,omitempty"`

	// Queries is a list of SQL queries to execute within a same transaction.
	//
	// Required.
	Queries []string `json:"queries"`
}

InputQueries is the input for the "Queries" workflow and activities.

func (*InputQueries) Validate

func (input *InputQueries) Validate(config *Config) error

Validate can be used to validate the workflow/activity's input. It's the validation function used in the local activity ActivityQueriesValidation.

type OutputOperations

type OutputOperations struct {

	// Status is the status of the workflow or activity. It's one of "success",
	// "failure".
	Status lifecycle.Status `json:"status"`

	// IsDiscarded indicates if the set of operations was successfully discarded.
	// This can only be true if the input's Discard was also true and if the workflow
	// is successful.
	//
	// This allows external clients such as the "tsql" CLI to browse the Temporal
	// history and be aware of operations set to ignore.
	IsDiscarded bool `json:"is_discarded"`

	// Mutex holds the details of the mutex used to execute the operations set.
	Mutex integration.Mutex `json:"mutex,omitempty"`

	// FailedAt indicates the activity at which the DSL has failed, if applicable.
	FailedAt *dsl.FailedActivity `json:"failed_at,omitempty"`
}

OutputOperations is the output for the "Operations" workflow and activities.

type OutputQueries

type OutputQueries struct {

	// Status is the status of the workflow or activity. It's one of "success",
	// "failure".
	Status lifecycle.Status `json:"status"`
}

OutputQueries is the output for the "Queries" workflow and activities.

type Policies

type Policies struct {

	// Queries is the policy to apply by the activity used to run queries inside
	// a transaction.
	Queries lifecycle.ActivityPolicy `json:"queries"`

	// Operation is the policy to apply by the activities used to execute a SQL
	// operation. It's also automatically used by the mutex-related workflow and
	// activity.
	//
	// When executing a workflow that must run only once, it's up to client to set
	// the workflow timeouts and the following retry policy:
	//
	//   client.StartWorkflowOptions{
	//     ID:                       "workflow",
	//     TaskQueue:                "queue",
	//     WorkflowExecutionTimeout: 10 * time.Second,
	//     RetryPolicy: &temporal.RetryPolicy{
	//       MaximumAttempts: 1,
	//     },
	//   }
	//
	Operation lifecycle.OnlyOncePolicy `json:"operation"`
	// contains filtered or unexported fields
}

Policies represents the Temporal activity policies to apply within the workflows exposed by this package and the overlying integration.

type SQLike

type SQLike interface {

	// RegisterWithSQLike allows an end-user to register the sqlike specification
	// within an integration.
	RegisterWithSQLike(worker.Worker, Config) error
}

SQLike is the interface used by an overlying integration to leverage the sqlike specification.

type With

type With func(*Handler) error

With allows an integration to customize the behavior of the specification. It's in addition to Config. Since Config is designed for — and accessible by — end-users, With is specifically designed for integrations.

For example, an integration must call WithDB to attach the *sql.DB:

func New(myintegration, config, WithDB(db))

func WithDB

func WithDB(db *sql.DB) With

WithDB must be passed to New to attach the *sql.DB from the overlying integration:

func New(myintegration, config, WithDB(db))