Ironflow is pre-1.0: APIs can change between releases. Pin a version. What is ready →

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.

For equivalent Tier-1 capability examples, see the Node SDK and Python SDK references.

Installation

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

Requires Go 1.26+.

The canonical external import is github.com/sahina/ironflow-go/ironflow. Mirror tags match engine releases, so a version can be pinned with go get github.com/sahina/ironflow-go/ironflow@v0.43.1.

Tracing

Configure your application’s OpenTelemetry tracer provider before starting a worker or serving push requests. The SDK uses the OTel API at runtime and does not install an exporter or provider. The instrumentation scope is github.com/sahina/ironflow-go/ironflow.

Each poll job, streaming job, or push execution starts a process consumer span parented on the engine’s W3C traceparent and tracestate. Poll and stream request spans never become job parents. Malformed parents start a new trace; invalid or oversized state is discarded without losing a valid parent. Unsampled contexts retain their trace identity. Baggage is not carried into execution.

Run, StepClient.Run, and RunWithBranch create a step.run child of the process span only when the callback executes. Replayed steps create no span. Timeouts and failures set error status. Cancellation and durable yields leave status unset. Spans end on success, failure, yield, cancellation, timeout, and panics that unwind through the execution goroutine. An unrecovered panic in a timeout callback or parallel branch can terminate the process before spans end. This existing recovery gap is tracked in #3515. Tracing does not change the SDK’s panic recovery behavior. Error events use a generic message so handler errors and panic values do not expose credentials or payloads. Span attributes do not include secrets or business data.

Use Context.Context() or HandlerContext.Context() for application spans and context-aware I/O. Run-level contexts return the process context. Explicit RunWithContext and RunWithBranchContext callbacks receive the child step context; nested scoped calls parent their spans on that context. Legacy step callbacks do not receive a separate step context.

func handle(ctx ironflow.Context) (any, error) {
    goCtx, span := otel.Tracer("orders").Start(ctx.Context(), "validate-order")
    defer span.End()
    return validateOrder(goCtx)
}

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",
    Description: "Sends the welcome email", // optional

    // Event triggers
    Triggers: []ironflow.Trigger{
        {Event: "user.created"},
        {Event: "order.*"},
        {Event: "order.placed", Expression: `data.total > 100`}, // CEL filter
        {Cron: "*/5 * * * *"},                                    // schedule; each tick starts a run
    },

    // Retry configuration. Every field is optional: an unset field is not
    // sent, and the server resolves it when each run is created
    // (default: 25 attempts, 1s, 2.0, 5m, pause on exhaustion).
    Retry: &ironflow.RetryConfig{
        MaxAttempts:   5,
        InitialDelay:  1 * time.Second,
        BackoffFactor: 2.0,
        MaxDelay:      5 * time.Minute,
        OnExhausted:   ironflow.OnExhaustedFail, // or OnExhaustedPause (server default)
    },

    // Push-endpoint circuit breaker. A zero field keeps the server default.
    // The server raises a window under the longest probe (21.3s by default).
    CircuitBreaker: &ironflow.CircuitBreakerConfig{
        FailureThreshold: 5,
        OpenDuration:     time.Minute,
        MaxOpenDuration:  10 * time.Minute,
    },

    // Soft start-rate limit: at most Limit runs start per Period (per Key
    // value when Key is set). Runs over it wait in the queue. Retries are not
    // throttled. Limit >= 1, Period >= 1s.
    Throttle: &ironflow.ThrottleConfig{Limit: 100, Period: time.Minute, Key: "customerId"},

    // Execution timeout. Unset sends nothing and the engine applies its own
    // limits (push is bounded by the push timeout, pull is unlimited).
    Timeout:     30 * time.Minute,
    StepTimeout: 60 * time.Second, // Default timeout for all steps

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

    // Execution mode
    Mode: ironflow.PullMode, // or ironflow.PushMode (default); a Worker registers every function as pull regardless

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

    // Collapse rapid-fire events into one run after a quiet period.
    // Async only; Period must be at least 1s. MaxWait is an optional cap.
    Debounce: &ironflow.DebounceConfig{Period: 5 * time.Second, Key: "userId", MaxWait: time.Minute},

    ActorKey:    "event.data.userId",                  // JSON path; stored by the server, not used for routing yet
    EndpointURL: "https://app.example.com/api/ironflow", // HTTP endpoint for push mode
    Metadata:    map[string]any{"team": "payments"},   // custom metadata

    // Audit recording: which event families to capture.
    // RecordingProfileAll (run lifecycle, steps and compensations),
    // RecordingProfileRunLifecycle (run.created and run.status_changed) or
    // RecordingProfileSteps (steps and compensations). Empty keeps the
    // legacy Recording bool.
    RecordingProfile: ironflow.RecordingProfileSteps,

    // Check the payload before the handler runs. A returned error fails the
    // run without a retry. Cron ticks skip it. Local only: the server never sees it.
    Validate: func(data json.RawMessage) error { return nil },

    // 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, MaxAttempts, OnExhausted, StartedAt, Environment)
    Secrets SecretsReader // Read-only access to resolved secrets
}

To make a Client call from inside a handler count as part of the run, pass ctx.RunContext() as its context.Context. It derives from context.Background(). To keep the cancellation of another context, use ironflow.WithRunID(parent, ctx.Run.ID).

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:   "customerId",
        },
    },
})

The HandlerContext provides:

ctx.Event     // Typed event data
ctx.EventMeta // Event metadata (ID, Name, Version, Timestamp, IdempotencyKey, Source, Metadata)
ctx.Run       // Run info (ID, FunctionID, Attempt, MaxAttempts, OnExhausted, StartedAt, Environment)
ctx.Secrets   // SecretsReader
ctx.Step      // StepClient (Run, Sleep, SleepUntil, WaitForEvent, Parallel, Map, Compensate, Invoke, InvokeAsync, Publish)
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))

A step output and the function’s return value must encode as JSON. A value that does not encode (NaN, ±Inf, a channel, a function or a cycle) fails the run with the non-retryable code SERIALIZATION_ERROR, so compensations run. The steps before the bad one keep their outputs.

Callback panics propagate to the goroutine that called Run, including when a timeout is configured. Parallel, Map, and their branch variants join started branches before propagating an ordinary panic to their caller. Ordinary panics take precedence over durable yields, even with OnError: "allSettled". If multiple started branches panic, the lowest branch index wins. A handler can recover these panics and run deferred cleanup.

A timeout does not stop the callback. If the callback panics after the timeout has returned, the SDK contains that panic and keeps the recorded timeout failure. The SDK does not log or serialize recovered panic values. Callbacks must still manage their own cancellation and side effects.

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.

