Documentation
¶
Overview ¶
Package eventbus is the generic RabbitMQ event system: the BaseEvent envelope, topic publisher/subscriber with signing, fanout broadcast, per-type dispatch with real outcome metrics, and DLQ support. It carries no domain event registry — applications define their own event-type constants and, if they want, thin constructors on top of NewEvent.
Index ¶
- Variables
- func GetEventSignatureInfo(signedEvent *SignedEvent) map[string]any
- func RecordEventAcknowledged(eventType, service, queue string)
- func RecordEventDLQ(eventType, service, queue, reason string)
- func RecordEventProcessed(eventType, service, queue, status string)
- func RecordEventPublished(eventType, service, status string)
- func RecordEventReceived(eventType, service, queue string)
- func RecordEventRequeued(eventType, service, queue string)
- func UpdateConnectionStatus(service, connectionType string, connected bool)
- func UpdateQueueDepth(queue, service string, depth float64)
- func UserID(ctx context.Context) (string, bool)
- func VerifyEvent(signedEvent *SignedEvent, config *SignatureConfig) error
- func WithBroadcast(ctx context.Context, event *BroadcastEvent) context.Context
- func WithUserID(ctx context.Context, userID string) context.Context
- type BaseEvent
- type BroadcastConfig
- type BroadcastEvent
- type BroadcastEventHandler
- type BroadcastPublisher
- type BroadcastSubscriber
- type Dispatcher
- type EventHandler
- type EventSigner
- type EventVerifier
- type HandlerFunc
- type Publisher
- type PublisherInterface
- type SecurityMetrics
- type SignatureConfig
- type SignedEvent
- type Subscriber
- type SubscriberInterface
Constants ¶
This section is empty.
Variables ¶
var ( // EventsPublished tracks total events published EventsPublished = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_published_total", Help: "Total number of events published to RabbitMQ", }, []string{"event_type", "service", "status"}, ) // EventsReceived tracks total events received EventsReceived = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_received_total", Help: "Total number of events received from RabbitMQ", }, []string{"event_type", "service", "queue"}, ) // EventsProcessed tracks total events processed EventsProcessed = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_processed_total", Help: "Total number of events successfully processed", }, []string{"event_type", "service", "queue", "status"}, ) // EventProcessingDuration tracks event processing time EventProcessingDuration = promauto.NewHistogramVec( prometheus.HistogramOpts{ Name: "rabbitmq_event_processing_duration_seconds", Help: "Duration of event processing in seconds", Buckets: []float64{0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, }, []string{"event_type", "service", "queue"}, ) // EventsInDLQ tracks events sent to dead letter queue EventsInDLQ = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_dlq_total", Help: "Total number of events sent to dead letter queue", }, []string{"event_type", "service", "queue", "reason"}, ) // EventsRequeued tracks events requeued for retry EventsRequeued = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_requeued_total", Help: "Total number of events requeued for retry", }, []string{"event_type", "service", "queue"}, ) // EventsAcknowledged tracks events acknowledged EventsAcknowledged = promauto.NewCounterVec( prometheus.CounterOpts{ Name: "rabbitmq_events_acknowledged_total", Help: "Total number of events acknowledged", }, []string{"event_type", "service", "queue"}, ) // QueueDepth tracks current queue depth (gauge) QueueDepth = promauto.NewGaugeVec( prometheus.GaugeOpts{ Name: "rabbitmq_queue_depth", Help: "Current depth of RabbitMQ queues", }, []string{"queue", "service"}, ) // ConnectionStatus tracks connection status ConnectionStatus = promauto.NewGaugeVec( prometheus.GaugeOpts{ Name: "rabbitmq_connection_status", Help: "RabbitMQ connection status (1 = connected, 0 = disconnected)", }, []string{"service", "connection_type"}, ) )
RabbitMQ event metrics
var BroadcastExchanges = struct { CertificateEvents string SecurityAlerts string SystemEvents string }{ CertificateEvents: "certificate.broadcast", SecurityAlerts: "security.broadcast", SystemEvents: "system.broadcast", }
BroadcastExchanges defines standard broadcast exchanges
var CertificateEventTypes = struct { CertificateRevoked string CertificateExpiring string CACompromised string EmergencyRenewal string CertificateProvisioned string }{ CertificateRevoked: "certificate.revoked", CertificateExpiring: "certificate.expiring", CACompromised: "ca.compromised", EmergencyRenewal: "certificate.emergency.renewal", CertificateProvisioned: "certificate.provisioned", }
CertificateEventTypes defines certificate-related broadcast event types
var DefaultExchange = "events"
DefaultExchange is the topic exchange publishers and subscribers declare and bind against when no other exchange is configured. Deployments with an established wire contract override it at startup (before any publisher or subscriber is created).
var GlobalSecurityMetrics = &SecurityMetrics{}
Global security metrics (can be exported to Prometheus)
Functions ¶
func GetEventSignatureInfo ¶
func GetEventSignatureInfo(signedEvent *SignedEvent) map[string]any
GetEventSignatureInfo extracts signature information for debugging/logging
func RecordEventAcknowledged ¶
func RecordEventAcknowledged(eventType, service, queue string)
RecordEventAcknowledged records an event acknowledgment
func RecordEventDLQ ¶
func RecordEventDLQ(eventType, service, queue, reason string)
RecordEventDLQ records an event sent to DLQ
func RecordEventProcessed ¶
func RecordEventProcessed(eventType, service, queue, status string)
RecordEventProcessed records an event processing result
func RecordEventPublished ¶
func RecordEventPublished(eventType, service, status string)
RecordEventPublished records an event publication
func RecordEventReceived ¶
func RecordEventReceived(eventType, service, queue string)
RecordEventReceived records an event reception
func RecordEventRequeued ¶
func RecordEventRequeued(eventType, service, queue string)
RecordEventRequeued records an event requeue
func UpdateConnectionStatus ¶
UpdateConnectionStatus updates the connection status
func UpdateQueueDepth ¶
UpdateQueueDepth updates the current queue depth
func UserID ¶
UserID returns the user a message was published on behalf of, if the publisher set one.
func VerifyEvent ¶
func VerifyEvent(signedEvent *SignedEvent, config *SignatureConfig) error
VerifyEvent validates the cryptographic signature of a signed event
func WithBroadcast ¶
func WithBroadcast(ctx context.Context, event *BroadcastEvent) context.Context
WithBroadcast returns a context carrying the broadcast envelope — the originating exchange, publishing instance and headers — that delivered the event.
Types ¶
type BaseEvent ¶
type BaseEvent struct {
ID string `json:"id"`
Type string `json:"type"`
Timestamp time.Time `json:"timestamp"`
Source string `json:"source"`
UserID string `json:"user_id,omitempty"`
Data map[string]any `json:"data"`
}
BaseEvent represents the common structure for all events
func (*BaseEvent) GetEventAction ¶
GetEventAction extracts the action from event type (e.g., "registered" from "user.registered")
func (*BaseEvent) GetEventDomain ¶
GetEventDomain extracts the domain from event type (e.g., "user" from "user.registered")
func (*BaseEvent) ValidateEvent ¶
ValidateEvent validates that the event has the required fields
type BroadcastConfig ¶
type BroadcastConfig struct {
// ExchangeName for broadcast messages (fanout exchange)
ExchangeName string
// ServiceName for the consuming service
ServiceName string
// InstanceID unique identifier for this service instance
InstanceID string
// TTL for broadcast messages (optional)
MessageTTL time.Duration
// MaxRetries for broadcast message processing
MaxRetries int
}
BroadcastConfig configures broadcast messaging
type BroadcastEvent ¶
type BroadcastEvent struct {
*BaseEvent
Exchange string `json:"exchange"`
InstanceID string `json:"instance_id"`
Headers map[string]string `json:"headers"`
}
BroadcastEvent represents a broadcast event with additional metadata
type BroadcastEventHandler ¶
type BroadcastEventHandler func(event *BroadcastEvent) error
BroadcastEventHandler handles broadcast events with context
type BroadcastPublisher ¶
type BroadcastPublisher struct {
*Publisher
}
BroadcastPublisher extends the regular publisher for broadcast messaging
func NewBroadcastPublisher ¶
func NewBroadcastPublisher(rabbitmqURL, serviceName string) (*BroadcastPublisher, error)
func (*BroadcastPublisher) PublishBroadcast ¶
func (bp *BroadcastPublisher) PublishBroadcast(exchangeName string, event *BaseEvent) error
PublishBroadcast publishes a message to all service instances via fanout exchange
func (*BroadcastPublisher) ServiceName ¶
func (bp *BroadcastPublisher) ServiceName() string
NewBroadcastPublisher creates a publisher for broadcast messages ServiceName returns the service name this publisher stamps on events.
type BroadcastSubscriber ¶
type BroadcastSubscriber struct {
*Subscriber
// contains filtered or unexported fields
}
BroadcastSubscriber extends the regular subscriber to support broadcast messaging where all service instances receive the same message
func NewBroadcastSubscriber ¶
func NewBroadcastSubscriber(rabbitmqURL string, config BroadcastConfig) (*BroadcastSubscriber, error)
NewBroadcastSubscriber creates a subscriber for broadcast messages
func (*BroadcastSubscriber) SubscribeBroadcast ¶
func (bs *BroadcastSubscriber) SubscribeBroadcast(exchangeName, routingKey string, handler EventHandler) error
SubscribeBroadcast subscribes to broadcast messages using fanout exchange Each service instance gets its own temporary queue
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher routes events to per-type handlers and records processing metrics with the real outcome. It replaces the hand-rolled switch-plus-deferred-metrics prologue that every service copied — which hardcoded "success" and therefore reported a 100% success rate to Prometheus even when handlers returned errors.
Dispatcher implements EventHandler, so it plugs directly into Subscriber.Subscribe:
d := events.NewDispatcher("skills-service", "skills-queue").
Register(events.EventUserRegistered, h.handleUserRegistered).
Register(events.EventSkillAdded, h.handleSkillAdded)
subscriber.Subscribe(events.EventUserRegistered, "skills-queue", d)
func NewDispatcher ¶
func NewDispatcher(service, queue string) *Dispatcher
NewDispatcher creates a dispatcher for the given service and queue; the names label the processing metrics.
func (*Dispatcher) Fallback ¶
func (d *Dispatcher) Fallback(fn HandlerFunc) *Dispatcher
Fallback sets the handler for event types with no Register entry. Without one, unregistered events are logged, recorded with status "unhandled" and acknowledged (nil error).
func (*Dispatcher) Handle ¶
func (d *Dispatcher) Handle(ctx context.Context, event *BaseEvent) error
Handle implements EventHandler. It routes the event, then records the processing duration and outcome ("success", "error" or "unhandled").
func (*Dispatcher) Register ¶
func (d *Dispatcher) Register(eventType string, fn HandlerFunc) *Dispatcher
Register maps an event type to its handler and returns the dispatcher for chaining.
type EventHandler ¶
EventHandler defines the interface for handling events
func WithMetrics ¶
func WithMetrics(service, queue string, handler EventHandler) EventHandler
WithMetrics wraps an existing EventHandler (typically a switch-based handler) so that processing duration and the real outcome are recorded, without restructuring the handler into per-type registrations. It is the minimal migration path away from the copy-pasted deferred-metrics prologue that always reported "success".
type EventSigner ¶
type EventSigner struct {
// contains filtered or unexported fields
}
EventSigner provides a convenient interface for signing events
func NewEventSigner ¶
func NewEventSigner(config *SignatureConfig) *EventSigner
NewEventSigner creates a new event signer with the given configuration
func (*EventSigner) Sign ¶
func (s *EventSigner) Sign(event *BaseEvent) (*SignedEvent, error)
Sign signs an event and returns a signed event
func (*EventSigner) Verify ¶
func (s *EventSigner) Verify(signedEvent *SignedEvent) error
Verify verifies a signed event
type EventVerifier ¶
type EventVerifier struct {
// contains filtered or unexported fields
}
EventVerifier provides a convenient interface for verifying events
func NewEventVerifier ¶
func NewEventVerifier(config *SignatureConfig) *EventVerifier
NewEventVerifier creates a new event verifier with the given configuration
func (*EventVerifier) Verify ¶
func (v *EventVerifier) Verify(signedEvent *SignedEvent) error
Verify verifies a signed event
type HandlerFunc ¶
HandlerFunc handles a single event type.
type Publisher ¶
type Publisher struct {
// contains filtered or unexported fields
}
Publisher handles publishing events to RabbitMQ
func NewPublisher ¶
NewPublisher creates a new event publisher
func NewPublisherWithService ¶
NewPublisherWithService creates a new event publisher with service name
func NewPublisherWithSigning ¶
func NewPublisherWithSigning(rabbitmqURL, serviceName string, signingConfig *SignatureConfig, requireSigning bool) (*Publisher, error)
NewPublisherWithSigning creates a new event publisher with optional message signing
func (*Publisher) PublishSigned ¶
PublishSigned publishes a cryptographically signed event
func (*Publisher) TestConnection ¶
TestConnection tests the RabbitMQ connection health
type PublisherInterface ¶
type PublisherInterface interface {
Publish(ctx context.Context, event *BaseEvent) error
TestConnection() error
Close() error
}
PublisherInterface defines the common interface for event publishers
type SecurityMetrics ¶
type SecurityMetrics struct {
SignedEventsPublished int64
UnsignedEventsPublished int64
VerificationSuccesses int64
VerificationFailures int64
ExpiredSignatures int64
InvalidSignatures int64
}
SecurityMetrics tracks security-related metrics for monitoring
type SignatureConfig ¶
type SignatureConfig struct {
SigningKey []byte
ExpiryWindow time.Duration // How long a signature is valid
RequiredClaims []string // Required fields in the event for signing
}
SignatureConfig holds configuration for event signing
func DefaultSignatureConfig ¶
func DefaultSignatureConfig() *SignatureConfig
DefaultSignatureConfig returns the default configuration for event signing
func LoadSignatureConfig ¶
func LoadSignatureConfig(store secrets.Store) (*SignatureConfig, error)
LoadSignatureConfig returns the standard event signature configuration with the signing key resolved, in order, from: the given secrets store (kms/event-signing-key), the EVENT_SIGNING_KEY environment variable, and the development fallback key. Pass a nil store to have one constructed from the environment (SECRETS_* configuration), which degrades to env-only resolution when no remote backend is configured.
This replaces the getEventSigningKey() helper that was copy-pasted into every service entrypoint.
func RotateSigningKey ¶
func RotateSigningKey(oldConfig *SignatureConfig, newKey []byte) *SignatureConfig
RotateSigningKey provides a mechanism for key rotation (for future implementation)
type SignedEvent ¶
type SignedEvent struct {
Event *BaseEvent `json:"event"`
Signature string `json:"signature"`
SignedAt time.Time `json:"signed_at"`
ExpiresAt time.Time `json:"expires_at"`
}
SignedEvent extends BaseEvent with cryptographic signature
func SignEvent ¶
func SignEvent(event *BaseEvent, config *SignatureConfig) (*SignedEvent, error)
SignEvent creates a cryptographically signed version of the event
type Subscriber ¶
type Subscriber struct {
// contains filtered or unexported fields
}
Subscriber handles subscribing to events from RabbitMQ
func NewSubscriber ¶
func NewSubscriber(rabbitmqURL string) (*Subscriber, error)
NewSubscriber creates a new event subscriber
func NewSubscriberWithService ¶
func NewSubscriberWithService(rabbitmqURL, serviceName string) (*Subscriber, error)
NewSubscriberWithService creates a new event subscriber with service name
func NewSubscriberWithSigning ¶
func NewSubscriberWithSigning(rabbitmqURL, serviceName string, signingConfig *SignatureConfig, requireSigning bool) (*Subscriber, error)
NewSubscriberWithSigning creates a new event subscriber with optional signature verification
func (*Subscriber) Close ¶
func (s *Subscriber) Close() error
Close closes the subscriber connection
func (*Subscriber) Subscribe ¶
func (s *Subscriber) Subscribe(queueName string, routingKey string, handler EventHandler) error
Subscribe subscribes to events with the given routing key
type SubscriberInterface ¶
type SubscriberInterface interface {
Subscribe(queueName string, routingKey string, handler EventHandler) error
Close() error
}
SubscriberInterface defines the common interface for event subscribers
Source Files