storage

package
v0.8.0 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 9 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func Initialize 

func Initialize(db *sql.DB, schema string) error

Initialize executes the schema generated by GenerateSchema. It's a convenience helper for development and testing. For production, it's recommended to use a proper migration tool.

Initialize creates the table only; it does not migrate a pre-existing table or create the due index. Use EnsureSchema for the M4 due index (ListDue / Manager.FireDue), which is idempotent on both fresh and pre-existing tables.

func RunInTx added in v0.8.0 

func RunInTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) (err error)

RunInTx runs fn inside a database transaction. It commits if fn returns nil and rolls back if fn returns an error or panics (the panic is re-raised after the rollback). It lets callers persist workflow state and append history atomically, so a crash mid-transition cannot leave the state and the audit log disagreeing: 

err := storage.RunInTx(ctx, db, func(tx *sql.Tx) error {
    if err := store.SaveStateTx(ctx, tx, id, places, data); err != nil {
        return err
    }
    return hist.SaveTransitionTx(ctx, tx, record)
})

The same *sql.DB must back both stores for their writes to share the transaction.

Note: composing a versioned save manually here (via SaveVersionedStateTx) does NOT maintain the due index — the storage-level Tx saves leave the due column untouched. For a timed definition, drive saves through the Manager, or use SaveVersionedStateInTxWithDue, so the due index commits atomically with state.

Types

type Option 

type Option func(*config)

Option configures a SQL-backed storage implementation (SQLite or Postgres).

func WithContextColumn added in v0.8.0 

func WithContextColumn(name string) Option

WithContextColumn sets the name of the column that persists the workflow's full context map as a single JSON document. Default: "context".

Pass an empty name to disable it (e.g. for a pre-existing table without the column) — then only the keys configured via WithCustomFields survive a save/load round-trip and every other context key is dropped on save.

func WithCustomFields 

func WithCustomFields(fields map[string]string) Option

WithCustomFields defines the schema for additional application-specific data. The map key is the key used in the workflow's context map; the value is the full SQL column definition (e.g. "title TEXT", "amount INTEGER NOT NULL").

func WithDueColumn added in v0.8.0 

func WithDueColumn(name string) Option

WithDueColumn sets the name of the column that stores each instance's next-due time — the maintained index behind DueStorage.ListDue and Manager.FireDue. Default: "due_at".

Pass an empty name to disable the due index (e.g. for a pre-existing table you do not want to migrate). The backend then no longer advertises DueStorage, so the Manager keeps working but a fleet timer scan is unavailable for it.

func WithIDColumn 

func WithIDColumn(name string) Option

WithIDColumn sets the name of the column used to store the workflow ID. Default: "id".

func WithStateColumn 

func WithStateColumn(name string) Option

WithStateColumn sets the name of the column used to store the workflow's current places (state). Default: "state".

func WithTable 

func WithTable(name string) Option

WithTable sets the name of the table used to store workflow state. Default: "workflow_states".

func WithVersionColumn added in v0.8.0 

func WithVersionColumn(name string) Option

WithVersionColumn sets the name of the optimistic-concurrency version column. Default: "version".

type PostgresStorage added in v0.8.0 

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

PostgresStorage is a PostgreSQL-backed implementation of workflow.Storage and workflow.VersionedStorage. It mirrors SQLiteStorage but uses PostgreSQL syntax ($N placeholders and INSERT ... ON CONFLICT upserts).

Use it with any database/sql driver that speaks PostgreSQL; the pgx stdlib adapter is recommended: 

import _ "github.com/jackc/pgx/v5/stdlib"
db, _ := sql.Open("pgx", dsn)
store, _ := storage.NewPostgresStorage(db)
_ = storage.Initialize(db, store.GenerateSchema())

func NewPostgresStorage added in v0.8.0 

func NewPostgresStorage(db *sql.DB, opts ...Option) (*PostgresStorage, error)

NewPostgresStorage creates a new PostgresStorage with the given options.

