Typed State Machine Contracts & Runtime — state + stream + forge¶

package state

import "github.com/DaniDeer/go-codex/codex"

// Transition declares a single valid edge in the state graph.
type Transition[K comparable, E comparable] struct {
    From  K
    Event E
    To    K
}

// Spec is the machine's contract — a plain value, declared once, passed anywhere.
// S is the state type, K is the state-key type (comparable), E is the event type.
type Spec[S any, K comparable, E comparable] struct {
    // Codec validates the state shape on every Apply and ValidateState call —
    // the same codex.Codec[S] used elsewhere in the domain layer.
    Codec codex.Codec[S]

    // Key extracts the logical state key from a state value (e.g. order.Status).
    Key func(S) K

    // EventCodec validates the event before any transition lookup.
    // E may be a string, an integer enum, or a structured codec.Codec[E] type.
    EventCodec codex.Codec[E]

    // Transitions is the full transition graph — declare once.
    Transitions []Transition[K, E]

    // ConstraintsByKey holds per-state codex.Constraint[S] rules applied
    // before (source) and after (target) a transition.
    // Reuse any named constraint already declared for other boundaries.
    ConstraintsByKey map[K][]codex.Constraint[S]

    // Order is an optional forward-only rank; when set a transition whose
    // target rank ≤ source rank is rejected as TransitionDirectionError.
    Order map[K]int
}

type Machine[S any, K comparable, E comparable] struct { /* unexported */ }

// NewMachine compiles spec into an immutable, shareable Machine value.
// Returns SpecError on invalid declaration.
func NewMachine[S any, K comparable, E comparable](spec Spec[S, K, E]) (*Machine[S, K, E], error)

// ValidateState validates s against the state codec AND the per-key constraints.
func (m *Machine[S, K, E]) ValidateState(s S) error

// Can reports whether event is a valid transition from s's current key.
func (m *Machine[S, K, E]) Can(event E, s S) bool

// Apply validates the event (via EventCodec), resolves the transition,
// validates source-state constraints, calls mutate, then validates
// the returned state (codec + target-key constraints).
// One call — fully validated in, fully validated out.
func (m *Machine[S, K, E]) Apply(event E, s S, mutate func(cur S, to K) (S, error)) (S, error)

package state

type ReduceOptions struct {
    Name     string        // machine name for observer reporting
    Observer stats.Observer
}

// Reduce folds a stream of events onto an initial state using m.Apply.
// Each successfully-transitioned state is emitted downstream; transition
// or validation failures are sent to the stream's error channel.
func Reduce[S any, K comparable, E comparable](
    ctx context.Context,
    src stream.Stream[E],
    initial S,
    m *Machine[S, K, E],
    mutate func(cur S, evt E, to K) (S, error),
    opts ReduceOptions,
) stream.Stream[S]

// NewTransitionFunction wraps a Machine as a named forge.Function[E, S].
// The current state is carried in a closure (suitable for single-writer
// sequential pipelines). Reuse the same forge composition vocabulary:
// forge.Compose, forge.Registry.Register, forge.Topology.WithPort.
func NewTransitionFunction[S any, K comparable, E comparable](
    name string,
    initial S,
    m *Machine[S, K, E],
    mutate func(cur S, evt E, to K) (S, error),
) forge.Function[E, S]

// domain/order.go — declare the contract as a package-level value.

var orderMachine = codex.Must(state.NewMachine(state.Spec[Order, OrderStatus, OrderEvent]{
    Codec:      OrderCodec,         // already declared for REST/events/cache
    EventCodec: codex.String().Refine(validate.NonEmpty),
    Key:        func(o Order) OrderStatus { return o.Status },
    Transitions: []state.Transition[OrderStatus, OrderEvent]{
        {From: StatusPending,  Event: EventPay,    To: StatusPaid},
        {From: StatusPaid,     Event: EventShip,   To: StatusShipped},
        {From: StatusShipped,  Event: EventDeliver, To: StatusDelivered},
    },
    // Reuse the SAME constraints already on your REST route or forge function:
    ConstraintsByKey: map[OrderStatus][]codex.Constraint[Order]{
        StatusPaid:    {OrderPaidConstraint},      // same constraint as the payment forge fn
        StatusShipped: {OrderShippedConstraint},
    },
    Order: map[OrderStatus]int{
        StatusPending: 0, StatusPaid: 1, StatusShipped: 2, StatusDelivered: 3,
    },
}))

// pipeline/order_pipeline.go — apply in a stream pipeline.
func BuildOrderPipeline(ctx context.Context, events stream.Stream[OrderEvent], initial Order) stream.Stream[Order] {
    return state.Reduce(ctx, events, initial, orderMachine, applyOrderMutation,
        state.ReduceOptions{Name: "order", Observer: stats.ObserverFromContext(ctx)})
}

// handler/order_handler.go — apply directly in a REST handler.
func handleOrderEvent(ctx context.Context, req TransitionReq) (Order, error) {
    return orderMachine.Apply(req.Event, req.Current, func(cur Order, to OrderStatus) (Order, error) {
        cur.Status = to
        cur.UpdatedAt = time.Now()
        return cur, nil
    })
}

// forge integration — same machine, wrapped as a named forge Function.
var orderTransitionFn = forge.NewTransitionFunction("order.transition", Order{}, orderMachine, applyOrderMutation)

type StateObserver interface {
    RecordTransition(machine string, from, event, to string, success bool, d time.Duration)
}