Events

import (
    "github.com/gp-system/gpsystem/events"
    "github.com/gp-system/gpsystem/queue"
)

events is gpsystem's events-and-listeners package: every listener is queued from the start, running asynchronously by default. Application code dispatches an event, listeners subscribe by type, and every listener runs as an independent asynq task with its own retry budget: one event fans out to N listener tasks, and a failing listener's retry never re-runs the others.

Delivery is at-least-once end to end (outbox → relay → asynq → listener), so two rules hold the whole thing together: in state-changing flows dispatch through the outbox, and write your listeners to be idempotent.

Defining an event

An event is a plain struct with a stable EventName(). The shop's orderPlaced event was generated by add event shop orderPlaced; you add the payload fields:

package events

type OrderPlaced struct {
    OrderID string `json:"orderId"`
    Email   string `json:"email"`
}

func (OrderPlaced) EventName() string { return "shop.orderPlaced" }

The generator's naming convention is <module>.<eventName> in camelCase, and it also becomes the event:shop.orderPlaced task type. The name is persisted (in the outbox table and in Redis), so keep it stable across deploys; if you rename it, tasks already in flight find no handler under the new deploy.

Dispatch

The Dispatcher is injected into the module's Dependencies (the first add event wires it into both entry points). The shop's ordering rule (in the module's core/ package) dispatches inside the transaction:

func (s *Service) PlaceOrder(ctx context.Context, in PlaceOrderInput) (Order, error) {
    var order Order
    err := s.tx.WithinTransaction(ctx, func(ctx context.Context) error {
        var err error
        if order, err = s.orders.Insert(ctx, in); err != nil {
            return err
        }
        if err = s.stock.Decrement(ctx, in.Items); err != nil {
            return err
        }
        return s.dispatcher.Dispatch(ctx, events.OrderPlaced{
            OrderID: order.ID,
            Email:   order.Email,
        })
    })
    return order, err
}

Three implementations satisfy the events.Dispatcher interface:

Why the outbox is the default, and what's wrong with a direct enqueue on a state change, is spelled out on the outbox page.

The envelope and the delivery metadata

On dispatch, events.NewEnvelope(ctx, ev) builds a queue.Envelope: a fresh UUID (ID), the event name, the JSON payload, the context's trace context and a timestamp. Whichever dispatcher you use, the listener receives the same envelope.

Before invoking a listener, the worker puts delivery metadata from the envelope onto the context: the listener reads it with events.MetaFromContext(ctx):

Listeners

Listeners are registered with the generic events.Listen[T]: the event type comes from the handler's signature, no manual type assertion. add listener shop orderPlaced sendOrderConfirmation --queue mail generates two things. The registration line in the module's listeners/register.go:

package listeners

import (
    "github.com/acme/shop/internal/modules/shop"

    "github.com/gp-system/gpsystem/events"
    "github.com/gp-system/gpsystem/queue"
    // gpsystem:worker-imports
)

func Register(reg *events.Registry, deps shop.Dependencies) {
    events.Listen(reg, "sendOrderConfirmation", sendOrderConfirmation(deps), queue.OnQueue("mail"))
    // gpsystem:listeners
}

And the listener stub in its own file: the body is yours; here is the shop's filled-in version:

package listeners

import (
    "context"

    "github.com/gp-system/gpsystem/mail"
    "github.com/gp-system/gpsystem/mail/mjml"

    "github.com/acme/shop/internal/modules/shop"
    shopevents "github.com/acme/shop/internal/modules/shop/events"
)

func sendOrderConfirmation(deps shop.Dependencies) func(context.Context, shopevents.OrderPlaced) error {
    return func(ctx context.Context, ev shopevents.OrderPlaced) error {
        msg := mail.NewMessage().
            WithTo(ev.Email).
            WithSubject("Your order confirmation").
            WithBody(mjml.Template(templates, "templates/order_confirmation.mjml.tmpl",
                map[string]any{"OrderID": ev.OrderID}))
        return deps.Mailer.Send(ctx, msg)
    }
}

(For the mail/mjml API, see the Mail page.)

What to know about the registration:

• The name (the 2nd argument of Listen) is unique per event; it forms the listener:shop.orderPlaced:sendOrderConfirmation task type, visible in the worker's logs and in Asynqmon. Registering the same (event, name) pair twice panics at startup.
• The remaining arguments are per-listener enqueue options: queue.OnQueue for the queue (the shop's mail goes onto the mail queue so the worker can weight it), queue.MaxRetry, queue.Timeout for the retry budget.
• To subscribe to another module's event, use add listener <module> <producer>.<event> <name>; you import the event type from the producer's events package.

Fan-out: how one event becomes N tasks

The worker expands the event:shop.orderPlaced task and enqueues one listener:... task per listener. Every listener task gets a deterministic TaskID (<envelope-id>:<listener-name>), so if the fan-out is interrupted and re-runs, the already-submitted tasks are no-ops via ErrDuplicate: the fan-out itself is idempotent.

From there each listener lives its own life: it can sit on a different queue, it burns its own retry budget, and its failure re-runs only itself. This always-async-by-default behavior (rather than a synchronous event loop) is the defining trait of the package, and its price is the idempotency requirement.

Idempotency

With at-least-once delivery, a listener can run more than once for the same event (a retry, a worker crash mid-processing, a relay re-send). Meta.ID is the stable key for this: it is identical on every attempt of the same dispatch.

meta, _ := events.MetaFromContext(ctx)
// e.g. a unique key in a "did this happen already?" table, or the mail
// provider's idempotency key:
return deps.Mailer.SendOnce(ctx, meta.ID, msg)

Practical patterns:

• Natural key: when the side effect is a DB write, INSERT ... ON CONFLICT DO NOTHING on the event id (or the business key).
• External API: pass meta.ID as an idempotency-key header when the provider supports one.
• Read + condition: the race between checking "did we already send it?" and the send is only closed by a key-based approach; don't settle for a bare SELECT.

Where listeners run

Listeners run in the worker binary: the per-module listeners subpackages' Register functions are composed into a single events.Registry by the generated cmd/worker/main.go. For scheduled event firing (events dispatched on a cron), see the scheduler; for the generator commands (add event, add listener), the CLI worker generators.

Patterns used

At-least-once delivery + idempotent listener contract (Meta.ID as the dedup key, deterministic TaskID in fan-out): see Design patterns.