ago

package module
v0.0.1 Latest Latest
Warning

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

 Go to latest Published: Dec 11, 2025 License: MIT Imports: 11 Imported by: 0

README

ago

CI Status codecov Go Report Card CodeQL Go Reference License Go Version Release

Event-driven orchestration primitives for Go.

ago ("I do" in Latin) bridges capitan events with pipz pipelines, enabling distributed sagas, request/response patterns, and stateful coordination across processes.

Installation

go get github.com/zoobzio/ago

Requirements: Go 1.24+

Core Concepts

Flow

Flow[T] wraps a typed payload with correlation context and accumulated state:

// Create a flow from an event
flow := ago.NewFromEvent(event, orderKey)

// Or create directly
flow := ago.NewFlow(order, orderCreated)
flow.CorrelationID = "order-123"

All primitives implement pipz.Chainable[*Flow[T]], composable via pipz topologies.

Correlation

Flows carry correlation and causation IDs for distributed tracing:

pipeline := ago.Sequence("process-order",
    ago.Correlate[Order]("correlate"),           // Generate correlation ID
    ago.SagaStep(...).Build(),
    ago.Emit[Order](...).Build(),
)

Primitives

Saga Orchestration

Execute distributed transactions with automatic compensation on failure:

// Define signals
var (
    reserveInventory   = capitan.NewSignal("inventory.reserve", "Reserve inventory")
    releaseInventory   = capitan.NewSignal("inventory.release", "Release inventory")
    chargePayment      = capitan.NewSignal("payment.charge", "Charge payment")
    refundPayment      = capitan.NewSignal("payment.refund", "Refund payment")
)

// Build saga pipeline
pipeline := ago.Sequence("order-saga",
    ago.Correlate[Order]("correlate"),

    ago.NewSagaStep[Order](
        "reserve-inventory",
        store,
        orderKey,
        reserveInventory,    // Execute signal
        releaseInventory,    // Compensate signal
    ).WithTimeout(5 * time.Minute).Build(),

    ago.NewSagaStep[Order](
        "charge-payment",
        store,
        orderKey,
        chargePayment,
        refundPayment,
    ).Build(),
)

// On failure, trigger compensation
compensate := ago.NewCompensate[Order]("compensate", store, orderKey).Build()

Sagas provide:

  • Compensation registered atomically before execution
  • LIFO rollback order
  • Idempotency via compensation records
  • Crash recovery via RecoverSagas
  • Configurable timeouts
Request/Response

Synchronous request/response over async events:

request := ago.NewRequest[Order, PaymentResult](
    "charge-card",
    chargeRequest,      // Request signal
    chargeResponse,     // Response signal
    orderKey,           // Request payload key
    paymentResultKey,   // Response payload key
).Timeout(30 * time.Second).Build()

// Sends request, waits for correlated response
flow, err := request.Process(ctx, flow)
result, _ := ago.From(flow, paymentResultKey)
Await

Wait for a correlated event:

await := ago.NewAwait[Order, ShippingStatus](
    "await-shipment",
    shippingUpdated,
    shippingStatusKey,
).Timeout(24 * time.Hour).Build()
Emit

Fire-and-forget event emission:

emit := ago.NewEmit[Order](
    "emit-created",
    orderCreated,
    orderKey,
).Build()
Enrichment

Augment flows with external data:

// Fails on error
enrich := ago.Enrich[Order, Customer](
    "fetch-customer",
    customerKey,
    func(ctx context.Context, order Order) (Customer, error) {
        return customerService.Get(ctx, order.CustomerID)
    },
)

// Logs error but continues
enrichOptional := ago.EnrichOptional[Order, Discount](
    "fetch-discount",
    discountKey,
    fetchDiscount,
)
Integration

Publish to message brokers:

publish := ago.Publish[Order]("publish-order", kafkaProvider)

Route failures to dead letter queue:

deadLetter := ago.NewDeadLetter[Order]("dlq", orderKey).
    WithProvider(dlqProvider).
    Build()

Flow Control

ago provides typed wrappers around pipz flow control:

