Go SDK

The official Go SDK for Ironflow. Provides recorded execution (durable functions), event emission, real-time subscriptions, KV store, config management, entity streams, derived views (projections), and a full management API.

Installation

go get github.com/sahina/ironflow-go/ironflow

Requires Go 1.25+.

Quick Start

Define a Function

import "github.com/sahina/ironflow-go/ironflow"

var ProcessOrder = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:       "process-order",
    Triggers: []ironflow.Trigger{{Event: "order.placed"}},
}, func(ctx ironflow.Context) (any, error) {
    var order OrderData
    if err := ctx.Event.Data(&order); err != nil {
        return nil, err
    }

    // Run a step (automatically memoized)
    validated, err := ironflow.Run(ctx, "validate", func() (any, error) {
        return validateOrder(order)
    })
    if err != nil {
        return nil, err
    }

    // Sleep (durable - survives restarts)
    if err := ironflow.Sleep(ctx, "wait", 5*time.Minute); err != nil {
        return nil, err
    }

    // Wait for external event
    approval, err := ironflow.WaitForEvent[ApprovalEvent](ctx, "wait-approval", ironflow.EventFilter{
        Event:   "order.approved",
        Match:   "data.orderId",
        Timeout: 24 * time.Hour,
    })
    if err != nil {
        return nil, err
    }

    result, err := ironflow.Run(ctx, "process-payment", func() (any, error) {
        return processPayment(validated, approval)
    })
    return result, err
})

Push Mode (Serverless)

For serverless deployments (Vercel, Lambda):

handler := ironflow.Serve(ironflow.ServeConfig{
    Functions:  []ironflow.Function{ProcessOrder},
    SigningKey: os.Getenv("IRONFLOW_SIGNING_KEY"),
})

http.Handle("/api/ironflow", handler)
http.ListenAndServe(":3000", nil)

Pull Mode (Worker)

For long-running workers with no timeout limits:

worker := ironflow.NewWorker(ironflow.WorkerConfig{
    Functions:         []ironflow.Function{ProcessOrder},
    MaxConcurrentJobs: 10,
})

ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM)
defer cancel()

if err := worker.Run(ctx); err != nil {
    log.Fatal(err)
}

Function Definition

CreateFunction

var MyFunction = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:   "my-function",
    Name: "My Function",

    // Event triggers
    Triggers: []ironflow.Trigger{
        {Event: "user.created"},
        {Event: "order.*"},
        {Event: "order.placed", Expression: `data.total > 100`}, // CEL filter
    },

    // Retry configuration
    Retry: &ironflow.RetryConfig{
        MaxAttempts:   5,
        InitialDelay:  1 * time.Second,
        BackoffFactor: 2.0,
        MaxDelay:      5 * time.Minute,
    },

    // Execution timeout
    Timeout:     30 * time.Minute,
    StepTimeout: 60 * time.Second, // Default timeout for all steps

    // Concurrency limit
    Concurrency: &ironflow.ConcurrencyConfig{
        Limit: 10,
        Key:   "event.data.customerId", // Group by customer
    },

    // Execution mode
    Mode: ironflow.PullMode, // or ironflow.PushMode (default)

    // Secrets required at runtime
    Secrets: []string{"STRIPE_KEY", "DB_PASSWORD"},

    // Declarative cancel-on-event.
    // Auto-cancels in-flight runs when a matching event arrives.
    // OR semantics across specs. Tenant-isolated by env_id.
    // See how-to: ../../how-to-guides/workflows/cancel-on-event.md
    CancelOn: []ironflow.CancelOnConfig{
        {Event: "order.cancelled", Match: "orderId"},
    },
}, func(ctx ironflow.Context) (any, error) {
    return map[string]any{"status": "done"}, nil
})

The server rejects a function ID containing the NATS wildcards * or >, whitespace, a control character, or an empty . segment (a..b, .a, a.), and any ID over 228 bytes, with CodeInvalidArgument. The rule applies to new registrations only — a function already in the database keeps its ID and still re-registers, so an upgrade never stops a worker that holds a pre-existing malformed ID.

CreateFunction and CreateHandler apply the same rule client-side and panic at declaration rather than letting a bad ID reach registration, where the first error aborts the whole worker’s loop. They do not bound length — that one is the server’s, and you get a precise CodeInvalidArgument for it.

Before #1750 the SDK check was ^[a-zA-Z0-9-_]+$, which is narrower than the server rule: dotted and non-ASCII IDs are legal server-side but panicked here. Both now work. CreateHandler skipped the check entirely; a whitespace Options.ID that used to reach the server now panics at declaration.

Function Context

The handler receives a Context:

type Context struct {
    Event   Event         // Triggering event
    Run     RunInfo       // Run info (ID, FunctionID, Attempt, StartedAt)
    Secrets SecretsReader // Read-only access to resolved secrets
}

Event Data

func(ctx ironflow.Context) (any, error) {
    // Unmarshal event data into a struct
    var order OrderData
    if err := ctx.Event.Data(&order); err != nil {
        return nil, err
    }

    // Access raw event fields
    eventName := ctx.Event.Name       // e.g. "order.placed"
    eventID := ctx.Event.ID
    eventVersion := ctx.Event.Version // Schema version (default: 1)
    source := ctx.Event.Source        // "api", "cron", "webhook"

    // Access user-defined metadata attached when the event was emitted
    traceID, _ := ctx.Event.Metadata["traceId"].(string)
    eventSource, _ := ctx.Event.Metadata["source"].(string)

    return nil, nil
}

CreateHandler (Simplified API)

Type-safe generic API with auto-generated function IDs:

type OrderEvent struct {
    OrderID string  `json:"orderId"`
    Total   float64 `json:"total"`
}

// Minimal config — ID auto-generated as "order-placed-handler"
var OrderHandler = ironflow.CreateHandler(ironflow.HandlerConfig[OrderEvent]{
    Event: "order.placed",
    Handler: func(event OrderEvent, ctx *ironflow.HandlerContext) (any, error) {
        ctx.Logger.Info("Processing order", "id", event.OrderID)
        return map[string]any{"processed": true}, nil
    },
})

// With all options
var HighValueHandler = ironflow.CreateHandler(ironflow.HandlerConfig[OrderEvent]{
    Event: "order.placed",
    Handler: func(event OrderEvent, ctx *ironflow.HandlerContext) (any, error) {
        result, err := ctx.Step.Run("process", func() (any, error) {
            return processOrder(event)
        })
        return result, err
    },
    Options: &ironflow.HandlerOptions{
        ID:     "high-value-order-handler",
        Filter: `data.total > 1000`,
        Retry:  &ironflow.RetryConfig{MaxAttempts: 5},
        Concurrency: &ironflow.ConcurrencyConfig{
            Limit: 10,
            Key:   "event.data.customerId",
        },
    },
})

The HandlerContext provides:

ctx.Event     // Typed event data
ctx.EventMeta // Event metadata (ID, Name, Version, Timestamp, Source)
ctx.Run       // Run info (ID, FunctionID, Attempt)
ctx.Secrets   // SecretsReader
ctx.Step      // StepClient (Run, Sleep, SleepUntil, WaitForEvent, Parallel, Map, Compensate)
ctx.Logger    // Logger

Step Primitives

Run

Execute a memoized step. Results are cached — if the function restarts, completed steps are skipped.

result, err := ironflow.Run(ctx, "step-id", func() (any, error) {
    return someApiCall()
})

// With timeout override
result, err := ironflow.Run(ctx, "step-id", func() (any, error) {
    return someApiCall()
}, ironflow.WithTimeout(10*time.Second))

Sleep

Durable pause. Survives function restarts and server upgrades.

if err := ironflow.Sleep(ctx, "wait", 5*time.Minute); err != nil {
    return nil, err
}

SleepUntil

Sleep until a specific time.

