events

package
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package events provides a CloudEvents 1.0-aligned event bus for maniflex.

The core package is stdlib-only. Broker adapters live in sub-packages:

  • events/inproc — in-process fan-out (tests, single-binary apps)
  • events/outbox — transactional outbox over *sql.DB
  • events/redis — Redis Streams (XADD / XREADGROUP)
  • events/nats — NATS JetStream
  • events/kafka — kafka-go
  • events/rabbitmq — AMQP 0.9.1 (RabbitMQ)
  • events/cloudevents — CloudEvents 1.0 binary/structured codec

Index

Constants

View Source 
const (
	DefaultReadBackoffMin = 100 * time.Millisecond
	DefaultReadBackoffMax = 30 * time.Second
)

Default bounds for ReadBackoff. The floor is short enough that a one-off blip costs almost nothing; the ceiling is the point past which waiting longer stops helping, since a broker that has been down for 30s is an outage rather than a hiccup and the loop should keep checking at a steady, cheap rate.

Variables

This section is empty.

Functions

func Dedupe 

func Dedupe(store DedupeStore) func(Handler) Handler

Dedupe wraps a Handler with idempotent delivery semantics. Events whose ID is already recorded in store are silently dropped. Recommended with every at-least-once broker adapter: 

bus.Subscribe(ctx, events.Subscription{
    Handler: events.Dedupe(store)(myHandler),
})

The ID is claimed before the handler runs, so a concurrent delivery of the same event is dropped rather than processed twice. If the handler then fails, the claim is released so the retry can re-process it — see DedupeReleaser for what happens with a store that cannot release.

func DeliverWithRetry 

func DeliverWithRetry(ctx context.Context, pub Publisher, sub Subscription, e Event)

DeliverWithRetry calls sub.Handler up to sub.MaxRetry+1 times with exponential backoff. If all attempts fail and sub.DLQ is non-empty, the event is re-published under the DLQ type via pub so it flows through the same broker as the original event.

Roadmap §11B.11 / checkpoint H11: the retry sleep honours ctx cancellation so shutdown doesn't leave handler goroutines blocked on a long backoff. The DLQ event gets a fresh ID (downstream dedupers no longer see it as a duplicate of the original) and publish errors are logged rather than silently swallowed.

func Emit 

func Emit(pub Publisher, cfgs ...EmitConfig) maniflex.MiddlewareFunc

Emit returns a maniflex.MiddlewareFunc that builds an Event from ServerContext and publishes it to pub after the DB step succeeds. Register it with AtPosition(After) on the DB pipeline step: 

server.Pipeline.DB.Register(
    events.Emit(bus, events.EmitConfig{Source: cfg.ServiceName}),
    maniflex.ForOperation(maniflex.OpCreate, maniflex.OpUpdate, maniflex.OpDelete),
    maniflex.AtPosition(maniflex.After),
)

Transactional outbox: when pub implements TxPublisher and ctx.Tx is non-nil, the event INSERT commits atomically with the business write. Otherwise the publish runs fire-and-forget in a background goroutine.

func NewID added in v0.3.0 

func NewID() string

NewID returns a fresh event identifier in the format Publish assigns. Exported for subpackages that mint events of their own — the outbox relayer needs one when it dead-letters, so that the dead-letter is a new event downstream rather than a duplicate of the one that failed.

Types

type AckMode 

type AckMode int

AckMode controls how events are acknowledged. 

const (
	// AutoAck acknowledges automatically when the handler returns nil.
	// A non-nil return triggers retry (up to Subscription.MaxRetry times).
	AutoAck AckMode = iota
	// ManualAck is reserved for a future extension.
	ManualAck
)

type Bus 

type Bus interface {
	Publisher
	Subscribe(ctx context.Context, sub Subscription) (Cancel, error)
}

Bus extends Publisher with the consumer-side Subscribe method. Adapters that support both directions (inproc, Redis, NATS, Kafka) implement Bus.

type Cancel 

type Cancel func()

Cancel stops a subscription. Safe to call multiple times.

type DedupeReleaser added in v0.3.1 

type DedupeReleaser interface {
	// Release removes a previously recorded id so it can be processed again.
	// Releasing an id that is not recorded is not an error.
	Release(ctx context.Context, id string) error
}

DedupeReleaser is an optional interface a DedupeStore may implement to undo a claim that did not result in successful processing.

