events

README

events

An event system over asynq: application code dispatches events, listeners subscribe by event type, and each listener runs as an independent asynq task with its own retry budget. One event fans out to N listener tasks, so a failing listener retries without re-running the others.

Two subpackages extend the core dispatch model: outbox implements the transactional-outbox pattern (with outbox/bunx for projects using the bun query builder), and scheduler adds cron/interval-scheduled events and jobs, run replica-safely via a Valkey lease.

events is developed as part of the gp-system tooling. It depends on github.com/gp-system/queue for the Valkey/asynq envelope and task-type conventions, and (via outbox) on github.com/gp-system/dbx's pg and bunx driver subpackages for the outbox table's persistence.

The problem it solves

Dispatching an event and reacting to it in a background task usually gets tangled with delivery guarantees: dispatch directly to Valkey and an event is lost if the process dies between a database commit and the enqueue; dispatch inside the same transaction as the write and there is no delivery mechanism at all. events separates the three concerns: a Dispatcher interface for how an event gets published, a Registry for what listens to it, and (via outbox) an at-least-once delivery guarantee for the cases where losing an event is not acceptable.

Install

go get github.com/gp-system/events

go get github.com/gp-system/events/outbox        # transactional outbox (pgx-native)
go get github.com/gp-system/events/outbox/bunx   # transactional outbox (bun)
go get github.com/gp-system/events/scheduler      # cron/interval scheduling

Usage

Defining and dispatching an event

type ArticlePublished struct {
	ID string `json:"id"`
}

func (ArticlePublished) EventName() string { return "news.article_published" }

reg := events.NewRegistry()
events.Listen(reg, "notify-subscribers", func(ctx context.Context, ev ArticlePublished) error {
	return notifySubscribers(ctx, ev.ID)
})

dispatcher := events.NewDispatcher(queueClient)
dispatcher.Dispatch(ctx, ArticlePublished{ID: "42"})

Three Dispatcher implementations trade off delivery guarantees against simplicity:

• NewDispatcher enqueues directly to Valkey. Simple, but an event dispatched after a database commit is lost if the process dies before the enqueue.
• outbox.NewDispatcher writes to a Postgres outbox table in the caller's transaction; a Relay forwards committed rows to Valkey. Use this whenever events accompany database writes.
• NewSyncDispatcher runs listeners inline, in the caller's goroutine and transaction, for tests and local tooling.

Delivery is at-least-once end to end (outbox -> relay -> asynq -> listener), so listeners must be idempotent. events.MetaFromContext(ctx).ID is a stable idempotency key across retries and across all listeners of the same event.

outbox: transactional delivery

store := outbox.NewStore(db)         // github.com/gp-system/dbx/pg
// or: store := bunx.NewStore(db)    // github.com/gp-system/dbx/bunx

dispatcher := outbox.NewDispatcher(store)

tx.WithinTransaction(ctx, func(ctx context.Context) error {
	if err := articles.Insert(ctx, a); err != nil {
		return err
	}
	return dispatcher.Dispatch(ctx, ArticlePublished{ID: a.ID})
})

A Relay polls the outbox table and forwards committed rows to Valkey. It is safe to run on N replicas: each poll claims its batch with FOR UPDATE SKIP LOCKED, and a deterministic asynq task ID closes the crash-after-enqueue-before-mark window.

relay := outbox.MustNewRelay(ctx, cfg.Outbox, pool, cfg.Valkey)
go relay.Run(ctx)
defer relay.Shutdown(ctx)

outbox.MigrationSQL is the embedded goose migration that creates the outbox_events table; write it into a project's migrations directory once.

scheduler: cron and interval events

sched := scheduler.New().
	Daily(GenerateDailyReport{}).
	Job("cleanup", "0 3 * * *", cleanupHandler)

runner, err := scheduler.NewRunner(ctx, cfg.Scheduler, cfg.Valkey, sched)
go runner.Run(ctx)
defer runner.Shutdown(ctx)

Runner campaigns for a Valkey lease; only the holder runs asynq's scheduler, so with N worker replicas exactly one fires each tick.

Design rules

• Listeners are independent asynq tasks. One event fans out to N listener tasks, each with its own retry budget, so a failing listener retries without re-running the others.
• At-least-once, always. Every dispatch path can redeliver; Meta.ID is the idempotency key listeners are expected to use.
• The outbox is the boundary between "in a transaction" and "on the wire." outbox.Store only inserts a row; only the Relay talks to Valkey, so the write path never depends on Valkey being reachable.
• The scheduler owns leadership, not the schedule. Schedule is pure data (what fires and when); Runner is the only thing that decides which replica acts on it.

Documentation

Overview

Package events is an event system over asynq: application code dispatches events, listeners subscribe by event type, and each listener runs as an independent asynq task with its own retry budget. One event fans out to N listener tasks, so a failing listener retries without re-running the others.

Delivery is at-least-once end to end (outbox → relay → asynq → listener), so LISTENERS MUST BE IDEMPOTENT. Use Meta(ctx).ID as an idempotency key when a side effect must happen at most once.

Dispatch paths:

• NewDispatcher enqueues directly to Valkey. Simple, but an event dispatched after a DB commit is lost if the process dies before the enqueue.
• events/outbox.NewDispatcher writes to a Postgres outbox table in the caller's transaction; a relay forwards committed rows to Valkey. Use this whenever events accompany database writes.
• NewSyncDispatcher runs listeners inline — for tests and local tooling.

Index