func (*PostgresStorage) DeleteState added in v0.8.0 

func (s *PostgresStorage) DeleteState(ctx context.Context, id string) error

DeleteState removes a workflow's state.

func (*PostgresStorage) DeleteStateTx added in v0.8.0 

func (s *PostgresStorage) DeleteStateTx(ctx context.Context, tx *sql.Tx, id string) error

DeleteStateTx behaves like DeleteState but writes through the provided transaction.

func (*PostgresStorage) EnsureSchema added in v0.8.0 

func (s *PostgresStorage) EnsureSchema(ctx context.Context) error

EnsureSchema creates the state table if it does not exist and idempotently applies the migrations this library version needs — currently the M4 due index: it adds the due column to a pre-existing table (ADD COLUMN IF NOT EXISTS) and creates the supporting index. It is safe to call on every process start against both fresh and pre-existing tables, and is the recommended one-call setup for backends that use Manager.FireDue.

func (*PostgresStorage) GenerateSchema added in v0.8.0 

func (s *PostgresStorage) GenerateSchema() string

GenerateSchema returns the CREATE TABLE statement for the state table.

func (*PostgresStorage) ListDue added in v0.8.0 

func (s *PostgresStorage) ListDue(ctx context.Context, before time.Time, limit int) ([]string, error)

ListDue implements workflow.DueStorage, returning the IDs of instances whose stored next-due time is non-null and at or before `before`, ordered by due time ascending then by ID. A zero limit means no limit.

func (*PostgresStorage) ListIDs added in v0.8.0 

func (s *PostgresStorage) ListIDs(ctx context.Context, opts workflow.ListOptions) ([]string, error)

ListIDs implements workflow.ListableStorage, returning persisted workflow IDs ordered by ID for stable pagination. A zero opts.Limit means no limit.

func (*PostgresStorage) LoadState added in v0.8.0 

func (s *PostgresStorage) LoadState(ctx context.Context, id string) (workflow.Marking, map[string]any, error)

LoadState loads the workflow's marking and custom fields.

func (*PostgresStorage) LoadStateTx added in v0.8.0 

func (s *PostgresStorage) LoadStateTx(ctx context.Context, tx *sql.Tx, id string) (workflow.Marking, map[string]any, error)

LoadStateTx behaves like LoadState but reads through the provided transaction.

func (*PostgresStorage) LoadVersionedState added in v0.8.0 

func (s *PostgresStorage) LoadVersionedState(ctx context.Context, id string) (workflow.Marking, map[string]any, int64, error)

LoadVersionedState implements workflow.VersionedStorage.

func (*PostgresStorage) SaveState added in v0.8.0 

func (s *PostgresStorage) SaveState(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any) error

SaveState upserts the workflow's places and custom fields (last write wins). It does not maintain the due index (the due column is left untouched); for a timed definition, save through the Manager or SaveVersionedStateWithDue so the index stays consistent.

func (*PostgresStorage) SaveStateTx added in v0.8.0 

func (s *PostgresStorage) SaveStateTx(ctx context.Context, tx *sql.Tx, id string, marking workflow.Marking, ctxData map[string]any) error

SaveStateTx behaves like SaveState but writes through the provided transaction.

func (*PostgresStorage) SaveVersionedState added in v0.8.0 

func (s *PostgresStorage) SaveVersionedState(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64) (int64, error)

SaveVersionedState implements workflow.VersionedStorage. It preserves the due column (untouched on update, NULL on insert) but does not maintain the due index; for a timed definition, use SaveVersionedStateWithDue or go through the Manager so the index stays current.

func (*PostgresStorage) SaveVersionedStateInTx added in v0.8.0 

func (s *PostgresStorage) SaveVersionedStateInTx(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, effects ...workflow.TxSideEffect) (int64, error)

