changeset

Documentation

Overview

Package changeset batches value mutations into a reviewable draft that leaves live data untouched until it is published. A change-set moves draft → in_review → approved → published (or rejected); publish applies every mutation in one unit of work — atomic, with full events and activity, exactly like direct writes. An optional publish_at defers the apply to a scheduler.

Index

• Constants
• func ErrStaleVersion(id string, version int) error
• type ChangeSet
• type ClaimReclaimer
• type CreateInput
• type Interactor
  • func NewInteractor(store Store, values *appvalue.Interactor, attrs domainattribute.Repository, ...) *Interactor
  • func (i *Interactor) AddMutation(ctx context.Context, rawID string, m appvalue.Mutation) (*ChangeSet, error)
  • func (i *Interactor) Approve(ctx context.Context, rawID string) (*ChangeSet, error)
  • func (i *Interactor) Create(ctx context.Context, in CreateInput) (*ChangeSet, error)
  • func (i *Interactor) Get(ctx context.Context, rawID string) (*ChangeSet, error)
  • func (i *Interactor) List(ctx context.Context) ([]ChangeSet, error)
  • func (i *Interactor) OnPublishFailure(fn func(cs ChangeSet, err error))
  • func (i *Interactor) Publish(ctx context.Context, rawID string) (*ChangeSet, error)
  • func (i *Interactor) PublishDue(ctx context.Context) (int, error)
  • func (i *Interactor) Reject(ctx context.Context, rawID string) (*ChangeSet, error)
  • func (i *Interactor) Submit(ctx context.Context, rawID string) (*ChangeSet, error)
• type State
• type Store

Constants

const PublishClaimTTL = 15 * time.Minute

PublishClaimTTL bounds how long a publish claim is honoured. A publish claims the set (state publishing) before it applies the mutations, so a request that ends mid-publish — a client timeout, a load-balancer idle timeout, a pod eviction — could leave the claim behind. Once the claim is this old, the set is publishable again: the scheduler reclaims it, and so does an explicit publish.

The bound is generous because it is an upper bound on one publish. A set still being published must never be reclaimed under the transaction that holds it.

Functions

func ErrStaleVersion

func ErrStaleVersion(id string, version int) error

ErrStaleVersion is the conflict a store returns when a change-set was modified between the caller's read and its write. The caller re-reads and re-applies; nothing is written from a stale view.

Types

type ChangeSet

type ChangeSet struct {
	ID       ulid.ID               `json:"id"`
	TenantID valueobjects.TenantID `json:"tenant_id"`
	Name     string                `json:"name"`
	State    State                 `json:"state"`
	// RequireApproval demands an approver distinct from the author before
	// the set may publish.
	RequireApproval bool                `json:"require_approval"`
	Author          string              `json:"author,omitempty"`
	Approver        string              `json:"approver,omitempty"`
	Mutations       []appvalue.Mutation `json:"mutations"`
	PublishAt       *time.Time          `json:"publish_at,omitempty"`
	CreatedAt       time.Time           `json:"created_at"`
	UpdatedAt       time.Time           `json:"updated_at"`
	PublishedAt     *time.Time          `json:"published_at,omitempty"`
	// Version increments on every mutation and guards against a lost update.
	// Without it, two reviewers editing one set overwrote each other's
	// mutations, and an edit that raced an approval wrote the pre-approval
	// state back — silently reverting the approval. Every other aggregate in
	// this repository takes a row lock; change-sets take this instead,
	// because a review artifact is edited over minutes, not milliseconds, and
	// a conflict should be reported to the reviewer rather than serialized
	// behind a lock they cannot see.
	Version int `json:"version"`
}

ChangeSet is a named, reviewable batch of value mutations.

type ClaimReclaimer

type ClaimReclaimer interface {
	// StalePublishing returns publishing change-sets last updated at or
	// before the cutoff, across all tenants.
	StalePublishing(ctx context.Context, before time.Time) ([]ChangeSet, error)
}

ClaimReclaimer reports change-sets stranded in the publishing state, so the scheduler can retry them. A store that does not implement it keeps working: a stranded set is then recovered by publishing it again through the API, which is allowed once the claim is stale.

It is a separate interface rather than a Store method so an embedder's own store still satisfies Store (see docs/api-stability.md).

type CreateInput