pipeline := ago.Sequence("order-pipeline",
    // Retry with backoff
    ago.Backoff("retry-payment", paymentStep, 3, time.Second),

    // Circuit breaker
    ago.CircuitBreaker("inventory-breaker", inventoryStep, 5, time.Minute),

    // Rate limiting
    ago.RateLimiter[Order]("rate-limit", 100, 10),

    // Timeout
    ago.Timeout("shipping-timeout", shippingStep, 30*time.Second),

    // Fallback on failure
    ago.Fallback("payment-fallback", primaryPayment, backupPayment),

    // Parallel execution
    ago.Concurrent("parallel-notify",
        func(original *ago.Flow[Order], results map[pipz.Name]*ago.Flow[Order], errors map[pipz.Name]error) *ago.Flow[Order] {
            return original
        },
        emailNotify,
        smsNotify,
        pushNotify,
    ),
)

Storage

ago requires a Store for saga state and coordination:

// In-memory (testing)
store := ago.NewMemoryStore()

// PostgreSQL (production)
store := ago.NewCerealStore(db)
store.Migrate(ctx) // Create tables

Recovery

Handle crashed or timed-out sagas:

// Run periodically
recovered, err := ago.RecoverSagas(ctx, store, orderKey, capitan.Default())
for _, state := range recovered {
    log.Printf("Recovered saga %s", state.CorrelationID)
}

Example: Order Processing

package main

import (
    "context"
    "time"

    "github.com/zoobzio/ago"
    "github.com/zoobzio/capitan"
    "github.com/zoobzio/pipz"
)

type Order struct {
    ID         string
    CustomerID string
    Total      float64
}

var (
    // Signals
    orderCreated     = capitan.NewSignal("order.created", "Order created")
    inventoryReserve = capitan.NewSignal("inventory.reserve", "Reserve inventory")
    inventoryRelease = capitan.NewSignal("inventory.release", "Release inventory")
    paymentCharge    = capitan.NewSignal("payment.charge", "Charge payment")
    paymentRefund    = capitan.NewSignal("payment.refund", "Refund payment")

    // Keys
    orderKey = capitan.NewKey[Order]("order", "app.Order")
)

func main() {
    store := ago.NewMemoryStore()

    // Build order processing pipeline
    pipeline := ago.Sequence("process-order",
        ago.Correlate[Order]("correlate"),

        ago.NewSagaStep[Order]("reserve", store, orderKey,
            inventoryReserve, inventoryRelease,
        ).WithTimeout(5 * time.Minute).Build(),

        ago.NewSagaStep[Order]("charge", store, orderKey,
            paymentCharge, paymentRefund,
        ).Build(),

        ago.NewEmit[Order]("emit-created", orderCreated, orderKey).Build(),
    )

    // Process an order
    flow := ago.NewFlow(Order{ID: "ORD-123", Total: 99.99}, orderCreated)
    result, err := pipeline.Process(context.Background(), flow)
    if err != nil {
        // Trigger compensation
        compensate := ago.NewCompensate[Order]("compensate", store, orderKey)
        compensate.Process(context.Background(), flow)
    }

    capitan.Shutdown()
}

Testing

Run tests:

make test

Run integration tests:

make test-integration

Run with coverage:

make coverage

Contributing

Contributions welcome! See CONTRIBUTING.md for guidelines.

License

MIT License - see LICENSE file for details.

Documentation

Overview

Package ago provides event-driven pattern primitives for pipz.

ago ("I do" in Latin) bridges capitan events with pipz pipelines, enabling distributed sagas, request/response patterns, and stateful coordination across processes.

Flow[T] wraps a typed payload with correlation context and accumulated state. All primitives implement pipz.Chainable[*Flow[T]], composable via pipz topology or flume schema configuration.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// CorrelationKey identifies related events across services.
	CorrelationKey = capitan.NewStringKey("correlation_id")

	// CausationKey identifies the direct parent event.
	CausationKey = capitan.NewStringKey("causation_id")

	// IdempotencyKey provides a deterministic key for downstream handlers
	// to ensure exactly-once execution with external systems.
	//
	// IMPORTANT: Signal handlers MUST use this key when calling external systems
	// (databases, APIs, payment processors, etc.) to ensure idempotent operations.
	// ago guarantees the key is unique per step execution, but handlers are
	// responsible for using it appropriately.
	//
	// The key format is "{correlationID}:{stepName}" for execution signals
	// and "{correlationID}:compensate:{stepName}" for compensation signals.
	//
	// Example usage in a handler:
	//
	//	c.Hook(chargePayment, func(ctx context.Context, e *capitan.Event) {
	//		idempotencyKey, _ := ago.IdempotencyKey.From(e)
	//		// Use idempotencyKey with your payment processor
	//		paymentService.Charge(ctx, amount, idempotencyKey)
	//	})
	//
	// This is critical because ago may emit duplicate signals in edge cases
	// (e.g., store failures during idempotency marking). The IdempotencyKey
	// ensures external systems see each operation exactly once.
	IdempotencyKey = capitan.NewStringKey("idempotency_key")
)