Matching retained events stored since the run was created can satisfy the wait even if they arrive before it parks. See event wait timing for the environment, ordering, retention, and event reuse rules. agent.Approve uses the same engine behavior.

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
    // MatchValue: "ord_123",       // optional literal compared at Match; empty snapshots Match from the triggering event (needs Match)
})

For a timeout branch use WaitForEventOptional(ctx, name, filter), returning (*Event, error). An elapsed deadline returns nil, nil, persisted as a completed JSON null and replayed on every later execution. In parallel branches use WaitForEventOptionalWithBranch. These methods need no type argument. The existing value-returning wait methods retain fail-on-timeout behavior, and agent.Approve keeps its current default.

event.EmittedBy is the authenticated emitter that the engine recorded, for example apikey:<id> or user:<id>. The engine sets it from the credential of the caller. It is empty when the engine reports none. agent.ApproveResult.EmittedBy carries the same value. ApproveResult.Approver is different: it is free text from the event data, and the caller chooses it.

The type argument is required. T appears in no parameter, so Go cannot infer it, and the call does not compile without it. When the timeout ends, the scheduler fails the run. The timeout does not reach your code as err.

Sleep, SleepUntil, WaitForEvent, Invoke and InvokeAsync suspend the run with a panic of an internal yield value. Do not call recover() around them in your handler. recover() swallows the yield, and the run finishes without a wait. Only Run returns ordinary errors.

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.WaitForEventOptional`	`ironflow.WaitForEventOptionalWithBranch`
`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),
)

A failed or timed-out child returns *InvokeError. IsRetryable reports it as non-retryable. If you return it, the parent run fails at once, with no retries.

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
}

Pass an idempotency key to make the server drop a duplicate publish. The key applies to one topic in one environment. The server keeps it for IRONFLOW_RUN_RETENTION_DAYS days after the first publish (kept forever when unset or 0). Without one, the step sends a default key derived from its step ID; Publish also hashes the payload into it, so a root-scoped call inside Parallel never drops another branch’s message. At most 255 bytes, with no whitespace or control characters; the server rejects anything else with InvalidArgument.

err := ironflow.Publish(ctx, "notifications", payload,
    ironflow.WithPublishIdempotencyKey(order.ID))

Client

Timestamps

Every timestamp a client call returns is a time.Time, or a *time.Time when the server can omit it (ADR 0165). An unreadable value, or a missing value in a time.Time field, fails the call with INVALID_RESPONSE. A fake server in a test must send real timestamps. StreamSnapshot.CreatedAt is nil on the CreateSnapshot result, because the server answers that call with the snapshot ID only.

A time.Time keeps nanoseconds, so a value you read can go straight back to the server: ExpectedUpdatedAt: &source.UpdatedAt and EntityHistoryOpts{To: entry.Timestamp} both compare exactly. Timestamp filters such as GetAuditTrailOpts.FromTimestamp take a *time.Time, and nil leaves that side open. Compare two values with Equal, not ==, because == also compares the time zone and the monotonic clock reading.

RunEvent and PushEvent stay strings: they are frame and push payloads kept as sent.

EntityHistoryOpts.From and To are plain time.Time values, and zero leaves that side open.

A time.Time writes to JSON with the shortest fraction, so a projection that stores event.Timestamp holds .12Z for .120Z. It is the same instant, but the text does not sort or compare as time: 07.123Z sorts before 07.12Z. Parse before you compare. To sort as text, store a fixed width: event.Timestamp.UTC().Format("2006-01-02T15:04:05.000000000Z").

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

    // Optional: your own *http.Client in place of the SDK default.
    HTTPClient: &http.Client{},

    // Optional: sent as X-Ironflow-Environment-ID on the client's requests,
    // KV, config and webhooks included. Secret routes send their own
    // "current" value, and subscriptions do not send it. Empty sends no
    // header. The client does not read IRONFLOW_ENV. In a handler, pass
    // ctx.Run.Environment to use the run's environment.
    Environment: "env_default_staging",

    // Retry configuration (optional). The client waits at least the server's
    // Retry-After, but never longer than MaxDelay. It resends a write only on 429.
    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"}),
)

// With a namespace
result, err := client.Emit(ctx, "order.placed", data,
    ironflow.WithEmitNamespace("production"),
)

Emit returns when the server has stored the event and created the runs. It does not wait for the runs.

Keep event names in const values in one package, for example events.OrderPlaced = "order.placed", and use them in triggers, projections and Emit. Webhook event names are set at run time, so they can be built from strings.

ironflow.Context has no client. To emit from inside a function, build a client with NewClient and call Emit inside an ironflow.Run step, so that the emit is memoized.

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),
)

The timeout is a positional argument, not an option. 0 uses DefaultEmitSyncTimeout.

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: InvokeSync 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")

DeleteRun

Permanently delete a terminal run and its steps:

err := client.DeleteRun(ctx, "run_xyz789")

The run must already be completed, failed, or cancelled — a run still in flight is refused with failed_precondition; cancel it first. The delete cannot be undone, and it needs the runs:delete action, which only the admin role holds.

DeleteRuns

Delete terminal runs matching a filter in the current environment, and return how many THIS call removed:

until := time.Now().AddDate(0, -1, 0)
deleted, err := client.DeleteRuns(ctx, ironflow.RunDeleteFilter{
    FunctionID: "process-order",
    Status:     ironflow.RunStatusCompleted,
    Until:      &until,
})

One call stops at 10,000 runs, so a wide filter can never become a single unbounded sweep. Re-issue the same filter until it returns 0; deleted rows no longer match, so the same call resumes where it stopped. A count below 10000 does NOT mean drained — a page comes back short whenever another transaction holds some matching rows.

At least one field is required — an empty filter is rejected with invalid_argument rather than matching every run. Status, when set, must be terminal, and Until matches runs created before it. The server deletes in batches of 1000.

RedactEvent, RedactStep, RedactRun

Redaction replaces a stored payload with a placeholder. The row survives with its id, ordering and version intact, so replay, memoization and optimistic concurrency are unaffected; only the bytes go. It cannot be undone, and it costs events:redact — admin by default.

{"$redacted": true, "sha256": "<hex of the original bytes>", "redactedAt": "<RFC 3339>"}

All three calls are idempotent. Five things they do not reach: projection state, which is yours and is never rewritten — rebuild the projection to make it forget; a run’s copy of the event payload (input), which only RedactRun removes; a step’s own payload columns on a run redaction, and the run’s copy of a step’s output, which only the other call removes — redacting a step usually leaves a copy in the run’s output, so call both; a copy already published to a stream, which expires on its own retention window (at most 7 days), including a publish past its final read of the row when the redaction commits; and an object-storage copy awaiting successful cleanup. Run and step markers authorize repeated cleanup while the server and blob backend are available, with no cleanup deadline. Event snapshot cleanup is best effort; a failed sweep can leave copies behind without automatic retry. See Redacting an Event’s Payload.