type CreateInput struct {
	Name            string
	RequireApproval bool
	PublishAt       *time.Time
}

CreateInput carries a new change-set's fields.

type Interactor

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

Interactor implements the change-management usecases.

func NewInteractor

func NewInteractor(store Store, values *appvalue.Interactor, attrs domainattribute.Repository, now func() time.Time) *Interactor

NewInteractor wires the change-set usecases.

func (*Interactor) AddMutation

func (i *Interactor) AddMutation(ctx context.Context, rawID string, m appvalue.Mutation) (*ChangeSet, error)

AddMutation appends a value mutation to a draft change-set.

func (*Interactor) Approve

func (i *Interactor) Approve(ctx context.Context, rawID string) (*ChangeSet, error)

Approve marks an in-review change-set approved. When approval is required, the approver must differ from the author.

func (*Interactor) Create

func (i *Interactor) Create(ctx context.Context, in CreateInput) (*ChangeSet, error)

Create opens a draft change-set authored by the calling actor.

func (*Interactor) Get

func (i *Interactor) Get(ctx context.Context, rawID string) (*ChangeSet, error)

Get returns one change-set.

func (*Interactor) List

func (i *Interactor) List(ctx context.Context) ([]ChangeSet, error)

List returns the tenant's change-sets, newest first, with the field ACL applied to every mutation.

func (*Interactor) OnPublishFailure

func (i *Interactor) OnPublishFailure(fn func(cs ChangeSet, err error))

OnPublishFailure registers the observer for a scheduled publish failure. Wire it during composition.

func (*Interactor) Publish

func (i *Interactor) Publish(ctx context.Context, rawID string) (*ChangeSet, error)

Publish applies every mutation atomically and marks the set published. It requires approval when the set demands it.

The set is CLAIMED first (state publishing, version bumped), then the mutations are applied, then the claim is finalised. A constraint failure rolls the whole batch back and the claim is handed back, so the set is publishable again once the cause is fixed. A concurrent edit loses the race while the data is still untouched, rather than after it has been written.

func (*Interactor) PublishDue

func (i *Interactor) PublishDue(ctx context.Context) (int, error)

PublishDue publishes every approved change-set whose publish_at has arrived. It is the scheduler's tick; each set publishes in its own tenant context so events and activity attribute correctly. Returns how many published.

func (*Interactor) Reject

func (i *Interactor) Reject(ctx context.Context, rawID string) (*ChangeSet, error)

Reject closes a change-set without publishing; live data is untouched.

func (*Interactor) Submit

func (i *Interactor) Submit(ctx context.Context, rawID string) (*ChangeSet, error)

Submit moves a draft into review.

type State

type State string

State is a change-set's lifecycle stage.

const (
	StateDraft    State = "draft"
	StateInReview State = "in_review"
	StateApproved State = "approved"
	// StatePublishing is held while the mutations are being applied.
	//
	// It exists so the claim can be taken BEFORE the side effects. Publish
	// used to apply the mutations and only then compare-and-swap the record:
	// once optimistic locking made that second call able to fail, any
	// concurrent touch of the set — a reviewer rejecting it, a second
	// publish, the scheduler tick — left the data committed and the record
	// saying something else. Through PublishDue it compounded: the set stayed
	// approved with publish_at in the past, so every tick re-applied the same
	// mutations over whatever had been written in between.
	//
	// A set left in this state means a publish began and did not finish. The
	// scheduler does not pick it up (it selects approved), so it is visible
	// and inert rather than silently repeating.
	StatePublishing State = "publishing"
	StatePublished  State = "published"
	StateRejected   State = "rejected"
)

The change-set lifecycle states.

type Store

type Store interface {
	Create(ctx context.Context, cs ChangeSet) error
	Get(ctx context.Context, tenant valueobjects.TenantID, id ulid.ID) (ChangeSet, error)
	List(ctx context.Context, tenant valueobjects.TenantID) ([]ChangeSet, error)
	// Update persists the set only if the stored version still matches
	// cs.Version, and returns a conflict otherwise. It increments the stored
	// version on success.
	Update(ctx context.Context, cs ChangeSet) error
	// DueForPublish returns approved change-sets whose publish_at has
	// arrived, across all tenants (the scheduler runs outside a request).
	DueForPublish(ctx context.Context, now time.Time) ([]ChangeSet, error)
}

Store persists change-sets, scoped by tenant.