Seen is a check-and-record in one atomic step, which is what makes a store safe when two workers are handed the same event at once — but it means the ID is recorded on handler *entry*, before the outcome is known. Without Release, a handler that returns a transient error has already marked the event as processed, so DeliverWithRetry's next attempt is dropped as a duplicate and the delivery reports success: the event is never processed and never dead-lettered (audit EV-2).

Both bundled stores implement this. A store that does not gets a warning from Dedupe when it is wrapped, because there is no correct fallback: the claim cannot be undone, so a transient failure still loses the event.

type DedupeStore 

type DedupeStore interface {
	// Seen reports whether id was processed before and marks it as seen atomically.
	// Returns (true, nil) for duplicates. Returns (false, nil) for new events.
	Seen(ctx context.Context, id string) (bool, error)
}

DedupeStore records seen event IDs for idempotent delivery. Implementations ship in this package: InMemoryDedupeStore and SQLDedupeStore.

type EmailMessage 

type EmailMessage struct {
	To      string
	Subject string
	HTML    string
	Text    string
}

EmailMessage is the message passed to Mailer.Send.

type EmailTemplateFunc 

type EmailTemplateFunc func(ctx context.Context, e Event) *EmailMessage

EmailTemplateFunc builds an EmailMessage from an event. Return nil to skip sending (e.g. when template conditions are not met).

type EmitConfig 

type EmitConfig struct {
	// Source overrides ctx.ServiceName() when non-empty.
	Source string
	// Type derives the event Type field. Default: DefaultType.
	Type TypeFunc
	// Subject derives the event Subject field. Default: DefaultSubject.
	Subject SubjectFunc
	// TenantID extracts the tenant partition key from the context.
	// Default: reads ctx.Auth.Claims["tenant_id"] when present.
	TenantID func(ctx *maniflex.ServerContext) string
	// SchemaVer is stamped on every produced event. Default: 0 (unversioned).
	SchemaVer int
}

EmitConfig customises the Event fields produced by the Emit middleware. All fields are optional; unset fields fall back to sensible defaults.

type Event 

type Event struct {
	// CloudEvents 1.0 core attributes
	ID       string          `json:"id"`
	Source   string          `json:"source"`
	Type     string          `json:"type"`
	Subject  string          `json:"subject,omitempty"`
	Time     time.Time       `json:"time"`
	DataType string          `json:"datatype,omitempty"`
	Data     json.RawMessage `json:"data,omitempty"`

	// maniflex extension attributes
	Model     string             `json:"model,omitempty"`
	Operation maniflex.Operation `json:"operation,omitempty"`
	RecordID  string             `json:"recordid,omitempty"`
	ActorID   string             `json:"actorid,omitempty"`
	TenantID  string             `json:"tenantid,omitempty"`
	TraceID   string             `json:"traceid,omitempty"`
	SchemaVer int                `json:"schemaver,omitempty"`
	Headers   map[string]string  `json:"headers,omitempty"`
}

Event is a CloudEvents 1.0-aligned event envelope with maniflex extensions. CloudEvents 1.0 core attributes are first-class fields; maniflex-specific attributes ride as CE extension attributes (lowercase, no underscores per the CE spec).

type Handler 

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

Handler processes a single event. Return nil to ack; non-nil to nack and retry.

func Log 

func Log(logger *slog.Logger) Handler

Log returns a Handler that logs each event at INFO level. Useful for debugging and as a reference subscriber implementation. 

bus.Subscribe(ctx, events.Subscription{
    Patterns: []string{"*"},
    Handler:  events.Log(slog.Default()),
})

func SendEmail 

func SendEmail(mailer Mailer, tmpl EmailTemplateFunc) Handler

SendEmail returns a Handler that sends a transactional email for each event. Use it as a subscriber on a Bus: 

bus.Subscribe(ctx, events.Subscription{
    Patterns: []string{"user.created"},
    Handler: events.SendEmail(mailer, func(ctx context.Context, e events.Event) *events.EmailMessage {
        var record map[string]any
        _ = json.Unmarshal(e.Data, &record)
        email, _ := record["email"].(string)
        return &events.EmailMessage{
            To:      email,
            Subject: "Welcome!",
            HTML:    "<p>Thanks for signing up.</p>",
        }
    }),
})

func Webhook 

func Webhook(cfg WebhookConfig) Handler

Webhook returns a Handler that POSTs each received event as JSON to cfg.URL. Use it as a subscriber on a Bus: 

bus.Subscribe(ctx, events.Subscription{
    Patterns: []string{"invoice.*", "order.*"},
    Handler:  events.Webhook(events.WebhookConfig{
        URL:    "https://hooks.example.com/receive",
        Secret: "whsec_abc123",
    }),
})