// The event's data. Snapshots of an entity stream derived from it are dropped
// with it. No terminal-run guard: consumers seeing the placeholder is the
// contract, so a pending publish delivering it is correct.
err := client.RedactEvent(ctx, "ev_01JD2KQ7X8Y9Z")

// The step's output, original output and error, its parent run's error,
// its attempt errors and its audit payloads. Sibling errors are unchanged.
// stepID is the step's ROW id, not its name. The run must be terminal.
err = client.RedactStep(ctx, "step_01JD2KQ7X8Y9Z")

// The run's input, output and error, every child step's error,
// plus every attempt error and audit payload for
// the run and its steps. The run must be terminal.
err = client.RedactRun(ctx, "run_abc123")

Error envelopes are replaced whole, including message, code, stack, details and encrypted envelopes. Run redaction leaves step outputs unchanged; use step redaction for those. Stored error placeholders survive later whole-row saves.

A non-terminal run is refused with failed_precondition: a memoized resume reads steps.output and a resume reads runs.input, so redacting under a live run would change results rather than surface a placeholder.

Use ironflow.IsRedacted in a reducer before trusting a payload. Upcasters need no guard — UpcasterRegistry skips a redacted payload and returns it untouched, so the marker reaches the reducer intact:

if ironflow.IsRedacted(event.Data) {
    return state, nil // keep what we had
}

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 RecordingProfile: ironflow.RecordingProfileAll (or the legacy 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: webhook sources
    Webhooks: []ironflow.Webhook{StripeWebhook},

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

    // Optional: environment for durable Publish steps and ctx.Run.Environment.
    // Default: IRONFLOW_ENV, else no header (the API key's environment).
    Environment: "env_default_staging",

    // 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,
    Environment:        "env_default_staging", // default: IRONFLOW_ENV, else the API key's environment
    CheckpointInterval: time.Second,          // default: 1s (polling worker only)
    PollWait:           20 * time.Second,     // default: 20s, max 25s; negative disables (polling worker only)
    DrainTimeout:       30 * time.Second,     // default: 30s
    APIKey:             "ifkey_...",          // default: IRONFLOW_API_KEY
    OnError: func(err error, ctx ironflow.ErrorContext) { // runs when a job fails; does not change retries
        log.Printf("%s failed (attempt %d): %v", ctx.FunctionID, ctx.Attempt, err)
    },
})

Environment must match the API key. The worker sends the value (or IRONFLOW_ENV) as X-Ironflow-Environment-ID. If the API key is scoped to a different environment, the server answers 403. A name answers 400. The worker logs the value it uses at start. The value scopes worker requests, projections and durable Publish steps. The handler reads it as ctx.Run.Environment, and agent memory uses it. For a Client built in a handler, pass ClientConfig{Environment: ctx.Run.Environment}. agent.ExposeMcp takes its own ExposeMcpConfig.Environment (default IRONFLOW_ENV).

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.

Workers track active assignments by job ID and execution sequence. If recovery assigns the same job to this worker with a new sequence, both executions keep their own slot until they finish. An older completion cannot remove the newer execution from heartbeats. An exact redelivery of an active sequence is ignored. This applies to polling and ConnectRPC streaming workers; streaming cancellation by job ID signals each locally tracked execution of that job. A canceled streaming execution cannot start another SDK step. Cancellation does not forcibly stop arbitrary handler code already running outside SDK steps.

Worker Methods

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

// Graceful drain (wait up to DrainTimeout for active jobs, then cancel them)
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 transport is a single persistent HTTP/2 connection instead of polling.

Differences from NewWorker.

A job the worker cannot run. The engine tracks MaxConcurrentJobs, but it can send more jobs than that, for example after a reconnect. The worker refuses a job that arrives when it is full or during a drain. It sends a nack, and the engine makes the job runnable again at once. A refusal does not use a retry attempt. The engine does not send more jobs to a draining worker. For each refusal from a full worker, the engine holds back one of that worker’s slots. It frees the slot when the worker’s heartbeat shows a free slot. With an engine older than the nack, the engine ignores the refusal. It then recovers the job after the lease expires (about 90 s).
Graceful close. When the active jobs are done, a drain sends the queued results, half-closes the stream and waits for the engine to close it, within the same deadline. Stop() closes the stream immediately.
Stream lost during a drain. The worker stops and does not reconnect. Without a stream it cannot report results.
Drain timeout. An engine shutdown message starts a drain. The worker uses the drain timeout in the message (the engine sends 25 seconds). If the message has no drain timeout, the worker uses DrainTimeout. The Python SDK default for drain_timeout is 60 seconds, not 30.

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.

Upgrade order. Upgrade the engine before the streaming workers, and upgrade Go and Node streaming workers in the same rollout. An older Go or Node streaming SDK against a newer engine reads a failed invoke as success with empty output.

Graceful Shutdown

Drain() accepts no new jobs and waits up to DrainTimeout (default 30 seconds) for the active jobs. At the deadline it cancels the jobs that remain and stops, so the engine can reclaim their leases.

A cancelled Run context also starts a drain, and Run returns the context error when the drain is done. This applies to NewWorker and NewStreamingWorker:

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

if err := worker.Run(ctx); err != nil && 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.*")

The filter is a required positional argument. Pass "" to list all keys. The Go KV client does no marshaling: values are []byte, and you encode and decode JSON yourself.

Watch

Stream key changes over WebSocket. Returns a *KVWatcher; call Stop() to end the watch. Omit WithWatchKey to watch every key in the bucket.

watcher, err := bucket.Watch(ctx, ironflow.KVWatchCallbacks{
    OnUpdate: func(event ironflow.KVWatchEvent) {
        fmt.Printf("%s %s rev %d\n", event.Operation, event.Key, event.Revision)
    },
    OnError: func(err error) { log.Printf("Watch error: %v", err) }, // the watch stops after an error
    OnClose: func() { log.Println("Watch closed") },
}, ironflow.WithWatchKey("user.*"))
defer watcher.Stop()

File Storage

Buckets, files, signed URLs and file events. The FilesClient has no timeout on byte transfers: use the context to cancel. See the Files guide.

files := client.Files()               // Files() is a METHOD on *Client
maxBytes, emit, signed := int64(10<<20), true, true
_, err := files.CreateBucket(ctx, "inbox", ironflow.FileBucketConfig{
    MaxObjectBytes: &maxBytes, EmitEvents: &emit, AllowSignedURLs: &signed,
})

