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) ActivityMigrationExecution(ctx context.Context, input InputMigration) (OutputMigration, error)
  • func (h *Handler) ActivityMigrationValidation(ctx context.Context, input InputMigration) (OutputMigration, 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) 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) WorkflowMigration(ctx workflow.Context, input InputMigration) (OutputMigration, error)
  • func (h *Handler) WorkflowQueries(ctx workflow.Context, input InputQueries) (OutputQueries, error)
• type InputMigration
  • func (input *InputMigration) Validate(logger log.Logger, config Config) error
• type InputQueries
  • func (input *InputQueries) Validate(logger log.Logger, config Config) error
• type OutputMigration
• 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) ActivityMigrationExecution

func (h *Handler) ActivityMigrationExecution(ctx context.Context, input InputMigration) (OutputMigration, error)

ActivityMigrationExecution is the activity executing the migration in the SQL database.

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

func (*Handler) ActivityMigrationValidation

func (h *Handler) ActivityMigrationValidation(ctx context.Context, input InputMigration) (OutputMigration, error)

ActivityMigrationValidation 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 that the 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) 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) WorkflowMigration

func (h *Handler) WorkflowMigration(ctx workflow.Context, input InputMigration) (OutputMigration, error)

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

1. ActivityMigrationValidation: 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:

1. ActivityMigrationExecution: Executes the SQL queries passed in the input.

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 InputMigration

type InputMigration 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 migration. It's
	// a 14 length character, formatted with YYYYMMDDHHMISS.
	//
	// Example: "20220722140548"
	//
	// Required.
	Version time.Time `json:"version"`

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

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

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

	// Content is the content of the migration file to execute.
	//
	// Required.
	Content []byte `json:"content"`
}

InputMigration is the input for the "Migration" workflow and activities.

func (*InputMigration) Validate

func (input *InputMigration) Validate(logger log.Logger, config Config) error

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

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(logger log.Logger, 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 OutputMigration

type OutputMigration 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 migration 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 tsql to browse the Temporal history
	// and be aware of migrations to ignore.
	IsDiscarded bool `json:"is_discarded"`

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

OutputMigration is the output for the "Migration" 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 {

	// Execution is the policy to apply by the activity used to publish a message.
	//
	// Note: Since this is a short-live policy, activity's heartbeat is not used.
	// Therefore, Execution.HeartbeatTimeout is not applied. We advise to set a
	// short-live Execution.SingleAttemptTimeout, such as 3 seconds.
	Execution lifecycle.ActivityPolicy `json:"execution"`

	// Migration is the policy to apply by the activities used to execute a SQL
	// migration. 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,
	//     },
	//   }
	//
	Migration lifecycle.OnlyOncePolicy `json:"migration"`
	// 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))