if err := ironflow.SleepUntil(ctx, "wait-until", time.Date(2025, 12, 25, 0, 0, 0, 0, time.UTC)); err != nil {
    return nil, err
}

WaitForEvent

Wait for an external event matching a filter.

event, err := ironflow.WaitForEvent[ApprovalEvent](ctx, "wait-approval", ironflow.EventFilter{
    Event:   "order.approved",
    Match:   "data.orderId",        // JSON path for matching
    Timeout: 24 * time.Hour,        // Default: 7 days
})

Parallel

Execute multiple branches concurrently.

results, err := ironflow.Parallel(ctx, "fetch-all",
    []func(*ironflow.BranchContext) (any, error){
        func(b *ironflow.BranchContext) (any, error) {
            return ironflow.RunWithBranch(b, "fetch-user", func() (any, error) {
                return fetchUser(userID)
            })
        },
        func(b *ironflow.BranchContext) (any, error) {
            return ironflow.RunWithBranch(b, "fetch-orders", func() (any, error) {
                return fetchOrders(userID)
            })
        },
    },
    ironflow.ParallelOptions{Concurrency: 2, OnError: "failFast"},
)

The *BranchContext each branch receives is load-bearing, not stylistic. This applies to Map below as well. A branch is not itself a recorded step — Parallel and Map only call your callback. Work done through the branch context is memoized under that branch’s scope; work done directly in the callback body persists nothing and re-runs in full on every retry.

Reaching for the enclosing function’s ctx instead is a third case: those branches do record real steps, but at the function’s top level rather than under the parallel — a flat timeline, and IDs that stay unique only as long as the branch names differ.

All three forms compile, so the SDK warns — once per distinct step name per run — on two shapes:

A step claimed on the enclosing ctx while the fan-out was running. These branches do persist steps, but their index comes from a counter shared by every branch, so which branch gets which index depends on goroutine scheduling — and a resume can hand one branch another branch’s output.
Every branch skipped the branch context. Nothing was persisted under the fan-out at all.

Treat it as a best-effort lint, not a guarantee. It stays silent when any branch errors, and when a branch yields — an enclosing Sleep/SleepUntil/ WaitForEvent unwinds before the check runs, though enclosing Run/Invoke/ Publish are caught. For a fan-out that genuinely has nothing to memoize (a pure in-memory transform run through Map only for its concurrency limit), set ParallelOptions{SkipScopedClientCheck: true} to silence both.

Draining before you migrate. Switching an existing branch from the enclosing ctx to the branch context changes its step ID from {runID}:{name}:0 to {runID}:{parallelName}:{branchIndex}:{name}:0. Nothing bridges the two — preferLegacyStepID exists only for the escaping rollout and early-returns for any name that escapes to itself. An in-flight run that resumes after the deploy (one sleeping, waiting for an event, or on a later attempt) finds no memoized row and re-executes those branch steps for real. Let in-flight runs drain before shipping the rewrite.

Map

Map over items with automatic parallelization.

results, err := ironflow.Map(ctx, "process-items", items,
    func(item Item, b *ironflow.BranchContext, index int) (Result, error) {
        return ironflow.RunWithBranch(b, fmt.Sprintf("process-%d", index), func() (Result, error) {
            return processItem(item)
        })
    },
    ironflow.ParallelOptions{Concurrency: 5},
)

Branch-scoped steps

Inside a branch, use the *BranchContext form of each primitive. The root form takes its step ID from the enclosing function, so it records the step outside the branch’s scope:

Root form	Branch-scoped form
`ironflow.Run`	`ironflow.RunWithBranch`
`ironflow.Sleep`	`ironflow.SleepWithBranch`
`ironflow.SleepUntil`	`ironflow.SleepUntilWithBranch`
`ironflow.WaitForEvent`	`ironflow.WaitForEventWithBranch`
`ironflow.Parallel`	`ironflow.ParallelWithBranch`
`ironflow.Map`	`ironflow.MapWithBranch`
`ironflow.Invoke`	`ironflow.InvokeWithBranch`
`ironflow.InvokeAsync`	`ironflow.InvokeAsyncWithBranch`
`ironflow.Publish`	`ironflow.PublishWithBranch`
`ironflow.Compensate`	`ironflow.CompensateInBranch`

Invoke and InvokeAsync are the sharpest case: they key their step ID on the functionID rather than a name you choose, so two branches invoking the same function drew from one shared counter. Which branch got :0 and which got :1 depended on goroutine scheduling, and on resume the assignment could flip and hand a branch the other branch’s memoized output. Use the branch-scoped form.

results, err := ironflow.ParallelWithBranch(b, "shards",
    []func(*ironflow.BranchContext) ([]Doc, error){
        func(shard *ironflow.BranchContext) ([]Doc, error) {
            return ironflow.MapWithBranch(shard, "ingest", files,
                func(f string, item *ironflow.BranchContext, i int) (Doc, error) {
                    return ironflow.RunWithBranch(item, fmt.Sprintf("ingest-%d", i), func() (Doc, error) {
                        return ingest(f)
                    })
                })
        },
    },
)

Invoke

Call another Ironflow function and wait for the result.

result, err := ironflow.Invoke[PaymentResult](ctx, "process-payment", map[string]any{
    "orderId": "123",
    "amount":  99.99,
})

// With custom timeout (default: 30s)
result, err := ironflow.Invoke[PaymentResult](ctx, "process-payment", input,
    ironflow.WithInvokeTimeout(60*time.Second),
)

InvokeAsync

Call another function without waiting for the result.

asyncResult, err := ironflow.InvokeAsync(ctx, "send-email", map[string]any{
    "to":      "user@example.com",
    "subject": "Order Confirmed",
})
fmt.Println("Child run:", asyncResult.RunID)

Compensate (Saga Pattern)

_, err := ironflow.Run(ctx, "charge-payment", func() (any, error) {
    return chargeCard(order.CardID, order.Total)
})
if err != nil {
    return nil, err
}

// Register compensation — runs in reverse order if a later step fails
ironflow.Compensate(ctx, "charge-payment", func() error {
    return refundCard(order.CardID, order.Total)
})

// If this step fails, "charge-payment" compensation runs automatically
_, err = ironflow.Run(ctx, "ship-order", func() (any, error) {
    return shipOrder(order.ID)
})

Publish (Step)

Publish a message to a developer pub/sub topic as a durable step.

if err := ironflow.Publish(ctx, "notifications", map[string]any{
    "type":    "order.shipped",
    "orderId": order.ID,
}); err != nil {
    return nil, err
}

Client

NewClient

client := ironflow.NewClient(ironflow.ClientConfig{
    ServerURL: "http://localhost:9123",  // default: IRONFLOW_SERVER_URL or http://localhost:9123
    APIKey:    "optional-api-key",       // default: IRONFLOW_API_KEY
    Timeout:   30 * time.Second,         // default: 30s

    // Retry configuration (optional)
    Retry: &ironflow.ClientRetryConfig{
        MaxAttempts:          3,
        InitialDelay:         100 * time.Millisecond,
        MaxDelay:             10 * time.Second,
        BackoffMultiplier:    2.0,
        ConnectionRetryDelay: 2 * time.Second,
        OnRetry: func(event ironflow.RetryEvent) {
            log.Printf("Retry %d/%d: %v (waiting %s)", event.Attempt, event.MaxAttempts, event.Error, event.Delay)
        },
    },

    // Custom logger (optional)
    Logger: ironflow.NewNoopLogger(), // disable logging
})

Event Emission

Emit

Fire-and-forget event emission:

result, err := client.Emit(ctx, "order.placed", map[string]any{
    "orderId": "123",
    "total":   99.99,
})
fmt.Println("Event ID:", result.EventID)
fmt.Println("Run IDs:", result.RunIDs)

Emit options:

// With idempotency key (deduplication)
result, err := client.Emit(ctx, "payment.processed", data,
    ironflow.WithEmitIdempotencyKey("payment-abc"),
)