inbox := files.Bucket("inbox")
info, err := inbox.Put(ctx, "scans/a.pdf", f, size, ironflow.PutFileOptions{ContentType: "application/pdf"})

obj, err := inbox.Get(ctx, "scans/a.pdf", ironflow.GetFileOptions{IfMatch: info.ETag})
defer obj.Body.Close()

FilesClient methods:

Method	Description
`CreateBucket(ctx, name, FileBucketConfig)`	Create a bucket. A nil field takes the server default.
`ListBuckets(ctx)`	List the buckets.
`GetBucket(ctx, name)`	One bucket’s configuration.
`UpdateBucket(ctx, name, FileBucketConfig)`	Change the set fields. The change applies to new writes only.
`DeleteBucket(ctx, name)`	Delete an empty bucket. A bucket with files returns `ErrConflict`.
`Bucket(name)`	A `*FileBucket` handle.

FileBucket methods:

Method	Description
`Put(ctx, path, body, size, PutFileOptions)`	Upload `size` bytes. `ContentType` is required. `IfMatch` and `IfNoneMatch` make the write conditional. An `io.ReadSeeker` body is retried. Any other reader is sent once.
`Get(ctx, path, GetFileOptions)`	A `*FileObject` with `Body`, `ContentType`, `ETag` and `Size`. Close `Body`. `IfMatch` and `Range` are options.
`Info(ctx, path)`	A file’s metadata, without the bytes.
`List(ctx, ListFilesOptions)`	A page of files. `NextCursor` is empty on the last page.
`Delete(ctx, path)`	Delete a file. A missing path is not an error.
`Move(ctx, from, to, MoveFileOptions)`	Rename one file. `ToBucket` and `IfMatch` are options.
`Copy(ctx, from, to, CopyFileOptions)`	Copy one file. `ToBucket` is an option.
`MovePrefix(ctx, from, to)`	Move every file under a prefix (both end in `/`). Returns the count.
`SignUpload(ctx, path, SignUploadOptions)`	A `*SignedURL` for a credential-free upload. Set `CreateOnly` to make the PUT fail with `412` if the path exists.
`SignDownload(ctx, path, ttl)`	A `*SignedURL` for a credential-free download.

Errors: ErrPreconditionFailed (412), ErrPayloadTooLarge (413), ErrUnsupportedMediaType (415) and ErrConflict (409). Use errors.Is. A 404 is an *IronflowError with Code == "not_found".

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

Create and Update

Conditional writes. Create fails if the name exists. Update replaces the config only if revision is its current revision. A failed condition returns an *IronflowError with code failed_precondition.

result, err := cfg.Create(ctx, "app-settings", map[string]any{"theme": "dark"})
result, err = cfg.Update(ctx, "app-settings", map[string]any{"theme": "light"}, result.Revision)

PatchIf(ctx, name, data, revision) and DeleteIf(ctx, name, revision) apply the same condition to Patch and Delete.

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
})

client.Secrets()

Manage the secrets of the current environment. Values are write-only: no call returns one, and Secret.Value stays empty. A function reads a value through ctx.Secrets.

s := client.Secrets()

secret, err := s.Set(ctx, "API_KEY", "sk_live_...")   // create
secret, err = s.Update(ctx, "API_KEY", "sk_live_new") // replace the value
meta, err := s.Get(ctx, "API_KEY")                    // metadata only
list, err := s.List(ctx)                              // []SecretListEntry: Name, CreatedAt, UpdatedAt

// Rename a secret or change its description without touching the value.
// Set at least one field.
newName := "STRIPE_KEY"
secret, err = s.Patch(ctx, "API_KEY", ironflow.PatchSecretInput{Name: &newName})

err = s.Delete(ctx, newName)

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,
})

// Direct gRPC client. It has no key field, so it reads IRONFLOW_API_KEY.
grpcClient := ironflow.NewGrpcSubscriptionClient(ironflow.GrpcSubscriptionClientConfig{
    ServerURL: "http://localhost:9123",
})

The gRPC client reconnects only for a subscription that sets StartAfterSequence. AutoReconnect on GrpcSubscriptionClientConfig stays for source compatibility and does nothing.

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.

sub.Unsubscribe() stops one subscription. sub.CurrentID() returns the server subscription ID bound to the handle now. The ID can change after a reconnect.

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.

Frames can arrive out of order across nodes. ironflow.NewRunVersionFilter() drops strictly lower positive versions, ranking each run and each step independently by its full identity, including dots in step IDs. Ties and zero or absent versions pass without lowering the remembered version. A deletion is kept and remembers the run ID; subsequent run and step frames for that ID are dropped, including unversioned frames. Other topics pass. The filter is safe for concurrent use and belongs to one subscription. Its shared cache holds at most MaxRuns run and step identities (default 10,000); eviction forgets both ordering and deletion history. ParseRunEvent decodes run payloads, including Version.

keep := ironflow.NewRunVersionFilter()
sub, _ := subClient.Subscribe(ctx, "system.run.>", nil)
for event := range sub.Events() {
    if !keep.Accept(event) {
        continue // an older status that arrived late
    }
    run, _ := ironflow.ParseRunEvent(event)
    fmt.Println(run.ID, run.Status, run.Version)
}

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)      // or sub.Term(event.ID) to reject with no redelivery
}

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

Joins over WebSocket by default. WithJoinTransport("grpc") joins over one PubSubService.SubscribeBidirectional stream instead; both transports can ack, nak and term. The group must exist; a missing group is NotFound. WithJoinNamespace targets another namespace. WithJoinConsumerID is accepted but not sent by the SDK today; the server mints the ack identity.

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)
}

The stream needs a server that includes this change (released after v0.47.0). Against an older server the join fails with an *IronflowError whose Code is ironflow.ErrCodeServerTooOld (SERVER_TOO_OLD). The message names the first server version with the stream (0.48.0). The SDK does not retry it. Use the WebSocket transport there.

With "grpc", Ack, Nak and Term wait for the server result, up to 30 seconds, and return:

*ironflow.AckError{EventID, Code, Message} when the server refuses the ack. Code is "invalid_argument", "not_found" or "internal". An "internal" ack is safe to retry; a "not_found" ack (the group or member is gone) is not. A Nak delay outside 0 to 2³¹−1 ms returns "invalid_argument" without a send.
ironflow.ErrStaleAck when the event arrived before a reconnect, the stream ended while the ack was pending, or this stream has already settled every copy of the event. In the first two cases the server redelivers the event.
ironflow.ErrSubscriptionClosed after Unsubscribe or a terminal error.