type InMemoryDedupeStore 

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

InMemoryDedupeStore is a thread-safe in-memory dedupe store with bounded size and per-entry TTL. Entries are evicted when full (LRU-style, O(n) eviction).

This store is lost on process restart. Use SQLDedupeStore for persistence.

func NewInMemoryDedupeStore 

func NewInMemoryDedupeStore(maxSize int, ttl time.Duration) *InMemoryDedupeStore

NewInMemoryDedupeStore creates a store that holds up to maxSize IDs for ttl. Typical values: maxSize=10_000, ttl=24*time.Hour.

func (*InMemoryDedupeStore) Release added in v0.3.1 

func (s *InMemoryDedupeStore) Release(_ context.Context, id string) error

Release drops id, so a later delivery is treated as new. See DedupeReleaser.

func (*InMemoryDedupeStore) Seen 

func (s *InMemoryDedupeStore) Seen(_ context.Context, id string) (bool, error)

type Mailer 

type Mailer interface {
	Send(ctx context.Context, msg EmailMessage) error
}

Mailer sends a transactional email. Implement using any provider (SMTP, SendGrid, Resend, Postmark, etc.).

type Publisher 

type Publisher interface {
	Publish(ctx context.Context, e Event) error
	PublishBatch(ctx context.Context, es []Event) error
	Close() error
}

Publisher is the producer-only interface. Publish-only backends (SNS, transactional outbox) implement this without implementing Subscribe.

type ReadBackoff added in v0.3.1 

type ReadBackoff struct {
	// Min is the delay after the first failure. Zero means DefaultReadBackoffMin.
	Min time.Duration
	// Max caps the delay. Zero means DefaultReadBackoffMax.
	Max time.Duration
	// contains filtered or unexported fields
}

ReadBackoff paces an adapter's consume loop when reading from the broker fails. It grows the delay exponentially between Min and Max and applies jitter, so a fleet of consumers that all lost the broker at the same instant does not reconnect in lockstep.

A read loop retries forever by design — there is no attempt limit, because the alternative to retrying is a consumer that silently stops consuming. The backoff is what keeps "forever" affordable.

The zero value is usable and applies the defaults above. It is not safe for concurrent use; give each consume loop its own. 

var bo events.ReadBackoff
for {
    msg, err := read(ctx)
    if err != nil {
        n, d, escalate := bo.Next()
        // ...log at WARN, or at ERROR when escalate is true...
        if !bo.Wait(ctx, d) {
            return
        }
        continue
    }
    bo.Reset()
    handle(msg)
}

func (*ReadBackoff) Next added in v0.3.1 

func (b *ReadBackoff) Next() (attempt int, delay time.Duration, escalate bool)

Next advances to the next attempt and returns its number (1 for the first failure), the delay to wait, and whether this is the first attempt to reach Max. Callers use escalate to log once at ERROR when a run of failures stops looking transient, without turning every subsequent retry into an ERROR.

The delay is jittered within the upper half of the computed interval: full jitter would allow a near-zero wait, which is the one outcome a loop meant to stop hammering a struggling broker cannot afford.

func (*ReadBackoff) Reset added in v0.3.1 

func (b *ReadBackoff) Reset()

Reset returns the backoff to its initial state. Call it after every successful read, so an unrelated failure hours later starts from Min rather than inheriting a stale delay — and so a recovered broker is allowed to escalate again if it fails a second time.

func (*ReadBackoff) Wait added in v0.3.1 

func (b *ReadBackoff) Wait(ctx context.Context, d time.Duration) bool

Wait blocks for d, reporting false if ctx ended first. A read loop must treat false as "stop": shutdown was requested, and unlike time.Sleep this does not make the caller wait out the remaining delay before noticing.

type SQLDedupeStore 

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

SQLDedupeStore is a *sql.DB-backed dedupe store. Call Migrate to create the required table before use.

func NewSQLDedupeStore 

func NewSQLDedupeStore(db *sql.DB, driver string) *SQLDedupeStore

NewSQLDedupeStore creates a store backed by db. driver is "postgres" or "sqlite".

func (*SQLDedupeStore) Migrate 

func (s *SQLDedupeStore) Migrate(ctx context.Context) error

Migrate creates the event_dedupe table if it does not exist.

func (*SQLDedupeStore) Release added in v0.3.1 

func (s *SQLDedupeStore) Release(ctx context.Context, id string) error