// With event version
result, err := client.Emit(ctx, "order.placed", data,
    ironflow.WithEmitVersion(2),
)

// With metadata
result, err := client.Emit(ctx, "order.placed", data,
    ironflow.WithEmitMetadata(map[string]any{"source": "api"}),
)

EmitSync

Emit an event and wait for every run it triggers. One event can match several triggers, so EmitSync returns a slice — one element per matched run, empty when the event matched nothing:

results, err := client.EmitSync(ctx, "order.placed", map[string]any{
    "orderId": "123",
}, 30*time.Second)
if err != nil {
    return err
}

for _, result := range results {
    fmt.Println("Function:", result.FunctionID)
    fmt.Println("Status:", result.Status)   // "completed", "failed"
    fmt.Println("Output:", result.Output)
    fmt.Println("Duration:", result.DurationMs, "ms")

    if result.WaitTimedOut {
        // The synchronous wait ended, but the durable run is still active.
        fmt.Println("Still running:", result.RunID, result.Status)
    }
}

// With an idempotency key — a repeat emit returns the original runs
results, err = client.EmitSync(ctx, "order.placed", data, 0,
    ironflow.WithSyncIdempotencyKey("order-123-placed"),
)

// With an event schema version (#1955). Omit for 1. The synchronous
// counterpart of WithEmitVersion; a no-op on InvokeSync, which shares the
// option type but generates no event and so has no schema to select.
results, err = client.EmitSync(ctx, "order.placed", data, 0,
    ironflow.WithSyncVersion(2),
)

A per-run failure is reported in result.Error, not in the returned error. The returned error covers transport and protocol failures only.

InvokeSync

Run one function by ID and wait for its result. Unlike EmitSync, which is keyed by an event and fans out, InvokeSync targets a single function and returns a single result (ADR 0067):

result, err := client.InvokeSync(ctx, "process-order", map[string]any{
    "orderId": "123",
}, 30*time.Second)
if err != nil {
    return err
}
if result.Status == ironflow.RunStatusCompleted {
    fmt.Printf("Order processed: %v\n", result.Output)
}

Accepts the same sync options as EmitSync — WithSyncIdempotencyKey and WithSyncMetadata. A protocol violation — a response carrying no result — is an IronflowError with code INVALID_RESPONSE.

Warning: InvokeFunctionSync ties the run’s lifetime to the request context. Cancelling ctx, or dropping the connection, cancels the run server-side. The timeout argument is a server-side wait budget: it does not cancel the run, and the SDK raises the transport deadline above it when the client’s own Timeout would undercut it.

Run Management

GetRun

run, err := client.GetRun(ctx, "run_xyz789")
fmt.Println("Status:", run.Status) // "waiting_for_capacity", "waiting", "running", "completed", "failed", "cancelled", "paused"

ListRuns

result, err := client.ListRuns(ctx, &ironflow.ListRunsOptions{
    FunctionID: "process-order",
    Status:     "completed",
    Limit:      50,
})

for _, run := range result.Runs {
    fmt.Printf("Run %s: %s\n", run.ID, run.Status)
}

Pass result.NextCursor back as Cursor to walk pages. It is empty on the last page, which is the loop’s termination condition — do not test the page length, since a full page can also be the final one:

var cursor string
for {
    result, err := client.ListRuns(ctx, &ironflow.ListRunsOptions{
        FunctionID: "process-order",
        Limit:      50,
        Cursor:     cursor,
    })
    if err != nil {
        return err
    }
    for _, run := range result.Runs {
        fmt.Printf("Run %s: %s\n", run.ID, run.Status)
    }
    if result.NextCursor == "" {
        break
    }
    cursor = result.NextCursor
}

Pagination is a keyset walk over (created_at, id) descending, so runs created while you page do not shift rows between pages or repeat them. The cursor is an opaque token — do not parse or construct one; a malformed value returns INVALID_ARGUMENT. TotalCount is the count of every matching run, not the number remaining after the cursor.

An unrecognized Status is rejected client-side. The filter lands on a protobuf enum field, and the server silently ignores a value it cannot parse rather than erroring, so the SDK catches it instead of letting you get unfiltered results.

CancelRun

run, err := client.CancelRun(ctx, "run_xyz789", "no longer needed")

PatchStep

Hot-patch a step’s output to fix a failed run:

err := client.PatchStep(ctx, "step_abc123", map[string]any{
    "correctedValue": 42,
}, "fixed bad API response")

History Editing (Scoped Injection)

Pause running workflows at step boundaries, inspect and modify step outputs, then resume:

// Pause a running workflow at the next step boundary
status, err := client.PauseRun(ctx, "run_abc123")
fmt.Println("Status:", status) // "paused"

// Get the paused state with completed steps
state, err := client.GetPausedState(ctx, "run_abc123")
for _, step := range state.Steps {
    fmt.Printf("Step %s: injected=%v output=%s\n", step.Name, step.Injected, step.Output)
}
fmt.Println("Next step:", state.NextStepHint)

// Inject modified output for a step
prevOutput, err := client.InjectStepOutput(ctx, "run_abc123", "step_xyz",
    json.RawMessage(`{"corrected": true}`), "Manual correction")
fmt.Println("Previous:", string(prevOutput))

// Resume the run
run, err := client.ResumeRun(ctx, "run_abc123", "")

Inspect historical run state at any timestamp. Requires Recording: true on the function.

// Get run state at a specific point in time
snapshot, err := client.GetRunStateAt(ctx, "run_abc123", time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC))
fmt.Println("Status at that time:", snapshot.Status)
for _, step := range snapshot.Steps {
    fmt.Printf("Step %s: status=%s patched=%v\n", step.Name, step.Status, step.Patched)
}

// Get full timeline of audit events for a run
events, err := client.GetRunTimeline(ctx, "run_abc123")
for _, evt := range events {
    fmt.Printf("[%s] %s: %s (significant=%v)\n", evt.Timestamp, evt.EventType, evt.Summary, evt.Significant)
}

// Get a specific step's output at a point in time
stepOutput, err := client.GetStepOutputAt(ctx, "run_abc123", "step_xyz",
    time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC))
fmt.Printf("Output: %s (patched=%v, injected=%v)\n", stepOutput.Output, stepOutput.Patched, stepOutput.Injected)

RegisterFunction

err := client.RegisterFunction(ctx, ProcessOrder)

Serve Handler

Serve

Create an HTTP handler for push mode:

handler := ironflow.Serve(ironflow.ServeConfig{
    // Required
    Functions: []ironflow.Function{fn1, fn2, fn3},

    // Optional: signing key for request verification
    SigningKey: os.Getenv("IRONFLOW_SIGNING_KEY"),

    // Optional: skip verification (dev only)
    SkipVerification: true,

    // Optional: projections to register
    Projections: []ironflow.Projection{OrderStats},

    // Optional: webhook sources
    Webhooks: []ironflow.Webhook{StripeWebhook},

    // Optional: server URL for emitting webhook events
    ServerURL: "http://localhost:9123",

    // Optional: event schema upcasters
    Upcasters: registry,
})

http.Handle("/api/ironflow", handler)
http.ListenAndServe(":3000", nil)

Worker

NewWorker

worker := ironflow.NewWorker(ironflow.WorkerConfig{
    ServerURL:         "http://localhost:9123",
    Functions:         []ironflow.Function{fn1, fn2},
    Projections:       []ironflow.Projection{OrderStats},
    MaxConcurrentJobs: 10,                    // default: 10
    Labels:            map[string]string{"region": "us-east-1"},
    HeartbeatInterval: 30 * time.Second,      // default: 30s
    ReconnectDelay:    5 * time.Second,       // default: 5s
    Logger:            ironflow.NewNoopLogger(),
    Upcasters:         registry,
})