On an AUTO group they return nil without sending, and log one warning. The stream lives until ctx is done or Unsubscribe is called; a done ctx goes to Errors() as canceled or deadline_exceeded. A retryable disconnect, or 90 s with no frame from the server, reconnects with backoff (1 s, factor 1.5, up to 30 s). Unauthenticated, PermissionDenied, NotFound and InvalidArgument end the subscription and go to Errors().

Get, list, update and delete take ironflow.WithConsumerGroupNamespace(ns) to target another namespace.

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
    ironflow.WithAppendMetadata(map[string]any{"correlationId": "req-1"}), // Event metadata
)

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

AppendStreamEvent takes one event per call, not a slice. AppendResult.Sequence is always 0. The server publishes the event to NATS later, through the transactional outbox, so it does not know the sequence when the append returns. For read-your-writes, pass result.EventID to client.WaitForEvent(ctx, eventID, projection, opts). It resolves the event ID to a sequence and waits until the projection reaches it. For the first append, skip GetStreamInfo and pass WithExpectedVersion(0).

An entity ID must use only letters, digits and the characters -, _, ., : and ~. The server rejects /, ?, #, %, & and white space when you write. To scope a stream to a tenant, put the tenant in one segment, for example tenant1-issue-42. Or filter by EntityType.

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)
}

ReadStream returns a slice of events. It does not return a {events, totalCount} envelope like the TypeScript SDK.

GetStreamInfo

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

GetStreamInfo returns the server error stream not found as err when the stream does not exist. The TypeScript streams.getInfo returns null in that case.

DeleteStream

Appends a $stream.deleted tombstone at the stream’s current version + 1 and drops its snapshots. Later appends fail with ConnectRPC code failed_precondition. purge also deletes every event row below the tombstone; the tombstone row itself always stays. Returns the tombstone’s version. Requires streams:delete (admin by default).

version, err := client.DeleteStream(ctx, "order-123", false)
// version is the tombstone's version, i.e. the stream's final version

// purge=true additionally deletes the events below the tombstone
version, err = client.DeleteStream(ctx, "order-123", true)

Deleting a stream cannot be undone. Calling it again on an already-deleted stream returns failed_precondition when purge is false; with purge true it re-runs the purge and returns the existing tombstone’s version, so a delete whose cleanup failed can be retried.

Projections do not see the tombstone unless they registered $stream.deleted in their event list — delivery is filtered by event name, and nothing rewrites projection state. See Deleting a Stream.

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)

A worker starts one ProjectionRunner per projection. To run one projection without a worker, build the runner yourself:

runner := ironflow.NewProjectionRunner(OrderStats, "http://localhost:9123",
    map[string]string{"Authorization": "Bearer " + apiKey}, ironflow.NewNoopLogger())
if err := runner.Start(); err != nil {
    log.Fatal(err)
}
defer runner.Stop()

headers goes on every request, so put the credential and X-Ironflow-Environment-ID there. Start registers the projection and then runs the stream loop in a goroutine. Authentication, authorization, and invalid-argument errors stop registration; other registration errors are retried. The runner holds a projection lease while it reads (ADR 0120). The lease is required: on a server older than v0.44.0, which has no lease RPCs, the runner logs a SERVER_TOO_OLD error and stops before it reads. A poll, a stream or a state load refused with failed_precondition, such as a projection bound to a PUBSUB stream that was recreated (ADR 0164) or one whose last rebuild failed, also stops the runner with a logged error instead of a reconnect, because a retry cannot fix it.

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

Verify returns only an error. It does not return the parsed payload. Transform receives the raw body as []byte. Set ServerURL on ServeConfig: the handler emits the transformed event only when ServerURL is not empty, and it has no environment variable fallback. With an empty ServerURL, the handler drops the event.

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

CreateOrganization needs a platform credential (ifplatform_ key or platform session). It calls POST /api/v1/platform/orgs. Get, list, update and delete work on the caller’s own organization.

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")

Policy set

The Rego policy set of the organization. See Authorization policy sets.

current, err := client.GetPolicySet(ctx) // Version 0 when the org has no set

// ifVersion is the version you read (0 for a new set). A new set starts in "monitor".
saved, err := client.PutPolicySet(ctx, ironflow.PolicySetInput{
    Modules: map[string]string{
        "authz.rego": `package ironflow.authz

deny contains "no production" if input.request.environment == "env_prod"
`,
    },
    Mode: "monitor", // omit to keep the current mode
}, current.Version)
if ironflow.IsVersionConflict(err) {
    // Another admin wrote first. Read the set again.
}

err = client.ValidatePolicySet(ctx, input)
result, err := client.DryRunPolicySet(ctx, ironflow.PolicySetDryRunInput{
    Set:   &input, // nil dry-runs the active set
    Input: map[string]any{"request": map[string]any{"action": "runs:delete"}, "subject": map[string]any{"roles": []string{"member"}}},
})
versions, err := client.ListPolicySetVersions(ctx)
v3, err := client.GetPolicySetVersion(ctx, 3)
restored, err := client.RollbackPolicySet(ctx, 3, saved.Version) // keeps the active mode

PolicySet.EnforcementActive is false when the server stores the set but does not enforce it.

Errors carry a code in IronflowError.Code. The constants are ironflow.PolicySetErrCompileError, PolicySetErrV0Syntax, PolicySetErrReservedNamespace, PolicySetErrBuiltinForbidden, PolicySetErrSchemaError, PolicySetErrEntrypointType, PolicySetErrBadFileName, PolicySetErrTooLarge, PolicySetErrCompileTimeout, PolicySetErrBadMode, PolicySetErrBadDecision, PolicySetErrSprintfFormat, PolicySetErrBadRequest, PolicySetErrForbidden, PolicySetErrNotFound, PolicySetErrVersionConflict, PolicySetErrIfMatchRequired, PolicySetErrSelfLockout, PolicySetErrRateLimited, PolicySetErrBusy and PolicySetErrNotPropagated. See Policy set errors.

A 412 matches ironflow.ErrPreconditionFailed, and IsVersionConflict reports it. For a 429 (PolicySetErrRateLimited, PolicySetErrBusy), wait for the Retry-After time before you send the request again.

A 503 PolicySetErrNotPropagated is not retryable. The write committed as the version in the message. Send the same request again with that version as ifVersion (a no-op that finishes propagation). Never send it again with the old version, because that fails with a 412.

History Inspection (Audit Trail)

