event

package
v0.7.2 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: MIT Imports: 6 Imported by: 0

Documentation

Overview

Package event provides typed events, publisher/coordinator contracts and an in-process bus with enrollment, actor and timestamp metadata.

Design

Subscribers can observe lifecycle and command outcomes without being coupled to the service implementation. The bus supports synchronous and asynchronous dispatch and configurable error reporting. WithAsync uses eight workers, a 1,024-event pending queue and a 30-second lifetime from acceptance. NewAsync accepts explicit limits. Publish rejects excess events with ErrQueueFull; rejection is counted and reported even if the caller ignores the error. Stats exposes backlog and cumulative outcomes without event contents.

Accepted events retain context values and survive request cancellation, but expire on the bus-owned deadline. Subscribers run in registration order for each event; concurrent events have no delivery-order guarantee. Handlers and error callbacks must return promptly and handlers must honor cancellation. A non-cooperating handler occupies a worker; it does not trigger another one.

Always Close an asynchronous bus. Close stops acceptance and drains within its context deadline, then cancels active contexts and abandons queued events. Subsequent Close calls may wait for handlers that have not returned. The bus has no durable storage or replay; its subscribers can miss events after overload, expiry, sink failure or abrupt termination. Queue limits bound event counts rather than payload byte sizes; producers must bound payloads.

Publisher and Coordinator allow persistent implementations without importing server code. Run coordinates participating local mutations and event capture; the reference server supplies server/eventstore for SQL-backed applications. Durable audit/webhook delivery then runs independently of this bus, while after-commit bus notifications and slog remain ephemeral.

Events may contain sensitive protocol data. External sinks in server/eventsink apply an explicit projection, and server/audit persists that projection when configured. Direct subscribers must apply their own disclosure policy. The DDM notifier consumes persistent change rows rather than relying on bus delivery.

References

Index

Constants

View Source 
const (
	DefaultWorkers         = 8
	DefaultQueueCapacity   = 1024
	DefaultDeliveryTimeout = 30 * time.Second
)

Default asynchronous dispatch limits bound events, not payload byte sizes.

Variables

View Source 
var ErrAsyncConfig = errors.New("event: invalid async configuration")

ErrAsyncConfig identifies invalid asynchronous dispatch limits.

View Source 
var ErrCapture = errors.New("event: required capture failed")

ErrCapture indicates required recording failed; the associated local mutation must roll back and the caller may retry once recording is available.

View Source 
var ErrClosed = errors.New("event: bus closed")

ErrClosed is returned by Publish after Close.

View Source 
var ErrQueueFull = errors.New("event: queue full")

ErrQueueFull means the event was not accepted. No automatic replay occurs.

Functions

func Run added in v0.7.1 

func Run(ctx context.Context, p Publisher, fn func(context.Context) error) error

Run joins a publisher's transaction when it implements Coordinator. A plain in-process publisher executes fn directly and provides no persistence promise.

Types

type AsyncConfig 

type AsyncConfig struct {
	Workers         int
	QueueCapacity   int
	DeliveryTimeout time.Duration
}

AsyncConfig bounds active deliveries, pending events and time since acceptance. Zero fields select defaults; negative fields are invalid.

type Bus 

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

Bus dispatches events to subscribers. It is safe for concurrent use.

func New 

func New(opts ...Option) *Bus

New creates a bus.

func NewAsync 

func NewAsync(cfg AsyncConfig, opts ...Option) (*Bus, error)

NewAsync creates a bounded asynchronous bus. Explicit limits take precedence over WithAsync. Error reporting options work as with New.

func (*Bus) Close 

func (b *Bus) Close(ctx context.Context) error

Close stops accepting events and waits for asynchronous deliveries, or until ctx is done. On timeout it cancels active delivery contexts and discards queued events. Handlers must honor cancellation: a stuck handler retains its worker, and subsequent Close calls can wait for it, without creating workers.

func (*Bus) Publish 

func (b *Bus) Publish(ctx context.Context, e Event) error

Publish delivers e to subscribers of e.Type and of All. In synchronous mode it returns the joined handler errors; in asynchronous mode it returns ErrQueueFull when saturated and otherwise accepts without waiting for delivery. Async delivery retains context values, including from cancelled requests, but has its own deadline starting at acceptance. Callers must not mutate referenced event data until delivery finishes. Close drains accepted events within its deadline. Queue rejection also reaches the error handler.

func (*Bus) Stats 

func (b *Bus) Stats() Stats

Stats reports bounded dispatch state without exposing event payloads.

func (*Bus) Subscribe 

func (b *Bus) Subscribe(t Type, h Handler) func()

Subscribe registers h for events of type t (or All). The returned function removes the subscription.

type Coordinator added in v0.7.1 

type Coordinator interface {
	Run(context.Context, func(context.Context) error) error
}

Coordinator groups local state mutations and required event capture. Nested calls join the caller's transaction. Remote requests belong outside Run.

type Event 