Authentication failures stop the worker. A 401 or 403 from function registration, worker registration, or job polling makes Run log the API-key and bootstrap-key-file guidance and return an error matching errors.Is(err, ironflow.ErrUnauthorized) / ironflow.ErrForbidden. The worker does not retry on the ReconnectDelay cadence — a missing or revoked key does not fix itself, so a supervisor restart (with the key set) is the recovery path. Every other connection error still reconnects as before. NewStreamingWorker and ProjectionRunner behave the same way.

Worker Methods

// Start the worker (blocks until stopped)
err := worker.Run(ctx)

// Graceful drain (wait for active jobs to complete)
worker.Drain()

// Force stop
worker.Stop()

NewStreamingWorker

Uses ConnectRPC bidirectional streaming instead of HTTP polling. Lower latency for step delivery and real-time step lifecycle visibility (StepStarted/Completed/Failed events streamed to the server).

worker := ironflow.NewStreamingWorker(ironflow.WorkerConfig{
    ServerURL:         "http://localhost:9123",
    Functions:         []ironflow.Function{fn1, fn2},
    Projections:       []ironflow.Projection{OrderStats},
    MaxConcurrentJobs: 10,
    Labels:            map[string]string{"region": "us-east-1"},
    HeartbeatInterval: 30 * time.Second,
    ReconnectDelay:    5 * time.Second,
})

Same WorkerConfig, same Run/Drain/Stop methods as NewWorker. The only difference is the transport: a single persistent HTTP/2 connection instead of polling.

When to use streaming vs polling:

Use NewStreamingWorker when you need lower step delivery latency or real-time step visibility in the dashboard.
Use NewWorker when you want the simplest deployment (HTTP/1.1 compatible, no HTTP/2 requirement).

Both support h2c (HTTP/2 cleartext) for development and TLS for production.

Graceful Shutdown

ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM)
defer cancel()

go func() {
    <-ctx.Done()
    worker.Drain()
}()

if err := worker.Run(ctx); err != nil {
    log.Fatal(err)
}

Idempotency

Both NewWorker and NewStreamingWorker use at-least-once delivery semantics. If a worker disconnects while executing a step, the server requeues the job. Completed steps are memoized (not re-executed), but the currently-running step may execute again.

This means steps with side effects (sending emails, charging cards, calling external APIs) must be idempotent. Use idempotency keys, database upserts, or check-before-act patterns to prevent duplicate actions.

This is the same guarantee used by Temporal, Inngest, and every other durable execution engine.

KV Store

client.KV()

kv := client.KV()

CreateBucket

info, err := kv.CreateBucket(ctx, ironflow.BucketConfig{
    Name:         "sessions",
    Description:  "User session data",
    TTL:          time.Hour,
    MaxValueSize: 65536,
    MaxBytes:     1 << 30, // 1GB
    History:      5,
})

DeleteBucket

err := kv.DeleteBucket(ctx, "sessions")

ListBuckets

buckets, err := kv.ListBuckets(ctx)
for _, b := range buckets {
    fmt.Printf("%s: %d keys, %d bytes\n", b.Name, b.Values, b.Bytes)
}

GetBucketInfo

info, err := kv.GetBucketInfo(ctx, "sessions")

Bucket Handle

bucket := kv.Bucket("sessions")

Get

entry, err := bucket.Get(ctx, "user:123")
// entry.Key, entry.Value ([]byte), entry.Revision, entry.CreatedAt, entry.Operation

Put

Unconditional write. Returns the new revision:

revision, err := bucket.Put(ctx, "user:123", []byte(`{"name":"Alice"}`))

Create (If-Not-Exists)

Write only if the key does not exist:

revision, err := bucket.Create(ctx, "user:456", []byte(`{"name":"Bob"}`))

Update (Compare-and-Swap)

Write only if the revision matches:

revision, err := bucket.Update(ctx, "user:123", []byte(`{"name":"Alice Updated"}`), entry.Revision)

Delete

Soft-delete (tombstone):

err := bucket.Delete(ctx, "user:123")

Purge

Permanently remove key and all history:

err := bucket.Purge(ctx, "user:123")

ListKeys

keys, err := bucket.ListKeys(ctx, "user.*")

Config Management

client.Config()

cfg := client.Config()

Set

Full replacement of a config document:

result, err := cfg.Set(ctx, "app-settings", map[string]any{
    "theme": "dark",
    "maxRetries": 3,
})
// result.Revision

Get

entry, err := cfg.Get(ctx, "app-settings")
// entry.Name, entry.Data, entry.Revision, entry.UpdatedAt

Patch

Shallow merge — only specified keys are updated:

result, err := cfg.Patch(ctx, "app-settings", map[string]any{
    "maxRetries": 5,
})

List

configs, err := cfg.List(ctx)
for _, entry := range configs {
    fmt.Printf("%s: rev %d\n", entry.Name, entry.Revision)
}

Delete

err := cfg.Delete(ctx, "app-settings")

Watch

Stream real-time updates to a config document. Returns a *ConfigWatcher; call Stop() to end the watch.

watcher, err := cfg.Watch(ctx, "app-settings", ironflow.ConfigWatchCallbacks{
    OnUpdate: func(event ironflow.ConfigWatchEvent) {
        fmt.Printf("Config %s updated to rev %d\n", event.Name, event.Revision)
    },
    OnError: func(err error) {
        log.Printf("Watch error: %v", err)
    },
    OnClose: func() {
        log.Println("Watch closed")
    },
})
defer watcher.Stop()

Secrets

Declare secrets in the function config, then access them at runtime:

var MyFunction = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:       "my-function",
    Triggers: []ironflow.Trigger{{Event: "my.event"}},
    Secrets:  []string{"API_KEY", "DB_PASSWORD"},
}, func(ctx ironflow.Context) (any, error) {
    // Get a required secret (returns error if not found)
    var apiKey string
    if err := ctx.Secrets.Get("API_KEY", &apiKey); err != nil {
        return nil, err
    }

    // Check if an optional secret exists
    if ctx.Secrets.Has("OPTIONAL_KEY") {
        var optional string
        ctx.Secrets.Get("OPTIONAL_KEY", &optional)
    }

    return nil, nil
})

Subscriptions

Real-time event subscriptions over WebSocket or gRPC.

Creating a Subscription Client

// From an existing client
subClient := client.CreateSubscriptionClient()     // WebSocket
grpcClient := client.CreateGrpcSubscriptionClient() // gRPC/HTTP streaming

// Or create directly
subClient := ironflow.NewSubscriptionClient(ironflow.SubscriptionClientConfig{
    WSURL:             "ws://localhost:9123/ws",
    AutoReconnect:     true,
    ReconnectDelay:    1 * time.Second,
    MaxReconnectDelay: 30 * time.Second,
    ReconnectBackoff:  2.0,
})

Connect / Close

if err := subClient.Connect(ctx); err != nil {
    log.Fatal(err)
}
defer subClient.Close()

sub, err := subClient.Subscribe(ctx, "events:order.*", &ironflow.SubscribeOptions{
    Replay:          10,            // Replay last 10 events
    IncludeMetadata: true,
    Filter:          "data.total > 100", // CEL expression
    Namespace:       "production",
})

for event := range sub.Events() {
    fmt.Printf("Received: %s %v\n", event.Topic, event.Data)
}

To resume a fan-out subscription after a global event sequence, pass a pointer so sequence 0 remains distinct from an unset cursor:

cursor := uint64(400)
sub, err := subClient.Subscribe(ctx, "events:order.*", &ironflow.SubscribeOptions{
    StartAfterSequence: &cursor,
})

The cursor is mutually exclusive with Replay and ConsumerGroup; the server returns INVALID_ARGUMENT for either combination. A set cursor opts WebSocket and ConnectRPC subscriptions into automatic reconnect, resuming after the last event delivered to Events(). Delivery is at least once, so an event in flight when the transport fails can be delivered again.

The WebSocket client applies Replay only to the initial subscribe request. Reconnects preserve Filter, ConsumerGroup, IncludeMetadata, AckMode, Backpressure, and Namespace, but omit the original replay count. A fan-out subscription without a cursor reconnects at the current tail and can miss events published while it was offline.