from := time.Date(2025, 1, 1, 0, 0, 0, 0, time.UTC)
result, err := client.GetAuditTrail(ctx, "run_xyz789",
    ironflow.GetAuditTrailOpts{
        EventType:     "step.completed",
        FromTimestamp: &from,
        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

Lists the HTTP pull workers registered in the environment. In a cluster the list covers every node, and ActiveJobs counts the job leases each worker holds.

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"

Ready

Health is liveness. Ready reads GET /ready, which checks the database, NATS and the engine streams. A server that is not ready returns an *ironflow.IronflowError with code NOT_READY; Details["issues"] names the failing dependencies, such as stream:PUBSUB.

ready, err := client.Ready(ctx)
if err != nil {
    return err // the server is not ready, or unreachable
}
fmt.Println(ready.Status) // "ready"
fmt.Println(ready.Warnings) // non-fatal advisories, if any

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

Seven resource namespaces hang off the client as sub-clients rather than as top-level methods. (client.KV(), client.Files(), 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"})
// Description is a *string on update: nil keeps it, a pointer to "" clears it.
cleared := ""
project, err := client.Projects().Update(ctx, "proj_id", ironflow.UpdateProjectInput{Description: &cleared})
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")
// Use a client authenticated as the user changing their own password.
err := client.Users().ChangePassword(ctx, currentUserID, 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 (SQLite or PostgreSQL; the @@, pgvector and ts_rank forms need PostgreSQL)
rows, err := p.ExecuteSQL(ctx, "SELECT * FROM proj_board WHERE status = 'OPEN'")

// Definition history: the definitions archived before a replacement or deletion.
// Newest first. The current definition is not in the history.
page, err := p.ListHistory(ctx, "order-totals", ironflow.DefinitionHistoryOptions{Limit: 20})
for page.HasMore {
    page, err = p.ListHistory(ctx, "order-totals", ironflow.DefinitionHistoryOptions{
        Limit:          20,
        BeforeRevision: page.NextBeforeRevision, // exclusive cursor; 0 means no more pages
    })
}
entry, err := p.GetRevision(ctx, "order-totals", 3) // entry.PreviousDefinition

Limit defaults to 50 and the maximum is 200.

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)
// TestUpcast returns the version chain in result.StepsApplied and the data
// unchanged: upcasters run in SDK workers, not on the server.
result, err := client.Schemas().TestUpcast(ctx, ironflow.TestUpcastInput{
  EventName: "order.placed", FromVersion: 1, ToVersion: 2, Data: map[string]any{"orderId": "123"},
})

// History is per (event name, version). Same paging rules as projection history.
page, err := client.Schemas().ListHistory(ctx, "order.placed", 2, ironflow.DefinitionHistoryOptions{Limit: 20})
entry, err := client.Schemas().GetRevision(ctx, "order.placed", 2, 1)

client.Deployments()

Immutable executable deployments (ADR 0081). A deployment identifies the code that serves a function. Registration only stages a deployment. It never changes which code serves new runs. See Executable Deployments for the fields, errors and permissions: Register needs functions:register, and the read methods need functions:read.

d := client.Deployments()

res, err := d.Register(ctx, ironflow.RegisterDeploymentInput{
    FunctionID:     "process-order",
    DeploymentID:   "2026-10-16-a",
    ArtifactDigest: "sha256:" + digestHex, // 64 lowercase hex digits
    Snapshot:       json.RawMessage(`{"triggers":[{"event":"order.placed"}],"timeoutMs":30000}`),
    Mode:           ironflow.PullMode,
    // EndpointURL: required for PushMode, must be empty for PullMode
})
// res.Created is false when the identical deployment already existed.
// A different artifact, snapshot, mode or endpoint under an existing
// DeploymentID fails with an `already_exists` *IronflowError.

dep, err := d.Get(ctx, "process-order", "2026-10-16-a")
list, err := d.List(ctx, "process-order", ironflow.ListDeploymentsOptions{Limit: 20}) // newest first
routing, err := d.GetExecutionRouting(ctx, "process-order")
// routing.Mode is ironflow.FunctionRoutingLegacy, FunctionRoutingAdopting or FunctionRoutingVersioned.

Snapshot is the execution snapshot in its ConnectRPC JSON form. Leave it nil to register an empty snapshot. The SDK sends snapshot_schema_version for you: 2 when the snapshot uses circuit_breaker, throttle or retry.on_exhausted, otherwise 1, so a plain snapshot still registers against an engine that predates schema 2.

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`	`[]TriggerBatchResult`
Runs
`GetRunSteps`	`runID string`	`*RunStepsResult`
`GetRunStreams`	`runID string`	`*RunStreamsResult`
Entity streams
`ListStreams`	—	`[]StreamListEntry`
`GetEntityHistory`	`entityID string, opts ...EntityHistoryOpts`	`[]EntityHistoryEntry`
`CreateSnapshot`	`entityID string, input CreateSnapshotInput`	`*StreamSnapshot`
`GetSnapshot`	`entityID string, opts ...GetSnapshotOpts`	`*StreamSnapshot` — the latest snapshot, or with `BeforeVersion` the latest at or before that entity version
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()`
`WaitForEvent`	`eventID, projection string, opts WaitForProjectionOpts`	`*WaitResult` — resolves an event ID to its sequence, then waits
Pub/Sub
`UpdateConsumerGroup`	`name string, input UpdateConsumerGroupInput, opts ...ConsumerGroupOption`	`*ConsumerGroup`
Auth & audit
`ListAuditEvents`	`opts ListAuditEventsOpts`	`*AuditTrailResult`
Agents
`ListAgentTools`	`cursor string`	`*ListAgentToolsResult`
Capacity (platform credentials; see below)
`ListCapacityLanes`	`opts CapacityListOptions`	`[]CapacityLane`
`ListCapacityQueue`	`opts CapacityListOptions`	`[]CapacityQueueEntry`
`ListCapacityLeases`	`opts CapacityListOptions`	`[]CapacityLease`
`ListCapacityBuckets`	`opts CapacityListOptions`	`[]CapacityBucket`
`ListCapacitySessions`	`opts CapacityListOptions`	`[]CapacityWorkerSession`
`ListCapacityCredits`	`opts CapacityListOptions`	`[]CapacityWorkerCredit`
`CapacityStats`	—	`[]CapacityStat`
Debounce
`ListDebounceEntries`	—	`[]DebounceEntry`
`CancelDebounce`	`environmentID, functionID, key string`	`error`
Escape hatch
`RestRequest`	`method, path string, body any, result any`	`error`

UpdateFunctionStatus takes ironflow.FunctionStatusActive, FunctionStatusPaused or FunctionStatusArchived.

The capacity views need platform credentials: an ifplatform_ API key or a platform session. A tenant key gets 403. CapacityListOptions has Environment, Function and Limit (0 means 100, and the server caps it at 1000). The leases and credits views ignore Function. The leases view always returns an empty FunctionID, so match a lease to its run by RunID. CapacityStats returns the totals of every environment.

CancelDebounce takes the raw debounce key and base64url-encodes it for the path. Cancelling an entry that is not armed succeeds.

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.

ironflow.NewCommandDedup[T](ctx, client.KV(), bucketName, ironflow.CommandDedupOptions{TTL: ...}) wraps a KV bucket in claim-first command dedup (TryClaim / Finalize / Release) for command reservations and cached results. A returned entry can be the unfinished claim. The application must distinguish completion and recover pending entries through reconciliation and idempotent effects. Never Release after Finalize.

TryClaim retries the claim up to 3 times when the winner releases between your write and your read, then returns an *IronflowError with Code == "COMMAND_DEDUP_RACE". A create that fails with any other error leaves the claim state unknown, and the helper does not retry it. To let a retry recognize its own orphaned claim, put a token that stays the same across retries in the claim and pass ironflow.TryClaimOptions[T]{IsOwner: func(p T) bool {...}}. IsOwner runs only on a prior entry. It must return false for a finalized result, or a finished command replays. Run one retry lineage at a time: two concurrent callers with the same token both win.

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) {
    // IsRetryable(invokeErr) is false: the parent run fails without retries.
    fmt.Println("Invoke failed:", invokeErr.FunctionID, invokeErr.ChildRunID)
}

// Check if retryable
if ironflow.IsRetryable(err) {
    // The server says a resend can succeed
}

IronflowError.Code is the lowercase Connect code (not_found, invalid_argument, already_exists, permission_denied, unauthenticated, failed_precondition, internal, …) on every transport. A domain code, such as BUCKET_NOT_EMPTY or FUNCTION_NOT_FOUND, keeps its own name. A server reply with no code gets the Connect code of its HTTP status. Match on the code, for example ie.Code == "not_found".

A REST error’s retryable flag comes from the server. When the server sends none, the SDK uses the status: 408, 429 and 5xx are retryable, except 501 and 507. The client resends reads (GET, HEAD, PUT, DELETE, and Connect reads sent as POST, which the protos mark NO_SIDE_EFFECTS) on any retryable error. It resends writes (POST, PATCH) only when the server rate-limited them (HTTP 429 or Connect resource_exhausted). After a 5xx or a network failure, a write may already have been applied, so the client returns the error and does not resend. The error you receive is the same either way.

Retry Delay

NewRetryAfterError(message, after) returns a retryable error that asks the engine to wait after before the next attempt, in place of the retry policy’s backoff. The engine caps the delay at the policy’s max delay. Setting IronflowError.RetryAfter on a retryable error does the same, including when the error is wrapped. The delay is sent as retry_after_ms on push responses, REST job updates and streaming JobFailed, and is ignored on a non-retryable error.

return nil, ironflow.NewRetryAfterError("upstream rate limited", 30*time.Second)

Exhaustion and Compensation

When a retryable failure has used every attempt, the run’s OnExhausted policy decides what happens: pause holds the run for ironflow run resume, fail ends it. The engine sends the policy as ctx.Run.OnExhausted next to ctx.Run.MaxAttempts. Compensations run on a terminal failure: a non-retryable error, or a retryable one on the last attempt (Attempt >= MaxAttempts) when the policy is "fail". With "pause", or against an engine that sends no policy, a retryable error does not compensate.

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))