type Event struct {
	// ID identifies this occurrence, independently of the enrollment identifier.
	// A persistent publisher assigns it before capture when empty.
	ID         string
	Type       Type
	At         time.Time
	Enrollment mdm.EnrollmentID
	// Actor is who caused the event: "device", "admin", or a system component.
	Actor string
	// Data carries type-specific detail, for example *mdm.Response for CommandResult.
	Data any
}

Event is one occurrence.

type Handler 

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

Handler receives events. Returning an error is reported through the bus error handler but does not stop other handlers.

type Option 

type Option func(*Bus)

Option configures a Bus.

func WithAsync 

func WithAsync() Option

WithAsync selects bounded asynchronous delivery with the default limits. Use NewAsync to configure those limits. The default is synchronous delivery in subscription order.

func WithErrorHandler 

func WithErrorHandler(f func(Event, error)) Option

WithErrorHandler receives handler errors, delivery expiry and queue rejection. It runs outside bus locks and must be concurrency-safe and return promptly. The default drops reports; Stats still counts asynchronous outcomes.

type Publisher added in v0.7.1 

type Publisher interface {
	Publish(context.Context, Event) error
}

Publisher accepts a typed occurrence. Persistent publishers must return a capture error before the associated operation reports success.

type Stats 

type Stats struct {
	Async           bool
	Closed          bool
	Workers         int
	QueueCapacity   int
	DeliveryTimeout time.Duration
	Queued          int
	InFlight        int
	Accepted        uint64
	Delivered       uint64
	Failed          uint64
	TimedOut        uint64
	Rejected        uint64
	Abandoned       uint64
}

Stats is a consistent snapshot. Counters are cumulative for this bus lifetime. Failed includes TimedOut; Abandoned counts queued events discarded at shutdown. Accepted = Delivered + Failed + Abandoned + Queued + InFlight. Synchronous buses report Async=false and zero dispatch counters.

type Type 

type Type string

Type names an event. 

const (
	// Security rejections carry metadata only; never attach credentials or remote errors.
	EnrollmentDenied          Type = "enrollment-denied"
	IdentityRejected          Type = "identity-rejected"
	CertificateStatusRejected Type = "certificate-status-rejected"
	PrivateHopRejected        Type = "private-hop-rejected"
	Enrolled                  Type = "enrolled"         // Authenticate accepted for a new enrollment
	Reenrolled                Type = "reenrolled"       // Authenticate accepted for an existing enrollment
	TokenUpdated              Type = "token-updated"    // TokenUpdate stored
	CheckedOut                Type = "checked-out"      // CheckOut received
	CertRotated               Type = "cert-rotated"     // enrollment identity certificate changed
	CommandQueued             Type = "command-queued"   // command enqueued for an enrollment
	CommandSent               Type = "command-sent"     // command delivered to the device
	CommandRejected           Type = "command-rejected" // server cleared a queued command that is no longer eligible
	CommandResult             Type = "command-result"   // Acknowledged, Error, CommandFormatError, or NotNow
	BootstrapTokenSet         Type = "bootstrap-token-set"
	// PushTokenInvalid is a token APNs says will never work again (410).
	// The enrollment is gone until it re-registers.
	PushTokenInvalid Type = "push-token-invalid"
	// PushRejected is a push APNs refused for a reason that is not the
	// device's: a wrong topic, a mismatched or expired push certificate, the
	// wrong environment, or a malformed request. It is the event to alert
	// on, because the cause is usually shared by every device on the topic
	// and no retry will clear it.
	PushRejected       Type = "push-rejected"
	DDMChanged         Type = "ddm-changed"
	DDMStatusReceived  Type = "ddm-status-received"
	CertReuseDenied    Type = "cert-reuse-denied"   // Authenticate presented a certificate another enrollment pinned before
	EnrollmentImported Type = "enrollment-imported" // record written by MigrationStore.Import
	UserAuthenticated  Type = "user-authenticated"  // UserAuthenticate digest accepted, AuthToken issued
	UserAuthFailed     Type = "user-auth-failed"    // UserAuthenticate digest rejected or challenge expired

	// ACMEChallengeValid is a device-attest-01 challenge that passed
	// verification and policy.
	ACMEChallengeValid Type = "acme-challenge-valid"
	// ACMEIssued is a device identity certificate issued through ACME.
	ACMEIssued Type = "acme-issued"
	// CertificateRevoked is an irreversible issuer-registry status change.
	CertificateRevoked Type = "certificate-revoked"
	// AttestationRejected is an attestation that failed verification, named
	// the wrong device, or was refused by policy. It is the event to alert
	// on: a device that fails here is either faulty or not what it claims.
	AttestationRejected Type = "attestation-rejected"

	// AdminAction is a mutating admin request that was allowed. Actor is the
	// principal name and Data names the action, method, path, and the
	// credential that acted, never the token and never the body.
	AdminAction Type = "admin-action"
	// AdminDenied records an administrative request refused by authorization.
	AdminDenied Type = "admin-denied"

	// All subscribes to every type.
	All Type = "*"
)

Event types published by the service layer.

Source Files

View all Source files

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.