Ackable Subscriptions

For consumer groups with manual acknowledgment:

sub, err := subClient.SubscribeAckable(ctx, "order.*", &ironflow.SubscribeOptions{
    AckMode:       ironflow.AckModeManual,
    ConsumerGroup: "order-processors",
})

for event := range sub.Events() {
    if err := processOrder(event); err != nil {
        sub.Nak(event.ID, 5*time.Second) // Redeliver after 5s
        continue
    }
    sub.Ack(event.ID)
}

Entity Stream Subscriptions

sub, err := subClient.SubscribeEntityStream(ctx, "order-123", ironflow.EntitySubscribeOptions{
    EntityType: "order",
    Replay:     10,
})

Pattern Helpers

ironflow.Patterns.AllRuns()              // "system.run.>"
ironflow.Patterns.Run("run_123")         // "system.run.run_123.>"
ironflow.Patterns.RunLifecycle("run_123") // "system.run.run_123.*"
ironflow.Patterns.RunSteps("run_123")    // "system.run.run_123.step.>"
ironflow.Patterns.AllFunctions()         // "system.function.>"
ironflow.Patterns.Function("my-fn")      // "system.function.my-fn.>"
ironflow.Patterns.UserEvent("order.placed") // "events:order.placed"
ironflow.Patterns.AllUserEvents()        // "events:>"
ironflow.Patterns.AllSecrets()           // "system.secret.*"
ironflow.Patterns.Secret("API_KEY")      // "system.secret.API_KEY.*"
ironflow.Patterns.SecretAction("updated") // "system.secret.*.updated"
ironflow.Patterns.Topic("my-topic")      // "topic:my-topic"
ironflow.Patterns.AllTopics()            // "topic:>"

Connection State

subClient.SetConnectionCallback(func(connected bool) {
    if connected {
        fmt.Println("Connected")
    } else {
        fmt.Println("Disconnected")
    }
})

fmt.Println(subClient.IsConnected())
fmt.Println(subClient.State()) // "connecting", "connected", "disconnected", "reconnecting"

Consumer Groups

Load-balanced event delivery across multiple consumers.

CreateConsumerGroup

group, err := client.CreateConsumerGroup(ctx, ironflow.ConsumerGroupConfig{
    Name:              "order-processors",
    Pattern:           "order.*",
    AckMode:           ironflow.AckModeManual,
    Backpressure:      ironflow.BackpressureBuffer,
    MaxInflight:       100,
    MaxRedeliveries:   3,
    RedeliverDelayMs:  5000,
})

JoinConsumerGroup

sub, err := client.JoinConsumerGroup(ctx, "order-processors")
for event := range sub.Events() {
    if err := processOrder(event); err != nil {
        sub.Nak(event.ID, 5*time.Second)
        continue
    }
    sub.Ack(event.ID)
}

ListConsumerGroups

groups, err := client.ListConsumerGroups(ctx)

GetConsumerGroup

group, err := client.GetConsumerGroup(ctx, "order-processors")

DeleteConsumerGroup

err := client.DeleteConsumerGroup(ctx, "order-processors")

Entity Streams

Append domain events to per-entity streams with optimistic concurrency.

AppendStreamEvent

result, err := client.AppendStreamEvent(ctx, "order-123", ironflow.AppendEventInput{
    Name:       "order.item_added",
    Data:       map[string]any{"itemId": "item-789"},
    EntityType: "order",
},
    ironflow.WithExpectedVersion(4),       // Optimistic concurrency
    ironflow.WithAppendIdempotencyKey("add-item-789"),
    ironflow.WithEventVersion(2),          // Event schema version
)

fmt.Println("New version:", result.EntityVersion) // 5
fmt.Println("Event ID:", result.EventID)

ReadStream

events, err := client.ReadStream(ctx, "order-123",
    ironflow.ReadStreamOpts{
        FromVersion: 1,
        Limit:       50,
        Direction:   "forward", // or "backward"
    },
)

for _, event := range events {
    fmt.Printf("v%d: %s %v\n", event.EntityVersion, event.Name, event.Data)
}

GetStreamInfo

info, err := client.GetStreamInfo(ctx, "order-123")
// info.EntityID, info.EntityType, info.Version, info.EventCount

Projections

Build read models from event streams. Two modes: managed (pure reducer, server stores state) and external (side-effect handler, server tracks position).

Managed Projection

var OrderStats = ironflow.CreateProjection(ironflow.ProjectionConfig{
    Name:   "order-stats",
    Events: []string{"order.*"},
    InitialState: func() map[string]any {
        return map[string]any{"total": 0.0, "count": 0.0}
    },
    Handler: func(state map[string]any, event ironflow.ProjectionEvent, ctx ironflow.ProjectionContext) (map[string]any, error) {
        amount, _ := event.Data["amount"].(float64)
        // State is deep-copied via JSON before each invocation, so every
        // number arrives as float64 — never assert .(int) here.
        count, _ := state["count"].(float64)
        total, _ := state["total"].(float64)
        return map[string]any{"total": total + amount, "count": count + 1}, nil
    },
})

External Projection

var EmailNotifier = ironflow.CreateProjection(ironflow.ProjectionConfig{
    Name:   "email-notifier",
    Events: []string{"order.completed"},
    Handler: func(_ map[string]any, event ironflow.ProjectionEvent, ctx ironflow.ProjectionContext) (map[string]any, error) {
        sendEmail(event.Data["email"].(string), "Order complete!")
        return nil, nil
    },
})

Running Projections

worker := ironflow.NewWorker(ironflow.WorkerConfig{
    Functions:   []ironflow.Function{ProcessOrder},
    Projections: []ironflow.Projection{OrderStats, EmailNotifier},
})
worker.Run(ctx)

ProjectionConfig

Field	Type	Description
`Name`	`string`	Unique projection name
`Events`	`[]string`	Event names to subscribe (supports wildcards)
`Mode`	`ProjectionMode`	`"managed"` or `"external"` (auto-detected from InitialState)
`Handler`	`ProjectionHandler`	Handler function
`InitialState`	`func() map[string]any`	Required for managed mode
`PartitionKey`	`string`	JSONPath for per-partition state
`MaxRetries`	`int`	Default: 3
`BatchSize`	`int`	Default: 100

Event Versioning

Migrate event schemas using an upcaster registry.

registry := ironflow.NewUpcasterRegistry()

// Register v1 → v2 transformer
registry.Register("order.created", 1, 2, func(data json.RawMessage) (json.RawMessage, error) {
    var v1 map[string]any
    json.Unmarshal(data, &v1)
    v1["currency"] = "USD" // Add default currency
    return json.Marshal(v1)
})

// Register v2 → v3 transformer
registry.Register("order.created", 2, 3, func(data json.RawMessage) (json.RawMessage, error) {
    var v2 map[string]any
    json.Unmarshal(data, &v2)
    v2["version"] = "v3"
    return json.Marshal(v2)
})

// Get latest version
latest := registry.LatestVersion("order.created") // 3

// Upcast from v1 → v3 (chains automatically)
upcasted, err := registry.UpcastToLatest("order.created", rawData, 1)

Pass the registry to Serve or NewWorker for automatic upcasting:

handler := ironflow.Serve(ironflow.ServeConfig{
    Functions: []ironflow.Function{ProcessOrder},
    Upcasters: registry,
})

worker := ironflow.NewWorker(ironflow.WorkerConfig{
    Functions: []ironflow.Function{ProcessOrder},
    Upcasters: registry,
})

Webhooks

Transform incoming webhooks into Ironflow events.