To build an error of your own, use ironflow.NewError(message, code, retryable), or ironflow.WrapError(err, message, code, retryable) to keep the cause.

Sentinel Errors

ironflow.ErrFunctionNotFound
ironflow.ErrRunNotFound
ironflow.ErrInvalidSignature
ironflow.ErrSignatureExpired
ironflow.ErrMissingSignature
ironflow.ErrTimeout
ironflow.ErrValidation
ironflow.ErrUnauthorized
ironflow.ErrEnterpriseLicenseRequired
ironflow.ErrForbidden
ironflow.ErrPreconditionFailed   // HTTP 412 — an If-Match or If-None-Match condition did not hold
ironflow.ErrPreconditionRequired // HTTP 428 — the write needs an If-Match header and carried none
ironflow.ErrPayloadTooLarge      // HTTP 413 — file storage
ironflow.ErrUnsupportedMediaType // HTTP 415 — file storage
ironflow.ErrConflict   // HTTP 409 / Connect AlreadyExists
ironflow.ErrContended  // HTTP 409 / Connect Aborted
ironflow.ErrInjectionUnverified // Connect Aborted — the step DID change; read it, do not reissue
ironflow.ErrStaleAck            // gRPC consumer-group Ack, Nak or Term for an event that is not pending on this stream
ironflow.ErrSubscriptionClosed  // gRPC consumer-group Ack, Nak or Term after Unsubscribe or a terminal stream error

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.

Correlated function logging

Context.Logger() *slog.Logger uses the application default slog handler and binds the execution context, including for plain Info calls. The existing HandlerContext.Logger delegates to it. Worker Logger configuration remains for SDK diagnostics.

Field	Type	Value
`run_id`	string	Full dispatch run ID
`function_id`	string	Dispatch function ID
`step_id`	string	Actual allocated memoization key during an explicit scoped callback; otherwise empty
`attempt`	integer	Engine run attempt; zero when unknown
`trace_id`	string	Valid current trace ID, including unsampled traces; otherwise empty

func RunWithContext[T any](ctx Context, name string, fn func(Context) (T, error), opts ...StepOption) (T, error)
func RunWithBranchContext[T any](branch *BranchContext, name string, fn func(Context) (T, error), opts ...StepOption) (T, error)

These variants share existing results, options, memoization, and timeout behavior. Nested scoped Run calls and branch creation retain the branch prefix. Other root primitives do not become branch-aware. Legacy zero-argument callbacks retain their captured outer context. Cached scoped loggers lose their step label on callback exit or timeout, while keeping valid inherited trace context. Memoized callbacks are skipped, so they emit no callback logs. Logs outside steps can repeat on resume and have no deduplication or replay-storage guarantee.

Application levels, handler errors, message bodies, explicit errors, and attributes retain native behavior. The five reserved fields override application collisions through the SDK wrapper, including With, and remain at its record root through WithGroup. Opaque grouping or bound attributes already configured inside the application handler remain application-owned.

An application-selected otelslog.NewHandler with otelslog.WithLoggerProvider(provider) receives each accepted record once, with intrinsic trace and span correlation. SDK production code installs no provider or exporter and imports no Logs bridge. Tracing alone does not export logs. Application code owns provider lifetime and any network exporters; Ironflow does not collect function logs. Tested optional application pins are go.opentelemetry.io/contrib/bridges/otelslog v0.21.0 and go.opentelemetry.io/otel/sdk/log v1.47.0, both declaring Go 1.26.

