Workflows

Workflows are durable, multi-step Go functions. Dispatch checkpoints each completed step so that if the process restarts mid-workflow, execution resumes from the last completed step — not from the beginning.

Defining a workflow01

import "github.com/xraph/dispatch/workflow"

type OrderInput struct {
    OrderID      string  `json:"order_id"`
    Amount       float64 `json:"amount"`
    CustomerID   string  `json:"customer_id"`
    PaymentToken string  `json:"payment_token"`
}

var ProcessOrder = workflow.NewWorkflow("process_order",
    func(wf *workflow.Workflow, input OrderInput) error {
        if err := wf.Step("validate", func() error {
            return validateOrder(input.OrderID)
        }); err != nil {
            return err
        }

        if err := wf.Step("charge", func() error {
            return chargeCard(input.PaymentToken, input.Amount)
        }); err != nil {
            return err
        }

        return wf.Step("fulfill", func() error {
            return fulfillOrder(input.OrderID)
        })
    },
)

Registering a workflow02

engine.RegisterWorkflow(eng, ProcessOrder)

Starting a workflow03

engine.StartWorkflow(ctx, eng, ProcessOrder, OrderInput{
    OrderID: "ORD-001",
    Amount:  99.99,
})

Steps04

wf.Step(name, fn) executes fn and persists a checkpoint. If the workflow is re-executed (after a crash or restart), a step with an existing checkpoint is skipped and its stored result is returned.

This guarantees that side effects (HTTP calls, DB writes, email sends) happen exactly once across process restarts.

Parallel steps05

Run independent steps concurrently:

if err := wf.Parallel(
    func() error { return sendConfirmationEmail(input.CustomerID) },
    func() error { return updateInventory(input.OrderID) },
    func() error { return notifyWarehouse(input.OrderID) },
); err != nil {
    return err
}

Parallel fails fast if any step returns an error. All non-failed goroutines complete before the error is returned.

Waiting for events06

A workflow can pause until an external event arrives:

payload, err := wf.WaitForEvent(ctx, "payment.confirmed", 24*time.Hour)
if err != nil {
    return err // timeout or context cancelled
}

To trigger the event from outside:

engine.TriggerEvent(ctx, eng, runID, "payment.confirmed", confirmationPayload)

Sleeping07

Pause execution for a duration (persists across restarts):

if err := wf.Sleep(ctx, 30*time.Minute); err != nil {
    return err
}

Run states08

State	Meaning
`running`	Workflow is executing
`completed`	All steps finished successfully
`failed`	A step failed terminally

← Previous

Jobs

Dead Letter Queue