var StripeWebhook = ironflow.CreateWebhook(ironflow.WebhookConfig{
    ID: "stripe",
    Verify: func(req *ironflow.WebhookRequest) error {
        // Verify Stripe signature
        sig := req.Header.Get("Stripe-Signature")
        return verifyStripeSignature(req.Body, sig)
    },
    Transform: func(payload []byte) (*ironflow.WebhookEvent, error) {
        var stripeEvent StripeEvent
        if err := json.Unmarshal(payload, &stripeEvent); err != nil {
            return nil, err
        }
        return &ironflow.WebhookEvent{
            Name:           "stripe." + stripeEvent.Type,
            Data:           payload,
            IdempotencyKey: stripeEvent.ID,
        }, nil
    },
})

// Register with Serve
handler := ironflow.Serve(ironflow.ServeConfig{
    Functions: []ironflow.Function{ProcessPayment},
    Webhooks:  []ironflow.Webhook{StripeWebhook},
    ServerURL: "http://localhost:9123",
})
// Webhook endpoint: POST /webhooks/stripe

Pub/Sub Topics

Developer pub/sub for topic-based messaging (separate from workflow event triggers).

Publish (Client)

result, err := client.Publish(ctx, "notifications", map[string]any{
    "type":    "order.shipped",
    "orderId": "123",
},
    ironflow.WithPublishIdempotencyKey("ship-123"),
)
fmt.Println("Event ID:", result.EventID)

ListTopics

topics, err := client.ListTopics(ctx)
for _, t := range topics {
    fmt.Printf("%s: %d messages, %d consumers\n", t.Name, t.MessageCount, t.ConsumerCount)
}

GetTopicStats

stats, err := client.GetTopicStats(ctx, "notifications")
fmt.Printf("Messages: %d, Lag: %d\n", stats.MessageCount, stats.Lag)

Auth Management

API Keys

// Create
key, err := client.CreateAPIKey(ctx, ironflow.CreateAPIKeyInput{
    Name:      "ci-deploy",
    RoleIDs:   []string{"role_admin"},
    ExpiresIn: "90d",
})
fmt.Println("Key:", key.Key) // Only available on create

// List
keys, err := client.ListAPIKeys(ctx)

// Get
info, err := client.GetAPIKey(ctx, "key_id")

// Rotate
newKey, err := client.RotateAPIKey(ctx, "key_id")

// Delete
err := client.DeleteAPIKey(ctx, "key_id")

Organizations

org, err := client.CreateOrganization(ctx, ironflow.CreateOrgInput{Name: "Acme Corp"})
orgs, err := client.ListOrganizations(ctx)
org, err := client.GetOrganization(ctx, "org_id")
org, err := client.UpdateOrganization(ctx, "org_id", ironflow.UpdateOrgInput{Name: "Acme Inc"})
err := client.DeleteOrganization(ctx, "org_id")

Roles

role, err := client.CreateRole(ctx, ironflow.CreateRoleInput{Name: "deployer"})
roles, err := client.ListRoles(ctx)
role, err := client.GetRole(ctx, "role_id")
role, err := client.UpdateRole(ctx, "role_id", ironflow.UpdateRoleInput{Name: "deploy-admin"})
err := client.DeleteRole(ctx, "role_id")

// Assign/remove policies
err := client.AssignPolicyToRole(ctx, "role_id", "policy_id")
err := client.RemovePolicyFromRole(ctx, "role_id", "policy_id")

Policies

policy, err := client.CreatePolicy(ctx, ironflow.CreatePolicyInput{
    Name:      "allow-emit",
    Effect:    "deny",
    Actions:   "emit:*",
    Resources: "*",
    Condition: `request.namespace == "production"`, // optional CEL
})
policies, err := client.ListPolicies(ctx)
policy, err := client.GetPolicy(ctx, "policy_id")
policy, err := client.UpdatePolicy(ctx, "policy_id", ironflow.UpdatePolicyInput{Name: "allow-all-emit"})
err := client.DeletePolicy(ctx, "policy_id")

History Inspection (Audit Trail)

result, err := client.GetAuditTrail(ctx, "run_xyz789",
    ironflow.GetAuditTrailOpts{
        EventType:     "step.completed",
        FromTimestamp: "2025-01-01T00:00:00Z",
        Limit:         100,
        Cursor:        "", // pagination
    },
)

for _, event := range result.Events {
    fmt.Printf("[%s] %s: %v\n", event.CreatedAt, event.EventType, event.Payload)
}
fmt.Println("Total:", result.TotalCount)

Server Introspection

ListFunctions

functions, err := client.ListFunctions(ctx)
for _, fn := range functions {
    fmt.Printf("%s (%s): %s\n", fn.ID, fn.Name, fn.Status)
}

ListWorkers

workers, err := client.ListWorkers(ctx)
for _, w := range workers {
    fmt.Printf("Worker %s: %d active jobs, functions: %v\n", w.ID, w.ActiveJobs, w.FunctionIDs)
}

Health

status, err := client.Health(ctx)
fmt.Println("Server status:", status) // "healthy"

GetCapabilities

caps, err := client.GetCapabilities(ctx)
fmt.Println("Version:", caps.Version)
fmt.Println("Transports:", caps.Transports)
fmt.Println("Features:", caps.Features)

DetectTransport

transport, err := client.DetectTransport(ctx)
// "grpc" or "websocket"

Admin Sub-Clients

Six resource namespaces hang off the client as sub-clients rather than as top-level methods. (client.KV(), client.Config() and client.Secrets() are sub-clients too; they have their own sections above.)

client.Projects()

Projects and the environments inside them.

projects, err := client.Projects().List(ctx)
project, err := client.Projects().Create(ctx, ironflow.CreateProjectInput{Name: "checkout"})
project, err := client.Projects().Update(ctx, "proj_id", ironflow.UpdateProjectInput{Name: "checkout-v2"})
err := client.Projects().Delete(ctx, "proj_id")

envs, err := client.Projects().ListEnvironments(ctx)
env, err := client.Projects().CreateEnvironment(ctx, ironflow.CreateEnvironmentInput{Name: "staging"})
env, err := client.Projects().UpdateEnvironment(ctx, "env_id", ironflow.UpdateEnvironmentInput{Name: "stage"})
err := client.Projects().DeleteEnvironment(ctx, "env_id")

client.Users()

user, err := client.Users().Create(ctx, ironflow.CreateUserInput{
    Email:    "dev@example.com",
    Name:     "Dev",
    Password: os.Getenv("NEW_USER_PASSWORD"),
    Roles:    []string{"developer"},
})
users, err := client.Users().List(ctx)
user, err := client.Users().Get(ctx, "user_id")

// Name and Email are pointers — leave one nil to keep it unchanged.
newName := "Dev Two"
user, err := client.Users().Update(ctx, "user_id", ironflow.UpdateUserInput{Name: &newName})
err := client.Users().Delete(ctx, "user_id")
err := client.Users().ChangePassword(ctx, "user_id", ironflow.ChangePasswordInput{
    CurrentPassword: currentPassword,
    NewPassword:     newPassword,
})

client.Tenants()

tenants, err := client.Tenants().List(ctx)
result, err := client.Tenants().Provision(ctx, ironflow.ProvisionTenantInput{
    OrgName: "Acme Corp",
    EnvName: "production",
})

client.Projections()

Materialized state plus the operational lifecycle. CreateProjection above defines a projection; this drives one that is already registered.

p := client.Projections()

state, err := p.Get(ctx, "order-totals")   // add ironflow.WithPartition(...) for partitioned state
statuses, err := p.List(ctx)
status, err := p.GetStatus(ctx, "order-totals")

job, err := p.Rebuild(ctx, "order-totals")
job, err = p.GetRebuildJob(ctx, "order-totals")
err = p.CancelRebuild(ctx, "order-totals")

err = p.Pause(ctx, "order-totals")
err = p.Resume(ctx, "order-totals")
err = p.Delete(ctx, "order-totals")

// SQL projections (PostgreSQL backend)
rows, err := p.ExecuteSQL(ctx, "SELECT * FROM proj_board WHERE status = 'OPEN'")

client.Webhooks()