• func ContextWithMeta(ctx context.Context, m Meta) context.Context
• func Listen[T Event](r *Registry, name string, fn func(ctx context.Context, ev T) error, ...)
• func NewEnvelope(ctx context.Context, ev Event) (queue.Envelope, error)
• type DirectDispatcher
  • func NewDispatcher(client *queue.Client, opts ...queue.Option) *DirectDispatcher
  • func (d *DirectDispatcher) Dispatch(ctx context.Context, ev Event) error
• type Dispatcher
• type Event
• type FanoutTask
• type Meta
  • func MetaFromContext(ctx context.Context) (Meta, bool)
• type Registry
  • func NewRegistry() *Registry
  • func (r *Registry) EventNames() []string
  • func (r *Registry) FanoutTasks(env queue.Envelope) ([]FanoutTask, error)
  • func (r *Registry) Handler(taskType string) (func(context.Context, queue.Envelope) error, bool)
• type SyncDispatcher
  • func NewSyncDispatcher(r *Registry) *SyncDispatcher
  • func (d *SyncDispatcher) Dispatch(ctx context.Context, ev Event) error

Functions

func ContextWithMeta

func ContextWithMeta(ctx context.Context, m Meta) context.Context

ContextWithMeta returns a context carrying m. Used by the worker and the sync dispatcher; listeners read it with MetaFromContext.

func Listen

func Listen[T Event](r *Registry, name string, fn func(ctx context.Context, ev T) error, opts ...queue.Option)

Listen registers a typed listener for an event under a name unique per event type (it forms the task type "listener:<event>:<name>" seen in Asynqmon). Options set per-listener enqueue defaults (queue, retry, timeout).

The event name is derived from the zero value of T, so T must be a value type whose EventName() works on the zero value. Registering the same (event, name) twice, or a listener whose EventName panics, fails fast at startup.

func NewEnvelope

func NewEnvelope(ctx context.Context, ev Event) (queue.Envelope, error)

NewEnvelope builds an envelope for ev: a fresh id, the JSON-encoded payload, the trace context carried by ctx, and the current time. Both the direct and the outbox dispatcher use it, so a dispatched event looks identical however it is delivered.

Types

type DirectDispatcher

type DirectDispatcher struct {
	// contains filtered or unexported fields
}

DirectDispatcher is a Dispatcher that enqueues events directly to Valkey. See NewDispatcher.

func NewDispatcher

func NewDispatcher(client *queue.Client, opts ...queue.Option) *DirectDispatcher

NewDispatcher returns a DirectDispatcher that enqueues events directly to Valkey. There is no delivery guarantee if the process dies between a database commit and the enqueue — use events/outbox for state-changing flows. opts apply to every dispatched event's fan-out task.

func (*DirectDispatcher) Dispatch

func (d *DirectDispatcher) Dispatch(ctx context.Context, ev Event) error

Dispatch implements Dispatcher.

type Dispatcher

type Dispatcher interface {
	Dispatch(ctx context.Context, ev Event) error
}

Dispatcher publishes events. See the package doc for the implementations.

type Event

type Event interface {
	EventName() string
}

Event is implemented by application event structs. EventName must be stable across deploys (it is persisted in the outbox and in Valkey) and must work on the zero value of the type — register a value type, not a pointer. Use a versioned, dotted name, e.g. "news.article_published".

type FanoutTask

type FanoutTask struct {
	Listener string
	Task     *asynq.Task
	Opts     []queue.Option
}

FanoutTask is one listener's task, produced when an event fans out.

type Meta

type Meta struct {
	// ID is the envelope id — a stable idempotency key across retries and
	// across all listeners of the same event.
	ID string
	// Name is the event name.
	Name string
	// OccurredAt is when the event was dispatched.
	OccurredAt time.Time
	// Attempt is the current retry count: 0 on first delivery.
	Attempt int
}

Meta carries per-delivery metadata into a listener. The worker injects it before invoking the listener; the sync dispatcher injects it too.

func MetaFromContext

func MetaFromContext(ctx context.Context) (Meta, bool)

MetaFromContext returns the delivery metadata injected for the current listener invocation.

type Registry

type Registry struct {
	// contains filtered or unexported fields
}

Registry maps event names to their listeners. Build it in the worker binary and pass it to worker.Run; NewSyncDispatcher also takes one for tests.

func NewRegistry

func NewRegistry() *Registry

NewRegistry returns an empty registry.

func (*Registry) EventNames

func (r *Registry) EventNames() []string

EventNames returns the registered event names, sorted.

func (*Registry) FanoutTasks

func (r *Registry) FanoutTasks(env queue.Envelope) ([]FanoutTask, error)

FanoutTasks expands an event envelope into one task per registered listener. Each task carries a deterministic TaskID (envelope id + listener name) so a re-run of the fan-out after a partial failure is idempotent.

func (*Registry) Handler

func (r *Registry) Handler(taskType string) (func(context.Context, queue.Envelope) error, bool)

Handler returns the listener invocation for a listener task type, and false when no such listener is registered (e.g. a task queued before a deploy that removed the listener).

type SyncDispatcher

type SyncDispatcher struct {
	// contains filtered or unexported fields
}

SyncDispatcher is a Dispatcher that runs every registered listener inline. See NewSyncDispatcher.

func NewSyncDispatcher

func NewSyncDispatcher(r *Registry) *SyncDispatcher

NewSyncDispatcher returns a SyncDispatcher that runs every registered listener inline, in the caller's goroutine and transaction — Laravel's "sync" driver. Listener errors are joined and returned. Intended for unit tests and local tooling, not production (a slow or failing listener blocks the caller).

func (*SyncDispatcher) Dispatch

func (d *SyncDispatcher) Dispatch(ctx context.Context, ev Event) error

Dispatch implements Dispatcher.