SaveVersionedStateInTx implements workflow.TransactionalStorage: the versioned save and every side effect run in one transaction, committing only if all succeed. Effects receive the *sql.Tx (as an any). It does not maintain the due index; for a timed definition use SaveVersionedStateInTxWithDue so the index commits atomically with state and effects.

func (*PostgresStorage) SaveVersionedStateInTxWithDue added in v0.8.0 

func (s *PostgresStorage) SaveVersionedStateInTxWithDue(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, due *time.Time, effects ...workflow.TxSideEffect) (int64, error)

SaveVersionedStateInTxWithDue implements workflow.TransactionalDueStorage: the versioned save, the due-index update, and every side effect run in one transaction, committing only if all succeed.

func (*PostgresStorage) SaveVersionedStateTx added in v0.8.0 

func (s *PostgresStorage) SaveVersionedStateTx(ctx context.Context, tx *sql.Tx, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64) (int64, error)

SaveVersionedStateTx behaves like SaveVersionedState but writes through the provided transaction. Like SaveVersionedState it does not maintain the due index; for a timed definition composed manually into a transaction, use SaveVersionedStateInTxWithDue (or the Manager) so the due index commits with it.

func (*PostgresStorage) SaveVersionedStateWithDue added in v0.8.0 

func (s *PostgresStorage) SaveVersionedStateWithDue(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, due *time.Time) (int64, error)

SaveVersionedStateWithDue implements workflow.DueStorage: it saves the versioned state and records the instance's next-due time (nil clears it) in the due-index column.

type SQLiteStorage 

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

SQLiteStorage provides a persistent storage implementation using SQLite. It is highly configurable to allow for custom table and column names, as well as storing arbitrary application-specific data alongside the workflow state. It implements workflow.Storage and workflow.VersionedStorage.

func NewSQLiteStorage 

func NewSQLiteStorage(db *sql.DB, opts ...Option) (*SQLiteStorage, error)

NewSQLiteStorage creates a new SQLiteStorage with the given options.

func (*SQLiteStorage) DeleteState 

func (s *SQLiteStorage) DeleteState(ctx context.Context, id string) error

DeleteState removes a workflow's state from the database.

func (*SQLiteStorage) DeleteStateTx added in v0.8.0 

func (s *SQLiteStorage) DeleteStateTx(ctx context.Context, tx *sql.Tx, id string) error

DeleteStateTx behaves like DeleteState but writes through the provided transaction.

func (*SQLiteStorage) EnsureSchema added in v0.8.0 

func (s *SQLiteStorage) EnsureSchema(ctx context.Context) error

EnsureSchema creates the state table if it does not exist and idempotently applies the migrations this library version needs — currently the M4 due index: it adds the due column to a pre-existing table (safely tolerating a table that already has it) and creates the supporting index. It is safe to call on every process start against both fresh and pre-existing tables, and is the recommended one-call setup for backends that use Manager.FireDue.

func (*SQLiteStorage) GenerateSchema 

func (s *SQLiteStorage) GenerateSchema() string

GenerateSchema returns the `CREATE TABLE` SQL statement based on the storage configuration. This allows the user to see, modify, or use the schema with a separate migration tool.

func (*SQLiteStorage) ListDue added in v0.8.0 

func (s *SQLiteStorage) ListDue(ctx context.Context, before time.Time, limit int) ([]string, error)

ListDue implements workflow.DueStorage, returning the IDs of instances whose stored next-due time is non-null and at or before `before`, ordered by due time ascending then by ID. A zero limit means no limit.

func (*SQLiteStorage) ListIDs added in v0.8.0 

func (s *SQLiteStorage) ListIDs(ctx context.Context, opts workflow.ListOptions) ([]string, error)

ListIDs implements workflow.ListableStorage, returning persisted workflow IDs ordered by ID for stable pagination. A zero opts.Limit means no limit.

func (*SQLiteStorage) LoadState 

func (s *SQLiteStorage) LoadState(ctx context.Context, id string) (workflow.Marking, map[string]any, error)

LoadState loads the workflow's marking and all configured custom fields into the context map.