The server-side webhook source registry that the dashboard and delivery tracking read. Distinct from CreateWebhook, which defines an in-process handler — see Webhooks.

wh := client.Webhooks()

// ingestToken is returned only here and on rotate (ADR 0048) — capture it now.
src, err := wh.CreateSource(ctx, ironflow.CreateWebhookSourceInput{
    Name:        "Stripe production",
    EventPrefix: "stripe",
})

sources, err := wh.ListSources(ctx)
current, err := wh.GetSource(ctx, src.ID)

// Name is required on every update. ExpectedUpdatedAt is optimistic
// concurrency — ABORTED if the row moved.
src, err = wh.UpdateSource(ctx, ironflow.UpdateWebhookSourceInput{
    ID:                src.ID,
    Name:              "Stripe production (EU)",
    ExpectedUpdatedAt: current.UpdatedAt,
})

src, err = wh.RotateSecret(ctx, ironflow.RotateWebhookSecretInput{ID: src.ID, VerifySecret: "whsec_new"})
src, err = wh.ExpireSecretPrev(ctx, src.ID)
src, err = wh.DisableSignatureVerification(ctx, src.ID, nil)
src, err = wh.RotateIngestToken(ctx, ironflow.RotateWebhookIngestTokenInput{ID: src.ID})

deliveries, total, err := wh.ListDeliveries(ctx, ironflow.ListWebhookDeliveriesOpts{
    SourceID: src.ID,
    Status:   "failed",
    Limit:    25,
})

err = wh.DeleteSource(ctx, src.ID)

ExpireSecretPrevWithInput and DisableSignatureVerificationWithInput take the full input struct when you need ExpectedUpdatedAt optimistic concurrency.

client.Schemas()

The event schema registry. See Event Versioning for how upcasters use these schemas at read time.

schema, err := client.Schemas().Register(ctx, ironflow.RegisterSchemaInput{Name: "order.placed"})
schemas, err := client.Schemas().List(ctx)
schema, err := client.Schemas().Get(ctx, "order.placed")
schema, err := client.Schemas().GetVersion(ctx, "order.placed", 2)
err := client.Schemas().Delete(ctx, "order.placed", 2)
result, err := client.Schemas().TestUpcast(ctx, ironflow.TestUpcastInput{})

Additional Client Methods

The rest of the Client surface, grouped by resource. Each takes ctx first.

Method	Signature (after `ctx`)	Returns
Functions
`GetFunction`	`functionID string`	`*RegisteredFunction`
`DeleteFunction`	`functionID string`	`error`
`UpdateFunctionStatus`	`functionID string, status FunctionStatus`	`*RegisteredFunction`
`ListFunctionHistory`	`functionID string, opts ListFunctionHistoryOptions`	`*ListFunctionHistoryResult`
`GetFunctionAtVersion`	`functionID string, version int64`	`*FunctionHistoryEntry`
`RollbackFunction`	`functionID string, version int64, changeReason string`	`*RegisteredFunction`
Events
`ListEvents`	`opts ListEventsOptions`	`*ListEventsResult`
`GetEvent`	`eventID string`	`*StoredEvent`
`ListEventNames`	`opts ListEventNamesOptions`	`*ListEventNamesResult`
`TriggerBatch`	`events []TriggerBatchEvent`	`[]EmitResult`
Runs
`GetRunSteps`	`runID string`	`*RunStepsResult`
`GetRunStreams`	`runID string`	`*RunStreamsResult`
Entity streams
`ListStreams`	—	`[]StreamListEntry`
`GetEntityHistory`	`entityID string`	`[]EntityHistoryEntry`
`CreateSnapshot`	`entityID string, input CreateSnapshotInput`	`*StreamSnapshot`
`GetSnapshot`	`entityID string`	`*StreamSnapshot`
Projections
`ListProjectionPartitions`	`name string, opts ListProjectionPartitionsOptions`	`*ListProjectionPartitionsResult`
`WaitForProjection`	`name string, opts WaitForProjectionOpts`	`*WaitResult`
`WaitForProjections`	`items []WaitItem, timeout time.Duration`	`[]WaitItemResult`
`WaitForProjectionStream`	`name string, opts WaitForProjectionOpts`	`<-chan WaitProgress, func()`
Pub/Sub
`UpdateConsumerGroup`	`name string, input UpdateConsumerGroupInput, opts ...ConsumerGroupOption`	`*ConsumerGroup`
Auth & audit
`ListRolePolicies`	`roleID string`	`[]PolicyInfo`
`ListAuditEvents`	`opts ListAuditEventsOpts`	`*AuditTrailResult`
Agents
`ListAgentTools`	`cursor string`	`*ListAgentToolsResult`
Escape hatch
`RestRequest`	`method, path string, body any, result any`	`error`

RestRequest sends an arbitrary REST call through the client’s configured transport, auth, and retry policy. Use it for a route the SDK has no wrapper for.

Signature Verification

Verify webhook request signatures.

// Generate a signature
signature := ironflow.SignPayload(payloadString, secret)
// "t=1234567890,v1=abcdef..."

// Verify a signature
err := ironflow.VerifySignature(payload, signature, secret, ironflow.DefaultSignatureTolerance)

// Boolean check
valid := ironflow.IsValidSignature(payload, signature, secret, 5*time.Minute)

// Parse signature header
params, err := ironflow.ParseSignature(header)
// params.Timestamp, params.Signatures["v1"]

// Compute expected signature
expected := ironflow.ComputeSignature(payload, secret, timestamp)

Error Handling

Built-in Errors

import "github.com/sahina/ironflow-go/ironflow"

// Check error types
var stepErr *ironflow.StepError
if errors.As(err, &stepErr) {
    fmt.Println("Step failed:", stepErr.StepID, stepErr.StepName)
}

var timeoutErr *ironflow.StepTimeoutError
if errors.As(err, &timeoutErr) {
    fmt.Println("Step timed out:", timeoutErr.StepName, timeoutErr.Timeout)
}

var invokeErr *ironflow.InvokeError
if errors.As(err, &invokeErr) {
    fmt.Println("Invoke failed:", invokeErr.FunctionID, invokeErr.ChildRunID)
}

// Check if retryable
if ironflow.IsRetryable(err) {
    // Will be automatically retried
}

Non-Retryable Errors

Mark errors as non-retryable to prevent automatic retries:

result, err := ironflow.Run(ctx, "validate", func() (any, error) {
    if !isValid(data) {
        return nil, ironflow.NewNonRetryableError("invalid data - do not retry")
    }
    return data, nil
})

// Or wrap an existing error
return nil, ironflow.WrapNonRetryable(fmt.Errorf("permanent failure: %w", err))

Sentinel Errors

ironflow.ErrFunctionNotFound
ironflow.ErrRunNotFound
ironflow.ErrInvalidSignature
ironflow.ErrSignatureExpired
ironflow.ErrMissingSignature
ironflow.ErrTimeout
ironflow.ErrValidation
ironflow.ErrUnauthorized
ironflow.ErrEnterpriseLicenseRequired
ironflow.ErrForbidden
ironflow.ErrConflict   // HTTP 409 / Connect AlreadyExists
ironflow.ErrContended  // HTTP 409 / Connect Aborted