Release deletes id's row, so a later delivery is treated as new. Deleting a row that is not there is not an error. See DedupeReleaser.

func (*SQLDedupeStore) Seen 

func (s *SQLDedupeStore) Seen(ctx context.Context, id string) (bool, error)

Seen atomically inserts id; returns true if a row already existed.

type SQLExecer 

type SQLExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

SQLExecer is satisfied by both *sql.DB and *sql.Tx. Used by TxPublisher.PublishWithExecer to INSERT outbox rows within an existing database transaction without holding a direct *sql.Tx reference.

type SubjectFunc 

type SubjectFunc func(ctx *maniflex.ServerContext) string

SubjectFunc derives the CloudEvents Subject from a Server pipeline context. 

var DefaultSubject SubjectFunc = func(ctx *maniflex.ServerContext) string {
	if ctx.ResourceID != "" {
		return lowerFirst(ctx.Model.Name) + "/" + ctx.ResourceID
	}
	return lowerFirst(ctx.Model.Name)
}

DefaultSubject is the default SubjectFunc: returns "{model}/{record_id}", e.g. "invoice/abc123". Falls back to "{model}" when no resource ID is set.

type Subscription 

type Subscription struct {
	// Group is the consumer-group name for Kafka, JetStream, and Redis XREADGROUP.
	// Ignored by in-process subscriptions.
	Group string

	// Patterns is a list of glob-style event Type patterns.
	// "invoice.*" matches "invoice.created", "invoice.updated", etc.
	// Translated to broker-native filters where supported.
	Patterns []string

	// Handler is called for each matching event.
	Handler Handler

	// Concurrency is the number of concurrent handler goroutines. Default: 1.
	Concurrency int

	// AckMode controls acknowledgement. Default: AutoAck.
	AckMode AckMode

	// MaxRetry is the number of retries before sending to the DLQ. Default: 3.
	MaxRetry int

	// Backoff returns the pre-retry delay. Default: linear (attempt × second).
	Backoff func(attempt int) time.Duration

	// DLQ is the event Type published after MaxRetry exhaustion.
	// Empty string disables dead-lettering.
	DLQ string
}

Subscription describes which events to consume and how to process them.

type TxPublisher 

type TxPublisher interface {
	Publisher
	PublishWithExecer(ctx context.Context, ex SQLExecer, e Event) error
}

TxPublisher is an optional interface implemented by outbox.Bus. When the active Publisher implements TxPublisher and a database transaction is in flight, Emit calls PublishWithExecer so the outbox INSERT commits atomically with the business write.

type TypeFunc 

type TypeFunc func(ctx *maniflex.ServerContext) string

TypeFunc derives the CloudEvents Type from a Server pipeline context. 

var DefaultType TypeFunc = func(ctx *maniflex.ServerContext) string {
	suffix := map[maniflex.Operation]string{
		maniflex.OpCreate: "created",
		maniflex.OpUpdate: "updated",
		maniflex.OpDelete: "deleted",
		maniflex.OpRead:   "read",
		maniflex.OpList:   "listed",
	}
	s, ok := suffix[ctx.Operation]
	if !ok {
		s = string(ctx.Operation)
	}
	return lowerFirst(ctx.Model.Name) + "." + s
}

DefaultType is the default TypeFunc: returns "{model}.{operation}" in lowercase, e.g. "invoice.created", "user.deleted".

type WebhookConfig 

type WebhookConfig struct {
	// URL to POST each event to.
	URL string
	// Secret signs the JSON body with HMAC-SHA256.
	// The signature is sent in X-Webhook-Signature: sha256=<hex>.
	// Leave empty to disable signing.
	Secret string
	// Timeout for each HTTP POST. Default: 5s.
	Timeout time.Duration
	// MaxRetries is the number of retry attempts on 5xx or network error. Default: 0.
	MaxRetries int
	// Headers are added to every webhook request.
	Headers map[string]string
}

WebhookConfig controls HTTP event delivery.

Source Files

View all Source files

Directories

Path Synopsis
Package cloudevents provides encoding and decoding helpers for CloudEvents 1.0.
 Package cloudevents provides encoding and decoding helpers for CloudEvents 1.0.
Package inproc provides an in-process event bus for maniflex.
 Package inproc provides an in-process event bus for maniflex.
kafka module
nats module
Package outbox provides a transactional outbox publisher for maniflex events.
 Package outbox provides a transactional outbox publisher for maniflex events.
rabbitmq module
redis module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.