provisioning

type Executor ¶

type Executor struct {
	Store  StateStore
	Steps  Steps
	Logger logger.Logger
}

Executor advances a persisted Job through the state machine, invoking the consumer-supplied Steps at each forward transition. Run is safe to call against a partially-completed job — the Executor loads the persisted state and resumes from there.

Crash recovery contract: each step MUST be idempotent. If the worker crashes after a step's side effect succeeds but before the Transition is persisted, the next Run call re-invokes the same step. Schema and role creation are idempotent by construction (see migration/roles.go); Migrate is idempotent because Flyway tracks applied versions in flyway_schema_history; Seed is the consumer's responsibility to design idempotently (typical pattern: INSERT ... ON CONFLICT DO NOTHING).

func NewExecutor(store StateStore, steps Steps, log logger.Logger) (*Executor, error)

func (e *Executor) Run(ctx context.Context, jobID string) error

type Job ¶

type Job struct {
	ID        string            // Idempotency key — re-running with the same ID converges on the same end state.
	TenantID  string            // Logical tenant identifier the job provisions.
	State     State             // Current state. Updated by every Transition.
	Attempts  int               // Number of forward-step attempts on the current state, used for retry bookkeeping.
	LastError string            // Most recent step error, surfaced in audit events. Empty on success paths.
	Metadata  map[string]string // Step-specific data that must survive a crash. Consumer-owned shape; see SECURITY note above.
	CreatedAt time.Time         // First persistence timestamp; set by Upsert.
	UpdatedAt time.Time         // Most recent persistence timestamp; updated on every Transition.
}

Job is the persisted record of a single per-tenant provisioning attempt. All fields except Metadata are owned by the StateStore; Metadata is opaque to the framework and gives consumers a place to record step-specific data that must survive a crash (e.g., the applied Flyway version range).

SECURITY: Metadata is persisted in plaintext (JSONB on PostgreSQL). Do not store credentials, secret tokens, or other sensitive material in it — use a secret manager and record only opaque references (e.g. the secret name) when continuity across crashes is needed.

type MemoryStore ¶

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

MemoryStore is an in-process StateStore implementation backed by a map. Used by unit tests and as a reference implementation for vendor authors adding new backends; not intended for production. All operations are goroutine-safe; transitions are serialized by a per-store mutex (not per-job) which is fine for test workloads.

func NewMemoryStore() *MemoryStore

func (*MemoryStore) CreateTable(_ context.Context) error

func (m *MemoryStore) Get(_ context.Context, jobID string) (*Job, error)

func (m *MemoryStore) Transition(
	_ context.Context, jobID string, from, to State, metadata map[string]string, lastError string,
) error

func (m *MemoryStore) Upsert(_ context.Context, job *Job) (*Job, error)

func (m *MemoryStore) WithClock(now func() time.Time) *MemoryStore

type PostgresStore ¶

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

PostgresStore is the PostgreSQL-backed reference implementation of StateStore. Construct via NewPostgresStore so the table name is validated.

func NewPostgresStore(db *sql.DB, tableName string) (*PostgresStore, error)

func (s *PostgresStore) CreateTable(ctx context.Context) error

func (s *PostgresStore) Get(ctx context.Context, jobID string) (*Job, error)

func (s *PostgresStore) Transition(
	ctx context.Context, jobID string, from, to State, metadata map[string]string, lastError string,
) error

func (s *PostgresStore) Upsert(ctx context.Context, job *Job) (*Job, error)

func (s *PostgresStore) WithClock(now func() time.Time) *PostgresStore

type State ¶

type State string

State enumerates the per-tenant provisioning states defined in issue #379. Transitions are validated by the package's static graph (see allowedNext); illegal transitions return ErrIllegalTransition and never silently succeed.