func (*SQLiteStorage) LoadStateTx added in v0.8.0 

func (s *SQLiteStorage) LoadStateTx(ctx context.Context, tx *sql.Tx, id string) (workflow.Marking, map[string]any, error)

LoadStateTx behaves like LoadState but reads through the provided transaction, so it observes the transaction's own uncommitted writes.

func (*SQLiteStorage) LoadVersionedState added in v0.8.0 

func (s *SQLiteStorage) LoadVersionedState(ctx context.Context, id string) (workflow.Marking, map[string]any, int64, error)

LoadVersionedState implements workflow.VersionedStorage. It loads the workflow's marking and context data along with its current optimistic-concurrency version. A never-saved workflow returns workflow.ErrWorkflowNotFound.

func (*SQLiteStorage) SaveState 

func (s *SQLiteStorage) SaveState(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any) error

SaveState saves the workflow's current places and any configured custom fields from its context. It does not maintain the due index (the due column is left untouched); for a timed definition, save through the Manager or SaveVersionedStateWithDue so the index stays consistent.

func (*SQLiteStorage) SaveStateTx added in v0.8.0 

func (s *SQLiteStorage) SaveStateTx(ctx context.Context, tx *sql.Tx, id string, marking workflow.Marking, ctxData map[string]any) error

SaveStateTx behaves like SaveState but writes through the provided transaction, so it can be committed atomically with other writes (e.g. a history record). See RunInTx.

func (*SQLiteStorage) SaveVersionedState added in v0.8.0 

func (s *SQLiteStorage) SaveVersionedState(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64) (int64, error)

SaveVersionedState implements workflow.VersionedStorage. It saves the workflow only if the stored version equals expectedVersion, returning the new version. Pass expectedVersion 0 to create a new workflow. A mismatch returns workflow.ErrConflict.

It preserves the due column (leaves it untouched on update, NULL on insert) but does not maintain the due index; for a timed definition, use SaveVersionedStateWithDue or go through the Manager so the index stays current.

func (*SQLiteStorage) SaveVersionedStateInTx added in v0.8.0 

func (s *SQLiteStorage) SaveVersionedStateInTx(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, effects ...workflow.TxSideEffect) (int64, error)

SaveVersionedStateInTx implements workflow.TransactionalStorage: the versioned save and every side effect run in one transaction, committing only if all succeed. Effects receive the *sql.Tx (as an any). It does not maintain the due index; for a timed definition use SaveVersionedStateInTxWithDue so the index commits atomically with state and effects.

func (*SQLiteStorage) SaveVersionedStateInTxWithDue added in v0.8.0 

func (s *SQLiteStorage) SaveVersionedStateInTxWithDue(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, due *time.Time, effects ...workflow.TxSideEffect) (int64, error)

SaveVersionedStateInTxWithDue implements workflow.TransactionalDueStorage: the versioned save, the due-index update, and every side effect run in one transaction, committing only if all succeed.

func (*SQLiteStorage) SaveVersionedStateTx added in v0.8.0 

func (s *SQLiteStorage) SaveVersionedStateTx(ctx context.Context, tx *sql.Tx, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64) (int64, error)

SaveVersionedStateTx behaves like SaveVersionedState but writes through the provided transaction, so a versioned state change and a history record can be committed atomically. See RunInTx. Like SaveVersionedState it does not maintain the due index; for a timed definition composed manually into a transaction, use SaveVersionedStateInTxWithDue (or the Manager) so the due index commits with it.

func (*SQLiteStorage) SaveVersionedStateWithDue added in v0.8.0 

func (s *SQLiteStorage) SaveVersionedStateWithDue(ctx context.Context, id string, marking workflow.Marking, ctxData map[string]any, expectedVersion int64, due *time.Time) (int64, error)

SaveVersionedStateWithDue implements workflow.DueStorage: it saves the versioned state and records the instance's next-due time (nil clears it) in the due-index column.

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.