events

Documentation

Overview

Package events is the transactional event substrate: Emit writes the event inside the caller's database transaction (outbox), a river worker picks it up and the dispatcher invokes the kind's handler.

Handler contract: exactly ONE handler per event kind — a second Subscribe for the same kind panics at wiring time. The substrate gives NO ordering across kinds: kinds execute asynchronously through river workers, so any sequencing a business flow needs (step A before step B) belongs inside the handler (or as explicit follow-up events), never in registration order. Multi-listener same-phase cascades with ordering guarantees are a consumer-side bus pattern, not provided here.

Index

• Constants
• func Migrate(ctx context.Context, pool *pgxpool.Pool) error
• func Subscribe(kind string, h Handler)
• type Emitter
  • func New(pool *pgxpool.Pool, wks *workers.Workers, opts ...Option) (*Emitter, error)
  • func (e *Emitter) Client() *river.Client[pgx.Tx]
  • func (e *Emitter) Emit(ctx context.Context, tx pgx.Tx, ev Event, opts ...river.InsertOpts) error
  • func (e *Emitter) EmitAt(ctx context.Context, tx pgx.Tx, ev Event, at time.Time) error
• type Event
• type Handler
• type Option

Constants

const QueueEvents = "events"

Functions

func Migrate

func Migrate(ctx context.Context, pool *pgxpool.Pool) error

Migrate brings the river job tables (river_job etc.) up to date (idempotent). Deliberately NOT part of ploykit migrations (001-999): the river schema follows the river library's own version sequence, managed adaptively by its official migrator - embedding it would pin us to one river version. Call it once in the product boot sequence BEFORE events.New and before workers.Start; deployments with their own migration pipeline may call it from there instead. It is the ONLY sanctioned way the river schema comes into existence — products must not copy the river SQL into their own migrations (a duplicate creation collides with the bookkeeping the migrator maintains).

func Subscribe

func Subscribe(kind string, h Handler)

Types

type Emitter

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

func New

func New(pool *pgxpool.Pool, wks *workers.Workers, opts ...Option) (*Emitter, error)

func (*Emitter) Client

func (e *Emitter) Client() *river.Client[pgx.Tx]

Client exposes the underlying river client for inspection and operation surfaces (job listing, retry, cancel, queue pause — see platform/devx). Emission must keep going through Emit; a nil client (Emitter built without New) also returns nil here.

func (*Emitter) Emit

func (e *Emitter) Emit(ctx context.Context, tx pgx.Tx, ev Event, opts ...river.InsertOpts) error

func (*Emitter) EmitAt

func (e *Emitter) EmitAt(ctx context.Context, tx pgx.Tx, ev Event, at time.Time) error

EmitAt schedules a one-shot event for a specific time (typed sugar over river's ScheduledAt). Use for "do X once at time T"; recurring plans still belong to the schedule domain. Must be called inside a transaction, like Emit.

type Event

type Event struct {
	Kind string

	WorkspaceID uuid.UUID

	Payload []byte

	IDempotencyKey string
}

type Handler

type Handler func(ctx context.Context, ev Event) error

type Option

type Option func(*river.Config)