operator

README

operator

operator is a small, opinionated Go library for running application operations with clear transactional boundaries, synchronous domain events, and simple adapters for exposing operations via HTTP.

It's a disciplined way to execute business logic:

• every request runs inside an operation, decoupled from HTTP
• transactions are lazy and explicit
• domain events are synchronous and transactional
• side effects run only after commit
• adapters (HTTP, CLI, Lambda etc) stay thin

What is a "Transaction"?

In operator, a transaction is an in-process consistency boundary - typically an SQL transaction.

This library is designed for systems where application state lives primarily in a single database that provides transactional guarantees. Distributed transactions, two-phase commit, sagas, and cross-system coordination are explicitly out of scope.

That said, operator is compatible with common patterns for crossing process boundaries. In particular, synchronous domain events make it easy to implement patterns such as the transactional outbox, where events are recorded transactionally and picked up by external systems using techniques such as CDC.

The goal is not to solve distributed consistency, but to make local consistency boring.

When Should You Use operator?

operator is a good fit when:

• your service has non-trivial business logic
• you care about transactional correctness
• you want domain events to be part of your consistency model
• you don't want transactional mechanisms leaking into HTTP handlers
• you prefer explicit control over framework magic

It works especially well for CRUD+ systems that have grown beyond "simple handlers", where concerns like auditing, invariants, and side effects start to accumulate.

You probably don’t need operator if:

• your service is a thin proxy or simple read-only API
• you rely heavily on distributed workflows or eventual consistency
• your persistence layer doesn’t support transactions at all

operator doesn’t try to be a framework - it provides a small, well-defined execution model for application services, and then gets out of the way.

Core Concepts

Operation

type Operation[Tx, I, O] func(ctx *operator.OpContext[Tx], input *I) (*O, error)

An operation is a unit of application work, with typed input/output, explicit access to an operation context, and no hidden global state. If an operation returns an error (or panics), everything rolls back.

Transactions

Not every operation requires a persitence, so transactions do not start automatically. Instead they are initialised lazily on first use:

tx, err := ctx.Tx()

Transactions are implemented by adopting a simple 2-method interface: Commit(context.Context) and Rollback(context.Context) so it's trivial to adapt operator to whatever persistence system you're using.

Hub

hub := operator.NewHub(beginTransaction)

Hub is the operator's central configuration object. It knows how to start a transaction, as well as maintaining a registry of event handlers. All operations are invoked through a Hub.

Domain Events

ctx.Emit(&UserCreated{ID: id})

Domain events enable an application's subdomains to react to actions that occur in other parts of the system.

Events are buffered until the emitting operation completes, and then dispatched to all registered handlers synchronously, before commit. If any event handler fails, the entire transaction is rolled back. Thus, events exist with operator's consistency boundary - they are not simply "fire and forget".

After-Commit Hooks

ctx.AfterFunc(func(ctx *OpContext[Tx]) {
    sendWelcomeEmail(...)
})

After-commit hooks run only after an operation succeeds (and after commit, if a transaction was started). They are intended for side-effects such as sending emails, enqueuing jobs, or triggering webhooks.

Basic Usage Example

1. Define a transaction type

First thing we need is a transaction type, since everything else in operator is generic with respect to it. Assuming we're using database/sql, we can easily adapt an *sql.Tx to match operator's requirements:

type Tx struct {
    *sql.Tx
}

func (t Tx) Commit(ctx context.Context) error { return t.Tx.Commit() }
func (t Tx) Rollback(ctx context.Context) error { return t.Tx.Rollback() }

The adapter is necessary because operator's methods accept a context.Context (Note: if you're using pgx, its transaction type will drop right in without the need for an adapter!)

2. Create a Hub

Next we need a Hub. This fulfils two roles:

1. provides a place from where transactions can be created
2. maintains a registry of event handlers

Note: event handlers are permitted to access the operation's transaction. This is fine - event handling is part of the consistency boundary.

hub := operator.NewHub(func(ctx context.Context) (Tx, error) {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil { return Tx{}, err }
    return Tx{tx}, nil
})

hub.RegisterEventHandler(&UserCreated{}, func(ctx *operator.OpContext[Tx], evt *UserCreated) error {
    tx, err := ctx.Tx()
    if err != nil { return err }
    return insertAuditLog(tx, "user created", evt.ID)
})

3. Define an Operation

An Operation is just a Go function that accepts an *operator.OpContext[Tx] and input arguments, and returns an output value or error.

Note that operation implementations never commit or roll back transactions directly - operator takes care of this as part of its operation invocation handler.

type CreateUserInput struct {
    Email string
}

type CreateUserOutput struct {
    ID int64
}

func CreateUser(ctx *operator.OpContext[Tx], in *CreateUserInput) (*CreateUserOutput, error) {
    tx, err := ctx.Tx()
    if err != nil { return nil, err }

    id, err := insertUser(tx, in.Email)
    if err != nil { return nil, err }

    ctx.Emit(&UserCreated{ID: id})

    return &CreateUserOutput{ID: id}, nil
}

4. Invoke the Operation

out, err := operator.Invoke(ctx, hub, CreateUser, &CreateUserInput{
    Email: "test@example.com",
})

That's all there is to it - transaction handling and event dispatch is handled automatically.

Binding Operations to HTTP

The httpbind package exposes operations over HTTP without leaking concerns - zero business logic in the handlers.

func HandleCreateUser(w http.ResponseWriter, r *http.Request) {
    httpbind.Bind(hub, CreateUser).
        WithInputMapper(httpbind.ParseJSON[CreateUserInput]).
        WithJSONOutput(func(w http.ResponseWriter, out *CreateUserOutput) any {
            return map[string]any{"id": out.ID}
        }).
        Go(w, r)
}