ErrConflict and ErrContended are the two Connect codes that both serialize to 409, with opposite advice (#2074). AlreadyExists becomes ErrConflict — an identical request is already in flight, so wait rather than retry; a 409 carrying no Connect code (every REST route) lands here too. Aborted becomes ErrContended — a concurrent write won and nothing was applied, so re-read and reissue. Neither is Retryable: resending the identical body is wrong in both cases.

ErrEnterpriseLicenseRequired (HTTP 402) is legacy, retained for wire compatibility. Ironflow ships a single build with no Core/Enterprise split (ADR 0015), so the server never returns 402.

Logging

// Create a logger
logger := ironflow.NewLogger(ironflow.LoggerConfig{
    Level:  ironflow.LogLevelDebug, // debug, info, warn, error, silent
    Prefix: "[my-app]",
})

logger.Info("Starting...", "port", 9123)
logger.Debug("Debug data", "key", "value")
logger.Warn("Slow query", "duration", "500ms")
logger.Error("Failed", "err", err)

// Disable logging
noopLogger := ironflow.NewNoopLogger()

// Parse level from string
level := ironflow.ParseLogLevel("debug")

// Get level from IRONFLOW_LOG_LEVEL env var
level := ironflow.GetLogLevel()

The Logger interface can be implemented with any logger (slog, zap, zerolog):

type Logger interface {
    Debug(msg string, args ...any)
    Info(msg string, args ...any)
    Warn(msg string, args ...any)
    Error(msg string, args ...any)
}

Constants & Environment Variables

Environment Variables

Variable	Description	Default
`IRONFLOW_SERVER_URL`	Server URL	`http://localhost:9123`
`IRONFLOW_SIGNING_KEY`	Request signing key	-
`IRONFLOW_API_KEY`	API key for authentication	-
`IRONFLOW_LOG_LEVEL`	Log level (`debug`, `info`, `warn`, `error`, `silent`)	`info`

Helper Functions

ironflow.GetServerURL()           // from IRONFLOW_SERVER_URL or default
ironflow.GetWebSocketURL(baseURL) // converts http→ws, adds /ws path
ironflow.GetSigningKey()          // from IRONFLOW_SIGNING_KEY
ironflow.GetAPIKey()              // from IRONFLOW_API_KEY

Default Constants

Constant	Value
`DefaultPort`	`9123`
`DefaultHost`	`"localhost"`
`DefaultServerURL`	`"http://localhost:9123"`
`DefaultWebSocketURL`	`"ws://localhost:9123/ws"`
`DefaultClientTimeout`	`30s`
`DefaultFunctionTimeout`	`10m`
`DefaultEmitSyncTimeout`	`30s` — the default wait budget for both `EmitSync` and `InvokeSync`
`DefaultRetryMaxAttempts`	`3`
`DefaultRetryInitialDelay`	`1s`
`DefaultRetryBackoffFactor`	`2.0`
`DefaultRetryMaxDelay`	`5m`
`DefaultClientRetryMaxAttempts`	`3`
`DefaultClientRetryInitialDelay`	`100ms`
`DefaultClientRetryMaxDelay`	`10s`
`DefaultClientRetryBackoffMultiplier`	`2.0`
`DefaultClientRetryConnectionDelay`	`2s`
`DefaultWorkerMaxConcurrentJobs`	`10`
`DefaultWorkerHeartbeatInterval`	`30s`
`DefaultWorkerReconnectDelay`	`5s`
`DefaultSignatureTolerance`	`5m` — the timestamp window `VerifySignature` accepts

Complete Example

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/sahina/ironflow-go/ironflow"
)

type OrderData struct {
    OrderID string  `json:"orderId"`
    Amount  float64 `json:"amount"`
    Email   string  `json:"email"`
}

var ProcessOrder = ironflow.CreateFunction(ironflow.FunctionConfig{
    ID:       "process-order",
    Name:     "Process Order",
    Triggers: []ironflow.Trigger{{Event: "order.placed"}},
}, func(ctx ironflow.Context) (any, error) {
    var order OrderData
    if err := ctx.Event.Data(&order); err != nil {
        return nil, err
    }

    // Step 1: Validate
    _, err := ironflow.Run(ctx, "validate", func() (any, error) {
        if order.Amount <= 0 {
            return nil, ironflow.NewNonRetryableError("invalid amount")
        }
        return map[string]any{"valid": true}, nil
    })
    if err != nil {
        return nil, err
    }

    // Step 2: Sleep
    if err := ironflow.Sleep(ctx, "delay", 5*time.Second); err != nil {
        return nil, err
    }

    // Step 3: Process
    result, err := ironflow.Run(ctx, "process", func() (any, error) {
        return map[string]any{
            "orderId": order.OrderID,
            "status":  "processed",
        }, nil
    })
    if err != nil {
        return nil, err
    }

    return result, nil
})

func main() {
    // Create client and emit events
    client := ironflow.NewClient(ironflow.ClientConfig{
        ServerURL: ironflow.GetServerURL(),
    })

    // Register function
    if err := client.RegisterFunction(context.Background(), ProcessOrder); err != nil {
        log.Printf("Registration: %v", err)
    }

    // Start as worker
    worker := ironflow.NewWorker(ironflow.WorkerConfig{
        Functions: []ironflow.Function{ProcessOrder},
    })

    ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM)
    defer cancel()

    fmt.Println("Worker starting...")
    if err := worker.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

Testing

ironflowtest

Use the ironflowtest package to run a whole function without a server. Mocks are required: Run and Invoke fail if no mock is registered, Sleep resolves immediately, and WaitForEvent consumes from the SendEvent queue.

import (
    "testing"

    "github.com/sahina/ironflow-go/ironflow"
    "github.com/sahina/ironflow-go/ironflow/ironflowtest"
)

func TestProcessOrder(t *testing.T) {
    tc := ironflowtest.NewClient(t, ironflowtest.Config{
        Functions: []ironflow.Function{ProcessOrder},
    })

    tc.MockStep("validate", func() (any, error) {
        return map[string]any{"valid": true}, nil
    })
    tc.MockInvoke("payment-service", func(input any) (any, error) {
        return map[string]any{"txId": "tx_123"}, nil
    })
    tc.SendEvent("order.approved", map[string]any{"approved": true})

    run := tc.Emit(t, "order.placed", map[string]any{"orderId": "123"})

    if run.Status != "completed" {
        t.Fatalf("expected completed, got %s: %v", run.Status, run.Error)
    }
    _ = run.Output                    // function return value
    _ = run.StepOutput("validate")    // one step's output
    _ = run.Steps                     // []TestStep{Name, Type, Output, Error}
    _ = run.CompensationsRan          // compensation step names, in order
}

See the Go SDK README for the full behavior matrix and field tables.

NewContextForTest

Use NewContextForTest to unit test a handler in isolation:

ctx := ironflow.NewContextForTest(&ironflow.PushRequest{
    RunID:      "test-run-1",
    FunctionID: "process-order",
    Event: ironflow.PushEvent{
        ID:   "evt-1",
        Name: "order.placed",
        Data: json.RawMessage(`{"orderId":"123","amount":99.99}`),
    },
})

result, err := ProcessOrder.Handler(ctx)

Go SDK

Installation

Quick Start

Define a Function

Push Mode (Serverless)

Pull Mode (Worker)

Function Definition

CreateFunction

Function Context

Event Data

CreateHandler (Simplified API)

Step Primitives

Run

Sleep

SleepUntil

WaitForEvent

Parallel

Map

Branch-scoped steps

Invoke

InvokeAsync

Compensate (Saga Pattern)

Publish (Step)

Client

NewClient

Event Emission

Emit

EmitSync

InvokeSync

Run Management

GetRun

ListRuns

CancelRun

PatchStep

History Editing (Scoped Injection)

History Navigation (Time-Travel Debugging)

RegisterFunction

Serve Handler

Serve

Worker

NewWorker

Worker Methods

NewStreamingWorker

Graceful Shutdown

Idempotency

KV Store

client.KV()

CreateBucket

DeleteBucket

ListBuckets

GetBucketInfo

Bucket Handle

Get

Put

Create (If-Not-Exists)

Update (Compare-and-Swap)

Delete

Purge

ListKeys

Config Management

client.Config()

Set

Get

Patch

List

Delete

Watch

Secrets

Subscriptions

Creating a Subscription Client

Connect / Close

Subscribe

Ackable Subscriptions

Entity Stream Subscriptions

Pattern Helpers

Connection State

Consumer Groups

CreateConsumerGroup

JoinConsumerGroup

ListConsumerGroups