const (
	// StatePending is the initial state when a job is first persisted.
	// The worker has not started any side-effecting work.
	StatePending State = "pending"

	// StateSchemaCreated marks the successful CREATE SCHEMA owned by the
	// migrator role (idempotent via CREATE SCHEMA IF NOT EXISTS).
	StateSchemaCreated State = "schema_created"

	// StateRoleCreated marks the successful creation of the per-tenant
	// runtime role with the locked-down privilege model from #378.
	StateRoleCreated State = "role_created"

	// StateMigrated marks the successful application of all pending Flyway
	// migrations against the tenant schema.
	StateMigrated State = "migrated"

	// StateSeeded marks the successful completion of reference-data seeding
	// and any external-system registrations the consumer wires in.
	StateSeeded State = "seeded"

	// StateReady is the terminal success state. The tenant is fully usable
	// by the runtime service. No further transitions are allowed.
	StateReady State = "ready"

	// StateCleanup is the rollback state entered when any forward step
	// returns an error. The cleanup step drops partially-provisioned
	// resources (schema, role).
	StateCleanup State = "cleanup"

	// StateFailed is the terminal failure state. Retry creates a new job
	// ID rather than reusing the existing one — this keeps the failure
	// trace intact for audit purposes.
	StateFailed State = "failed"
)

func (s State) IsTerminal() bool

type StateStore ¶

type StateStore interface {
	// Get loads the job by ID. Returns ErrJobNotFound when no record exists.
	Get(ctx context.Context, jobID string) (*Job, error)

	// Upsert writes a new job at StatePending if no record exists for jobID,
	// or returns the existing record unchanged if one does. This is the
	// idempotency contract referenced in #379: re-running with the same
	// jobID converges on the same end state without duplicate side effects,
	// because the executor will see the persisted state and resume (or
	// no-op if terminal).
	//
	// Implementations must populate Job.CreatedAt and Job.UpdatedAt on the
	// returned record. State, Attempts, LastError, and Metadata on the
	// passed-in job are used only when the record is newly inserted.
	Upsert(ctx context.Context, job *Job) (*Job, error)

	// Transition advances jobID's State from -> to atomically. Optimistic
	// concurrency: implementations MUST verify the row's current state is
	// `from` before applying the update; if it isn't, return ErrStaleRead
	// without mutating the row.
	//
	// metadata replaces the persisted Metadata in full (not merged) so
	// callers controlling the in-memory job can decide whether to carry
	// step-specific fields forward. lastError replaces the persisted
	// LastError (empty string clears a prior error).
	//
	// The transition is also validated against the static state graph;
	// illegal edges return ErrIllegalTransition.
	Transition(ctx context.Context, jobID string, from, to State, metadata map[string]string, lastError string) error

	// CreateTable creates the persistence table(s) and any required indexes,
	// idempotently. Safe to call on a database where the table already
	// exists. Operators who prefer to manage schema externally can skip
	// this and run the equivalent DDL via their migration tooling — see
	// the implementation's exported DDL constants.
	CreateTable(ctx context.Context) error
}

StateStore persists Job records and enforces optimistic-concurrency transitions. Implementations are typically backed by the same database as the tenant's business data; the framework ships a PostgreSQL reference implementation at NewPostgresStore.

Per ADR-021, this interface mirrors the *patterns* of outbox.Store (idempotent CreateTable, vendor-pluggable, in-memory mock for tests) but the method shape is finite-state-machine-oriented rather than queue-oriented: there is no FetchPending / MarkPublished pair — the dispatcher feeding the executor is out of scope for #379.

type Steps ¶

type Steps struct {
	CreateSchema func(ctx context.Context, job *Job) error
	CreateRole   func(ctx context.Context, job *Job) error
	Migrate      func(ctx context.Context, job *Job) error
	Seed         func(ctx context.Context, job *Job) error
	Cleanup      func(ctx context.Context, job *Job) error
}

Steps groups the consumer-supplied callbacks invoked by the Executor at each forward transition. The CreateSchema step runs to leave StatePending, the CreateRole step runs to leave StateSchemaCreated, and so on. Cleanup runs when any forward step errors; the executor moves the job to StateCleanup, invokes Cleanup, then transitions to StateFailed regardless of the cleanup outcome (cleanup errors are logged but do not block the failure transition).

Each step receives the current Job and may update Job.Metadata in place; the updated Metadata is persisted by the Executor as part of the subsequent Transition call.

func (s *Steps) Validate() error