Common keys for correlation and causation in distributed flows.

View Source 
var (
	FlowCreated   = capitan.NewSignal("ago.flow.created", "Flow created")
	FlowCompleted = capitan.NewSignal("ago.flow.completed", "Flow completed")
	FlowFailed    = capitan.NewSignal("ago.flow.failed", "Flow failed")
)

Flow lifecycle signals.

View Source 
var (
	SagaStarted       = capitan.NewSignal("ago.saga.started", "Saga started")
	SagaStepCompleted = capitan.NewSignal("ago.saga.step.completed", "Saga step completed")
	SagaCompensating  = capitan.NewSignal("ago.saga.compensating", "Saga compensating")
	SagaCompleted     = capitan.NewSignal("ago.saga.completed", "Saga completed")
	SagaFailed        = capitan.NewSignal("ago.saga.failed", "Saga failed")
)

Saga lifecycle signals.

View Source 
var (
	RequestSent      = capitan.NewSignal("ago.request.sent", "Request sent")
	ResponseReceived = capitan.NewSignal("ago.response.received", "Response received")
	RequestTimeout   = capitan.NewSignal("ago.request.timeout", "Request timed out")
)

Request/response signals.

View Source 
var (
	StepNameKey   = capitan.NewStringKey("step_name")
	SagaStatusKey = capitan.NewKey[SagaStatus]("saga_status", "ago.SagaStatus")
	ErrorKey      = capitan.NewErrorKey("error")
)

Common keys for signal payloads.

View Source 
var (
	DeadLetterRouted = capitan.NewSignal("ago.deadletter.routed", "Message routed to dead letter")
)

Dead letter signals.

View Source 
var ErrNotFound = errors.New("ago: state not found")

ErrNotFound indicates the requested state was not found.

View Source 
var ErrTimeout = errors.New("ago: request timeout")

ErrTimeout indicates a request timed out waiting for response.

Functions

func Backoff 

func Backoff[T any](name string, processor pipz.Chainable[*Flow[T]], maxAttempts int, baseDelay time.Duration) *pipz.Backoff[*Flow[T]]

Backoff creates a processor that retries with exponential backoff.

func CircuitBreaker 

func CircuitBreaker[T any](name string, processor pipz.Chainable[*Flow[T]], failureThreshold int, resetTimeout time.Duration) *pipz.CircuitBreaker[*Flow[T]]

CircuitBreaker creates a processor that prevents cascade failures.

func Concurrent 

func Concurrent[T any](name string, reducer func(original *Flow[T], results map[pipz.Name]*Flow[T], errors map[pipz.Name]error) *Flow[T], processors ...pipz.Chainable[*Flow[T]]) *pipz.Concurrent[*Flow[T]]

Concurrent runs all processors in parallel and returns the original flow.

func Correlate 

func Correlate[T any](name pipz.Name) pipz.Chainable[*Flow[T]]

Correlate ensures CorrelationID exists on the flow, generating one if missing.

func CorrelateFrom 

func CorrelateFrom[T any](name pipz.Name, parentCorrelation string) pipz.Chainable[*Flow[T]]

CorrelateFrom sets both CorrelationID and CausationID from a parent. If the flow has no correlation, a new one is generated.

func Do 

func Do[T any](name string, fn func(context.Context, *Flow[T]) (*Flow[T], error)) pipz.Processor[*Flow[T]]

Do creates a processor from a custom function that can fail.

func Effect 

func Effect[T any](name string, fn func(context.Context, *Flow[T]) error) pipz.Processor[*Flow[T]]

Effect creates a processor that performs a side effect without modifying the flow.

func Enrich 

func Enrich[T, V any](name pipz.Name, key capitan.GenericKey[V], enrichFn func(context.Context, T) (V, error)) pipz.Chainable[*Flow[T]]

Enrich fetches external data and adds it to the flow's fields. The enrichFn receives the payload and returns a field to add.

func EnrichOptional 

func EnrichOptional[T, V any](name pipz.Name, key capitan.GenericKey[V], enrichFn func(context.Context, T) (V, error)) pipz.Chainable[*Flow[T]]