Diagnostic 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`	The Go SDK does not read this variable. Your code reads it and passes it to `Serve` as `SigningKey`. With no key, the handler does not verify requests.	-
`IRONFLOW_API_KEY`	API key for authentication	-
`IRONFLOW_ENV`	Target environment for pull and streaming workers (`WorkerConfig.Environment` overrides it). Also read by `Serve` and `agent.ExposeMcp`. A `Client` does not read it. Must match the API key’s environment.	-
`IRONFLOW_LOG_LEVEL`	Log level (`debug`, `info`, `warn`, `error`, `silent`)	`info`
`IRONFLOW_NO_DISCOVERY`	Set to `1` to turn off the engine discovery file (ADR 0098).	-

When neither IRONFLOW_SERVER_URL nor IRONFLOW_API_KEY is set, the client, worker, Serve and subscription constructors read the nearest .ironflow/engine.json (loopback http URLs only) to fill the unset variables. Environment values always win.

Helper Functions

ironflow.GetServerURL()           // IRONFLOW_SERVER_URL, then default
ironflow.ConfiguredServerURL()    // same, but "" when neither is set
ironflow.GetWebSocketURL(baseURL) // converts http→ws, adds /ws path
ironflow.GetSigningKey()          // from IRONFLOW_SIGNING_KEY
ironflow.GetAPIKey()              // from IRONFLOW_API_KEY
ironflow.ParseDuration("7d")      // "1h", "30m", "7d" -> time.Duration
ironflow.PrintBanner()            // ASCII logo, printed once per process

Default Constants

Constant	Value
`DefaultPort`	`9123`
`DefaultHost`	`"localhost"`
`DefaultServerURL`	`"http://localhost:9123"`
`DefaultWebSocketURL`	`"ws://localhost:9123/ws"`
`DefaultClientTimeout`	`30s`
`DefaultEmitSyncTimeout`	`30s` — the default wait budget for both `EmitSync` and `InvokeSync`
`OnExhaustedPause`, `OnExhaustedFail`	`"pause"`, `"fail"` — values for `RetryConfig.OnExhausted`
`DefaultClientRetryMaxAttempts`	`3`
`DefaultClientRetryInitialDelay`	`100ms`
`DefaultClientRetryMaxDelay`	`10s`
`DefaultClientRetryBackoffMultiplier`	`2.0`
`DefaultClientRetryConnectionDelay`	`2s`
`DefaultWorkerMaxConcurrentJobs`	`10`
`DefaultWorkerHeartbeatInterval`	`30s`
`DefaultWorkerReconnectDelay`	`5s`
`DefaultWorkerPollWait`	`20s` — how long the server may hold an idle job poll
`DefaultSignatureTolerance`	`5m` — the timestamp window `VerifySignature` accepts
`DefaultCommandDedupTTL`	`7d` — the recommended `CommandDedupOptions.TTL`

Enum Constants

Type	Constants (value)
`ExecutionMode`	`PushMode` (`"push"`), `PullMode` (`"pull"`)
`RunStatus`	`RunStatusRunning` (`"running"`), `RunStatusCompleted` (`"completed"`), `RunStatusFailed` (`"failed"`), `RunStatusCancelled` (`"cancelled"`), `RunStatusPaused` (`"paused"`), `RunStatusWaitingForCapacity` (`"waiting_for_capacity"`), `RunStatusWaiting` (`"waiting"`). A server newer than the SDK can send another status, so keep a default branch.
`AckMode`	`AckModeAuto` (`"auto"`), `AckModeManual` (`"manual"`)
`BackpressureMode`	`BackpressureDrop` (`"drop"`), `BackpressureBlock` (`"block"`), `BackpressureBuffer` (`"buffer"`)
`ProjectionMode`	`ProjectionModeManaged` (`"managed"`), `ProjectionModeExternal` (`"external"`)
`ProjectionStatus`	`ProjectionStatusActive`, `ProjectionStatusRebuilding`, `ProjectionStatusPaused`, `ProjectionStatusError` (`"active"`, `"rebuilding"`, `"paused"`, `"error"`)
`EventSourceType`	`EventSourceAPI` (`"api"`), `EventSourceCron` (`"cron"`), `EventSourceWebhook` (`"webhook"`)
`FunctionStatus`	`FunctionStatusActive`, `FunctionStatusPaused`, `FunctionStatusArchived`, `FunctionStatusUnspecified` (`"active"`, `"paused"`, `"archived"`, `"unspecified"`)
`FunctionChangeType`	`FunctionChangeCreated`, `FunctionChangeUpdate`, `FunctionChangeStatusChange`, `FunctionChangeRollback`, `FunctionChangeDelete` (`"created"`, `"update"`, `"status_change"`, `"rollback"`, `"delete"`)
`FunctionRoutingMode`	`FunctionRoutingLegacy` (`"legacy"`), `FunctionRoutingAdopting` (`"adopting"`), `FunctionRoutingVersioned` (`"versioned"`)
`RecordingProfile`	`RecordingProfileAll` (`"all"`), `RecordingProfileRunLifecycle` (`"run_lifecycle"`), `RecordingProfileSteps` (`"steps"`)
`ConsumerGroupStatus`	`ConsumerGroupStatusActive`, `ConsumerGroupStatusPaused`, `ConsumerGroupStatusDeleted` (`"active"`, `"paused"`, `"deleted"`)
`ConnectionState`	`StateConnecting`, `StateConnected`, `StateDisconnected`, `StateReconnecting` (`"connecting"`, `"connected"`, `"disconnected"`, `"reconnecting"`)
`LogLevel`	`LogLevelDebug`, `LogLevelInfo`, `LogLevelWarn`, `LogLevelError`, `LogLevelSilent` (`0` to `4`)

HeaderEnvironmentID is "X-Ironflow-Environment-ID". HeaderRunID is "X-Ironflow-Run-ID", which WithRunID and ctx.RunContext() attach. ErrCodeServerTooOld is "SERVER_TOO_OLD", the IronflowError.Code for a call the server is too old to serve. SDKVersion is the SDK version that a worker reports to the server on registration.

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. An empty queue fails a default wait; WaitForEventOptional and its branch variant instead return nil immediately. SendEvent(name, nil) queues a real event with null data. Custom TestInterceptor.WaitForEventStep implementations accept the optional flag and return *Event, using nil only for the optional timeout outcome.

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

Tracing

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

Timestamps

NewClient

Event Emission

Emit

EmitSync

InvokeSync

Run Management

GetRun

ListRuns

CancelRun

DeleteRun

DeleteRuns

RedactEvent, RedactStep, RedactRun

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

Watch

File Storage

Config Management

client.Config()

Set

Create and Update

Get

Patch

List

Delete

Watch

Secrets

client.Secrets()

Subscriptions

Creating a Subscription Client

Connect / Close