Input, output, and error mapping is fully configurable and can be as simple or as complex as you need. Whether your input and output types map directly to JSON, or if you require something deeper, operator can adapt.

At the moment, only the stdlib's HTTP handler signature is supported - support for more frameworks will be added soon (PRs gladly accepted!).

Copyright & License

© 2026 Jason Frame, licensed under the MIT license.

Documentation

Index

• Variables
• func Invoke[Tx Transaction, I any, O any](ctx context.Context, hub *Hub[Tx], op Operation[Tx, I, O], input *I) (*O, error)
• func InvokeTx[Tx Transaction, I any, O any](ctx context.Context, hub *Hub[Tx], op TxOperation[Tx, I, O], input *I) (*O, error)
• type AfterFunc
• type Event
• type Hub
  • func NewHub[Tx Transaction](transactionProvider TransactionProvider[Tx]) *Hub[Tx]
  • func (h *Hub[Tx]) BeginOperation(ctx context.Context) *OpContext[Tx]
  • func (h *Hub[Tx]) RegisterEventHandler(event Event, hnd any)
• type OpContext
  • func (o *OpContext[T]) AfterFunc(fn AfterFunc[T]) error
  • func (o *OpContext[T]) Emit(evt Event) error
  • func (o *OpContext[T]) Tx() (T, error)
• type Operation
• type Transaction
• type TransactionProvider
• type TxOperation

Variables

var (
	ErrInvalidState               = errors.New("invalid state")
	ErrEventHandlerCoercionFailed = errors.New("failed to create event handler")
)

var ErrRecovered = errors.New("operation recovered from panic")

Functions

func Invoke

func Invoke[Tx Transaction, I any, O any](ctx context.Context, hub *Hub[Tx], op Operation[Tx, I, O], input *I) (*O, error)

InvokeTx() executes the supplied operation with the given input parameters.

The supplied *Hub is used as a transaction provider and event dispatcher.

Returns the operation's output on success, or error on failure.

func InvokeTx

func InvokeTx[Tx Transaction, I any, O any](ctx context.Context, hub *Hub[Tx], op TxOperation[Tx, I, O], input *I) (*O, error)

InvokeTx() begins a transaction then executes the supplied operation with the given input parameters. Prefer InvokeTx() over Invoke() to reduce boilerplate if the operation is guaranteed to use a transaction.

The supplied *Hub is used as a transaction provider and event dispatcher.

Returns the operation's output on success, or error on failure.

Types

type AfterFunc

type AfterFunc[Tx Transaction] func(*OpContext[Tx])

AfterFunc is a function that runs after an operation has successfully completed.

type Event

type Event interface {
	EventName() string
}

type Hub

type Hub[Tx Transaction] struct {
	// contains filtered or unexported fields
}

A Hub is the central object through which operations are invoked, comprising a transaction provider, and a registry of event handlers.

Once a Hub is configured, use the package-level Invoke() function to invoke operations.

func NewHub

func NewHub[Tx Transaction](transactionProvider TransactionProvider[Tx]) *Hub[Tx]

NewHub() returns a hub configured with a transaction provider.

func (*Hub[Tx]) BeginOperation

func (h *Hub[Tx]) BeginOperation(ctx context.Context) *OpContext[Tx]

Begin a new operation and returns its context. User code will usually not call BeginOperation directly; use Invoke().

func (*Hub[Tx]) RegisterEventHandler

func (h *Hub[Tx]) RegisterEventHandler(event Event, hnd any)

RegisterEventHandler() registers a handler to handle events whose type matches reflect.TypeOf(event).

The event handler hnd must be a function conforming to one of the following signatures, wherein *OpContext[Tx] must be assignable to C (this includes context.Context), and the event type must be assignable to E:

func(E) func(C, E) func(E) error func(C, E) error

Event handlers are invoked *after* the operation has returned, but before the transaction is committed. If an event handler returns an error, the transaction aborts and is rolled back - this is by design; event handlers are not intended for "fire and forget" use - use AfterFunc() for that.

type OpContext

type OpContext[T Transaction] struct {
	context.Context
	// contains filtered or unexported fields
}

OpContext represents the context of in-process operation including its current state, associated Hub, and active transaction (if any).

An OpContext is responsible for managing the lifecycle of its associated operation, including commit/rollback handling, event dispatch, and AfterFunc invocation. This functionality is not exposed publicly, instead, coordination is delegated to Invoke().

OpContext wraps context.Context so can be passed to any method that expects one of these.

func (*OpContext[T]) AfterFunc

func (o *OpContext[T]) AfterFunc(fn AfterFunc[T]) error

Register a function to be invoked upon completion of the operation. The callback is invoked after the transaction (if any) is committed. After callbacks can be registered by the main operation, as well as any triggered event handlers.

func (*OpContext[T]) Emit

func (o *OpContext[T]) Emit(evt Event) error

Register an event to be dispatched upon completion of the operation.

func (*OpContext[T]) Tx

func (o *OpContext[T]) Tx() (T, error)

Return the operation's transaction, creating a new transaction if not already started.

type Operation

type Operation[Tx Transaction, I any, O any] func(ctx *OpContext[Tx], input *I) (*O, error)

Operation represents a single operation with defined input/output parameters.

type Transaction

type Transaction interface {
	comparable

	Commit(context.Context) error
	Rollback(context.Context) error
}

type TransactionProvider

type TransactionProvider[Tx Transaction] func(context.Context) (Tx, error)

TransactionProvider is a transaction factory

type TxOperation

type TxOperation[Tx Transaction, I any, O any] func(ctx *OpContext[Tx], tx Tx, input *I) (*O, error)

TxOperation represents a single operation with an implied transaction, and defined input/output parameters. Use TxOperation to reduce boilerplate if your operation is guaranteed to start a transaction.