EnrichOptional fetches external data, logging but not failing on errors.

func EnrichWith 

func EnrichWith[T any](name string, fn func(context.Context, *Flow[T]) (*Flow[T], error)) pipz.Processor[*Flow[T]]

EnrichWith creates a processor that optionally enhances a flow. Unlike Do, errors are logged but don't stop the pipeline.

func Fallback 

func Fallback[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Fallback[*Flow[T]]

Fallback creates a processor that tries alternatives on failure.

func Filter 

func Filter[T any](name string, predicate func(context.Context, *Flow[T]) bool, processor pipz.Chainable[*Flow[T]]) *pipz.Filter[*Flow[T]]

Filter creates a conditional processor that either processes or passes through.

func From 

func From[T, V any](f *Flow[T], key capitan.GenericKey[V]) (V, bool)

From extracts a typed value from the flow's accumulated state. Returns the value and true if present, or zero value and false otherwise.

func Gate 

func Gate[T any](name string, predicate func(context.Context, *Flow[T]) bool) pipz.Processor[*Flow[T]]

Gate creates a simple pass/fail filter.

func Handle 

func Handle[T any](name string, processor pipz.Chainable[*Flow[T]], errorHandler pipz.Chainable[*pipz.Error[*Flow[T]]]) *pipz.Handle[*Flow[T]]

Handle creates a processor that handles errors without stopping the pipeline.

func Mutate 

func Mutate[T any](name string, fn func(context.Context, *Flow[T]) *Flow[T], predicate func(context.Context, *Flow[T]) bool) pipz.Processor[*Flow[T]]

Mutate creates a processor that conditionally modifies a flow.

func Publish 

func Publish[T any](name pipz.Name, provider herald.Provider) pipz.Chainable[*Flow[T]]

Publish publishes the flow's payload to a message broker via herald.

func Race 

func Race[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Race[*Flow[T]]

Race runs all processors in parallel and returns the first successful result.

func RateLimiter 

func RateLimiter[T any](name string, requestsPerSecond float64, burst int) *pipz.RateLimiter[*Flow[T]]

RateLimiter creates a processor that enforces rate limits.

func RecoverSagas 

func RecoverSagas[T any](ctx context.Context, store Store, key capitan.GenericKey[T], c *capitan.Capitan) error

RecoverSagas finds incomplete sagas and runs their compensations. This includes: - Sagas left in "running" or "compensating" state (from crashes) - Sagas that have exceeded their timeout

Call this at startup to recover from crashes or restarts.

func Retry 

func Retry[T any](name string, processor pipz.Chainable[*Flow[T]], maxAttempts int) *pipz.Retry[*Flow[T]]

Retry creates a processor that retries on failure up to maxAttempts times.

func Sequence 

func Sequence[T any](name string, processors ...pipz.Chainable[*Flow[T]]) *pipz.Sequence[*Flow[T]]

Sequence creates a sequential pipeline of flow processors.

func Switch 

func Switch[T any, K comparable](name string, condition func(context.Context, *Flow[T]) K) *pipz.Switch[*Flow[T], K]

Switch creates a router that directs flows to different processors.

func Tag 

func Tag[T any](name pipz.Name, key, value string) pipz.Chainable[*Flow[T]]

Tag adds a key/value pair to the flow's broker metadata.

func TagFrom 

func TagFrom[T any](name pipz.Name, key string, valueFn func(T) string) pipz.Chainable[*Flow[T]]

TagFrom adds a key/value pair where the value comes from a function.

func Timeout 

func Timeout[T any](name string, processor pipz.Chainable[*Flow[T]], duration time.Duration) *pipz.Timeout[*Flow[T]]

Timeout creates a processor that enforces a time limit on execution.

func Transform 

func Transform[T any](name string, fn func(context.Context, *Flow[T]) *Flow[T]) pipz.Processor[*Flow[T]]

Transform creates a processor from a pure transformation function.

func WorkerPool 

func WorkerPool[T any](name string, workers int, processors ...pipz.Chainable[*Flow[T]]) *pipz.WorkerPool[*Flow[T]]

WorkerPool creates a bounded parallel executor with a fixed number of workers.

Types

type Await 

type Await[T, V any] struct {
	// contains filtered or unexported fields
}

Await waits for a correlated event on a signal.

func NewAwait 

func NewAwait[T, V any](name pipz.Name, signal capitan.Signal, key capitan.GenericKey[V]) *Await[T, V]

NewAwait creates an await primitive.

func (*Await[T, V]) Build 

func (a *Await[T, V]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

func (*Await[T, V]) Close 

func (*Await[T, V]) Close() error

Close implements Chainable.

func (*Await[T, V]) Name 

func (a *Await[T, V]) Name() pipz.Name

Name returns the processor name.

func (*Await[T, V]) Process 

func (a *Await[T, V]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable.

func (*Await[T, V]) Timeout 

func (a *Await[T, V]) Timeout(d time.Duration) *Await[T, V]

Timeout sets the maximum wait time.

func (*Await[T, V]) WithCapitan 

func (a *Await[T, V]) WithCapitan(c *capitan.Capitan) *Await[T, V]

WithCapitan sets a custom capitan instance. Defaults to global.

type CerealStore 

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

CerealStore implements Store using PostgreSQL via cereal patterns. Requires tables: ago_pending_states, ago_saga_states.

func NewCerealStore 

func NewCerealStore(db *sqlx.DB) *CerealStore

NewCerealStore creates a Store backed by PostgreSQL.

func (*CerealStore) DeletePending 

func (s *CerealStore) DeletePending(ctx context.Context, correlationID string) error

DeletePending removes a pending state.

func (*CerealStore) DeleteSaga 

func (s *CerealStore) DeleteSaga(ctx context.Context, correlationID string) error

DeleteSaga removes a saga state.

func (*CerealStore) GetPending 

func (s *CerealStore) GetPending(ctx context.Context, correlationID string) (*PendingState, error)

GetPending retrieves a pending state.

func (*CerealStore) GetSaga 

func (s *CerealStore) GetSaga(ctx context.Context, correlationID string) (*SagaState, error)

GetSaga retrieves a saga state.

func (*CerealStore) IsCompensated 

func (s *CerealStore) IsCompensated(ctx context.Context, correlationID, stepName string) (bool, error)

IsCompensated checks if a step has already been compensated.

func (*CerealStore) ListIncompleteSagas 

func (s *CerealStore) ListIncompleteSagas(ctx context.Context) ([]*SagaState, error)

ListIncompleteSagas returns all sagas that are not completed or failed.

func (*CerealStore) MarkCompensated 

func (s *CerealStore) MarkCompensated(ctx context.Context, correlationID, stepName string) error

MarkCompensated records that a step has been compensated for idempotency.

func (*CerealStore) Migrate 

func (s *CerealStore) Migrate(ctx context.Context) error

Migrate creates the required tables if they don't exist.

func (*CerealStore) SetPending 

func (s *CerealStore) SetPending(ctx context.Context, correlationID string, state *PendingState) error

SetPending stores a pending state.

func (*CerealStore) SetSaga 

func (s *CerealStore) SetSaga(ctx context.Context, correlationID string, state *SagaState) error

SetSaga stores a new saga state.

func (*CerealStore) UpdateSaga 

func (s *CerealStore) UpdateSaga(ctx context.Context, correlationID string, state *SagaState) error

UpdateSaga updates an existing saga state.

func (*CerealStore) WithSaga 

func (s *CerealStore) WithSaga(ctx context.Context, correlationID string, fn func(*SagaState) (*SagaState, error)) error

WithSaga executes a callback with exclusive access to a saga's state. Uses a transaction with SELECT FOR UPDATE to ensure exclusive access.

type Compensate 

type Compensate[T any] struct {
	// contains filtered or unexported fields
}

Compensate runs the compensation stack in reverse for a saga.

func NewCompensate 

func NewCompensate[T any](name pipz.Name, store Store, key capitan.GenericKey[T]) *Compensate[T]

NewCompensate creates a compensation primitive.

func (*Compensate[T]) Build 

func (c *Compensate[T]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

Design note: Compensate uses multiple WithSaga calls rather than a single atomic operation:

  1. Initial call: transition status to "compensating" and capture compensation records
  2. Per-step: emit signal, then MarkCompensated (outside WithSaga - uses its own idempotency)
  3. Final call: transition status to "failed" (compensation complete)

This design allows signal emission and external idempotency tracking between state transitions. The initial WithSaga ensures only one caller proceeds; others see "compensating" and return early.

func (*Compensate[T]) Close 

func (*Compensate[T]) Close() error

Close implements Chainable.

func (*Compensate[T]) Name 

func (c *Compensate[T]) Name() pipz.Name

Name returns the processor name.

func (*Compensate[T]) Process 

func (c *Compensate[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable.

func (*Compensate[T]) WithCapitan 

func (c *Compensate[T]) WithCapitan(cpt *capitan.Capitan) *Compensate[T]

WithCapitan sets a custom capitan instance. Defaults to global.

type CompensationRecord 

type CompensationRecord struct {
	StepName string
	Signal   capitan.Signal
	Data     []byte
}

CompensationRecord stores data needed to execute a compensation action.

type DeadLetter 

type DeadLetter[T any] struct {
	// contains filtered or unexported fields
}

DeadLetter routes failed messages to a dead letter queue.

func NewDeadLetter 

func NewDeadLetter[T any](name pipz.Name, key capitan.GenericKey[T]) *DeadLetter[T]

NewDeadLetter creates a dead letter primitive.

func (*DeadLetter[T]) Build 

func (d *DeadLetter[T]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

func (*DeadLetter[T]) Close 

func (*DeadLetter[T]) Close() error

Close implements Chainable.

func (*DeadLetter[T]) Name 

func (d *DeadLetter[T]) Name() pipz.Name

Name returns the processor name.

func (*DeadLetter[T]) Process 

func (d *DeadLetter[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable.

func (*DeadLetter[T]) WithCapitan 

func (d *DeadLetter[T]) WithCapitan(c *capitan.Capitan) *DeadLetter[T]

WithCapitan sets a custom capitan instance.

func (*DeadLetter[T]) WithProvider 

func (d *DeadLetter[T]) WithProvider(provider herald.Provider) *DeadLetter[T]

WithProvider sets a broker provider for external DLQ.

func (*DeadLetter[T]) WithSignal 

func (d *DeadLetter[T]) WithSignal(signal capitan.Signal) *DeadLetter[T]

WithSignal sets a custom signal for dead letter events.

type Emit 

type Emit[T any] struct {
	// contains filtered or unexported fields
}

Emit emits a capitan signal with fields derived from the flow.

func NewEmit 

func NewEmit[T any](name pipz.Name, signal capitan.Signal, key capitan.GenericKey[T]) *Emit[T]

NewEmit creates an emit primitive.

func (*Emit[T]) Build 

func (e *Emit[T]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

func (*Emit[T]) Close 

func (*Emit[T]) Close() error

Close implements Chainable.

func (*Emit[T]) Name 

func (e *Emit[T]) Name() pipz.Name

Name returns the processor name.

func (*Emit[T]) Process 

func (e *Emit[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable.

func (*Emit[T]) WithCapitan 

func (e *Emit[T]) WithCapitan(c *capitan.Capitan) *Emit[T]

WithCapitan sets a custom capitan instance. Defaults to global.

type Flow 

type Flow[T any] struct {
	// Payload is the typed business data.
	Payload T

	// Origin metadata captured at creation.
	Signal    capitan.Signal
	Timestamp time.Time
	Severity  capitan.Severity

	// Correlation for saga and request/response patterns.
	CorrelationID string
	CausationID   string

	// Broker metadata for herald integration.
	Metadata map[string]string

	// Errors accumulated during processing.
	Errors []error
	// contains filtered or unexported fields
}

Flow wraps a typed payload with correlation context and accumulated state. T is the business payload type.

func NewFlow 

func NewFlow[T any](payload T, signal capitan.Signal) *Flow[T]

NewFlow creates a Flow with the given payload and signal.

func NewFromEvent 

func NewFromEvent[T any](e *capitan.Event, key capitan.GenericKey[T]) *Flow[T]

NewFromEvent creates a Flow from a capitan Event using a typed key. Returns nil if the key is not present in the event.

func (*Flow[T]) AddError 

func (f *Flow[T]) AddError(err error)

AddError appends an error to the flow's error accumulator.

func (*Flow[T]) Clone 

func (f *Flow[T]) Clone() *Flow[T]

Clone creates a deep copy of the Flow for parallel processing. Implements pipz.Cloner[*Flow[T]].

func (*Flow[T]) Fields 

func (f *Flow[T]) Fields() []capitan.Field

Fields returns all accumulated fields as a slice.

func (*Flow[T]) Get 

func (f *Flow[T]) Get(key capitan.Key) capitan.Field

Get retrieves a field by key, returning nil if not present.

func (*Flow[T]) HasErrors 

func (f *Flow[T]) HasErrors() bool

HasErrors returns true if any errors have been accumulated.

func (*Flow[T]) Set 

func (f *Flow[T]) Set(field capitan.Field)

Set adds or updates a typed field in the flow's accumulated state.

type MemoryStore 

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

MemoryStore is an in-memory Store implementation for testing and single-instance use.

func NewMemoryStore 

func NewMemoryStore() *MemoryStore

NewMemoryStore creates a new in-memory store.

func (*MemoryStore) DeletePending 

func (m *MemoryStore) DeletePending(_ context.Context, correlationID string) error

DeletePending removes a pending state.

func (*MemoryStore) DeleteSaga 

func (m *MemoryStore) DeleteSaga(_ context.Context, correlationID string) error

DeleteSaga removes a saga state.

func (*MemoryStore) GetPending 

func (m *MemoryStore) GetPending(_ context.Context, correlationID string) (*PendingState, error)

GetPending retrieves a pending state.

func (*MemoryStore) GetSaga 

func (m *MemoryStore) GetSaga(_ context.Context, correlationID string) (*SagaState, error)

GetSaga retrieves a saga state. Returns a deep copy to prevent external mutations from affecting stored state.

func (*MemoryStore) IsCompensated 

func (m *MemoryStore) IsCompensated(_ context.Context, correlationID, stepName string) (bool, error)

IsCompensated checks if a step has already been compensated.

func (*MemoryStore) ListIncompleteSagas 

func (m *MemoryStore) ListIncompleteSagas(_ context.Context) ([]*SagaState, error)

ListIncompleteSagas returns all sagas that are not completed or failed. Returns deep copies to prevent external mutations from affecting stored state.

func (*MemoryStore) MarkCompensated 

func (m *MemoryStore) MarkCompensated(_ context.Context, correlationID, stepName string) error

MarkCompensated records that a step has been compensated for idempotency.

func (*MemoryStore) SetPending 

func (m *MemoryStore) SetPending(_ context.Context, correlationID string, state *PendingState) error

SetPending stores a pending state.

func (*MemoryStore) SetSaga 

func (m *MemoryStore) SetSaga(_ context.Context, correlationID string, state *SagaState) error

SetSaga stores a new saga state. Stores a deep copy to prevent external mutations from affecting stored state.

func (*MemoryStore) UpdateSaga 

func (m *MemoryStore) UpdateSaga(_ context.Context, correlationID string, state *SagaState) error

UpdateSaga updates an existing saga state. Stores a deep copy to prevent external mutations from affecting stored state.

func (*MemoryStore) WithSaga 

func (m *MemoryStore) WithSaga(_ context.Context, correlationID string, fn func(*SagaState) (*SagaState, error)) error

WithSaga executes a callback with exclusive access to a saga's state. The saga lock is held for the duration of the callback.

type PendingState 

type PendingState struct {
	CorrelationID string
	Signal        capitan.Signal
	CreatedAt     time.Time
	Timeout       time.Duration
}

PendingState represents a request or await waiting for a response.

type Request 

type Request[T, R any] struct {
	// contains filtered or unexported fields
}

Request sends a request and waits for a correlated response.

func NewRequest 

func NewRequest[T, R any](
	name pipz.Name,
	requestSignal capitan.Signal,
	responseSignal capitan.Signal,
	requestKey capitan.GenericKey[T],
	responseKey capitan.GenericKey[R],
) *Request[T, R]

NewRequest creates a request/response primitive.

func (*Request[T, R]) Build 

func (r *Request[T, R]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

func (*Request[T, R]) Close 

func (*Request[T, R]) Close() error

Close implements Chainable.

func (*Request[T, R]) Name 

func (r *Request[T, R]) Name() pipz.Name

Name returns the processor name.

func (*Request[T, R]) Process 

func (r *Request[T, R]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable.

func (*Request[T, R]) Timeout 

func (r *Request[T, R]) Timeout(d time.Duration) *Request[T, R]

Timeout sets the maximum wait time.

func (*Request[T, R]) WithCapitan 

func (r *Request[T, R]) WithCapitan(c *capitan.Capitan) *Request[T, R]

WithCapitan sets a custom capitan instance. Defaults to global.

type SagaState 

type SagaState struct {
	CorrelationID string
	Status        SagaStatus
	CurrentStep   int
	Compensations []CompensationRecord
	CreatedAt     time.Time
	UpdatedAt     time.Time
	Error         string
	// Timeout specifies how long the saga may run before being considered expired.
	// Zero means no timeout. RecoverSagas will compensate expired sagas.
	Timeout time.Duration
}

SagaState tracks a saga's execution and compensation stack.

func (*SagaState) IsExpired 

func (s *SagaState) IsExpired() bool

IsExpired returns true if the saga has a timeout and has exceeded it.

type SagaStatus 

type SagaStatus string

SagaStatus represents the lifecycle state of a saga. 

const (
	// SagaStatusPending indicates the saga has been created but not started.
	SagaStatusPending SagaStatus = "pending"
	// SagaStatusRunning indicates the saga is actively executing steps.
	SagaStatusRunning SagaStatus = "running"
	// SagaStatusCompensating indicates the saga is rolling back via compensation.
	SagaStatusCompensating SagaStatus = "compensating"
	// SagaStatusCompleted indicates the saga finished successfully.
	SagaStatusCompleted SagaStatus = "completed"
	// SagaStatusFailed indicates the saga failed and compensation is complete.
	SagaStatusFailed SagaStatus = "failed"
)

type SagaStep 

type SagaStep[T any] struct {
	// contains filtered or unexported fields
}

SagaStep executes a saga step with compensation registration.

func NewSagaStep 

func NewSagaStep[T any](
	name pipz.Name,
	store Store,
	key capitan.GenericKey[T],
	execute capitan.Signal,
	compensate capitan.Signal,
) *SagaStep[T]

NewSagaStep creates a saga step. Store is required for saga state persistence and idempotency tracking.

func (*SagaStep[T]) Build 

func (s *SagaStep[T]) Build() pipz.Chainable[*Flow[T]]

Build creates the chainable processor.

func (*SagaStep[T]) Close 

func (*SagaStep[T]) Close() error

Close implements Chainable.

func (*SagaStep[T]) Name 

func (s *SagaStep[T]) Name() pipz.Name

Name returns the processor name.

func (*SagaStep[T]) Process 

func (s *SagaStep[T]) Process(ctx context.Context, f *Flow[T]) (*Flow[T], error)

Process implements Chainable by delegating to Build().

func (*SagaStep[T]) WithCapitan 

func (s *SagaStep[T]) WithCapitan(c *capitan.Capitan) *SagaStep[T]

WithCapitan sets a custom capitan instance. Defaults to global.

func (*SagaStep[T]) WithTimeout 

func (s *SagaStep[T]) WithTimeout(d time.Duration) *SagaStep[T]

WithTimeout sets the saga timeout. If the saga runs longer than this duration, RecoverSagas will trigger compensation. Zero means no timeout. Note: This only affects saga creation - if the saga already exists, timeout is unchanged.

type Store 

type Store interface {
	// Pending request/await state.
	SetPending(ctx context.Context, correlationID string, state *PendingState) error
	GetPending(ctx context.Context, correlationID string) (*PendingState, error)
	DeletePending(ctx context.Context, correlationID string) error

	// Saga state.
	SetSaga(ctx context.Context, correlationID string, state *SagaState) error
	GetSaga(ctx context.Context, correlationID string) (*SagaState, error)
	UpdateSaga(ctx context.Context, correlationID string, state *SagaState) error
	DeleteSaga(ctx context.Context, correlationID string) error
	ListIncompleteSagas(ctx context.Context) ([]*SagaState, error)

	// WithSaga executes a callback with exclusive access to a saga's state.
	// If the saga doesn't exist, callback receives nil and can return a new state to create it.
	// If callback returns a non-nil state, it is saved. If callback returns an error,
	// no changes are persisted.
	// Implementations must ensure the callback has exclusive access (mutex, transaction, etc.).
	//
	// NOTE: Signal emission typically happens AFTER WithSaga returns, outside the lock.
	// This means a crash between state commit and signal emission could leave state
	// updated but signal not emitted. This is acceptable because:
	// - Idempotency keys allow safe retry
	// - At-least-once delivery is the expected semantic
	// - Holding locks during signal emission would risk deadlocks
	WithSaga(ctx context.Context, correlationID string, fn func(*SagaState) (*SagaState, error)) error

	// Idempotency for compensation actions.
	MarkCompensated(ctx context.Context, correlationID, stepName string) error
	IsCompensated(ctx context.Context, correlationID, stepName string) (bool, error)
}

Store provides persistence for coordination and saga state. Implementations enable distributed coordination and restart recovery.

Source Files

View all Source files

Directories

Path Synopsis

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.