events

events

import "github.com/agentstation/starmap/internal/server/events"

Package events provides a unified event system for real-time catalog updates.

This package implements a broker pattern that connects Starmap's hooks system to multiple transport mechanisms (WebSocket, SSE, etc.) through a common event pipeline. This eliminates code duplication and provides a single point for event distribution.

Index

• Variables
• func TrySend[T any](ch chan<- T, item T) error
• type BackpressurePolicy
• type Broker
  • func NewBroker(logger *zerolog.Logger) *Broker
  • func (b *Broker) DeliveryStats() DeliveryStats
  • func (b *Broker) EventsDropped() uint64
  • func (b *Broker) EventsPublished() uint64
  • func (b *Broker) Publish(eventType EventType, data any)
  • func (b *Broker) QueueDepth() int
  • func (b *Broker) Run(ctx context.Context)
  • func (b *Broker) Subscribe(sub Subscriber)
  • func (b *Broker) SubscriberCount() int
  • func (b *Broker) Unsubscribe(sub Subscriber)
• type DeliveryResult
• type DeliveryStats
• type DeliveryTarget
• type Event
• type EventType
• type Fanout
  • func NewFanout[T any](policy BackpressurePolicy, logger *zerolog.Logger) *Fanout[T]
  • func (f *Fanout[T]) Deliver(targets []DeliveryTarget[T], item T) DeliveryResult
  • func (f *Fanout[T]) Stats() DeliveryStats
• type Subscriber

Variables

ErrBackpressure reports that a delivery target cannot accept an event now.

var ErrBackpressure = errors.New("event delivery backpressure")

func TrySend

func TrySend[T any](ch chan<- T, item T) error

TrySend attempts a non-blocking channel send and reports backpressure if full.

type BackpressurePolicy

BackpressurePolicy controls what fan-out does when a target is full.

type BackpressurePolicy string

const (
    // BackpressureSkip drops the event for the slow target and keeps it connected.
    BackpressureSkip BackpressurePolicy = "skip"
    // BackpressureDisconnect drops the slow target from future fan-out.
    BackpressureDisconnect BackpressurePolicy = "disconnect"
)

type Broker

Broker manages event distribution to multiple subscribers. It provides a central hub for catalog events, fanning them out to all registered subscribers (WebSocket, SSE, etc.) concurrently.

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

func NewBroker

func NewBroker(logger *zerolog.Logger) *Broker

NewBroker creates a new event broker.

func (*Broker) DeliveryStats

func (b *Broker) DeliveryStats() DeliveryStats

DeliveryStats returns cumulative subscriber delivery counters.

func (*Broker) EventsDropped

func (b *Broker) EventsDropped() uint64

EventsDropped returns the total number of events dropped.

func (*Broker) EventsPublished

func (b *Broker) EventsPublished() uint64

EventsPublished returns the total number of events published.

func (*Broker) Publish

func (b *Broker) Publish(eventType EventType, data any)

Publish sends an event to all subscribers.

func (*Broker) QueueDepth

func (b *Broker) QueueDepth() int

QueueDepth returns the current number of events in the queue.

func (*Broker) Run

func (b *Broker) Run(ctx context.Context)

Run starts the broker's event loop. Should be called in a goroutine. The broker will run until the context is cancelled.

func (*Broker) Subscribe

func (b *Broker) Subscribe(sub Subscriber)

Subscribe registers a new subscriber to receive events.

func (*Broker) SubscriberCount

func (b *Broker) SubscriberCount() int

SubscriberCount returns the current number of subscribers.

func (*Broker) Unsubscribe

func (b *Broker) Unsubscribe(sub Subscriber)

Unsubscribe removes a subscriber from receiving events.

type DeliveryResult

DeliveryResult describes one fan-out attempt.

type DeliveryResult struct {
    Sent         int
    Skipped      int
    Disconnected int
    Failed       int
}

type DeliveryStats

DeliveryStats contains cumulative fan-out counters.

type DeliveryStats struct {
    Sent         uint64
    Skipped      uint64
    Disconnected uint64
    Failed       uint64
}

type DeliveryTarget

DeliveryTarget is one event fan-out destination.

type DeliveryTarget[T any] struct {
    ID    string
    Send  func(T) error
    Close func() error
}

type Event

Event represents a catalog event with type, timestamp, and data.

type Event struct {
    Type      EventType `json:"type"`
    Timestamp time.Time `json:"timestamp"`
    Data      any       `json:"data"`
}

type EventType

EventType represents the type of catalog event.

type EventType string

Event types for catalog changes.

const (
    // Model events (from Starmap hooks).
    ModelAdded   EventType = "model.added"
    ModelUpdated EventType = "model.updated"
    ModelDeleted EventType = "model.deleted"

    // Sync events (from sync operations).
    SyncStarted   EventType = "sync.started"
    SyncCompleted EventType = "sync.completed"
    // CatalogPublished is emitted once a durable generation becomes visible.
    CatalogPublished EventType = "catalog.published"

    // Client events (from transport layers).
    ClientConnected EventType = "client.connected"
)

type Fanout

Fanout delivers events to targets using one explicit backpressure policy.

type Fanout[T any] struct {
    // contains filtered or unexported fields
}

func NewFanout

func NewFanout[T any](policy BackpressurePolicy, logger *zerolog.Logger) *Fanout[T]

NewFanout creates a fan-out dispatcher.

func (*Fanout[T]) Deliver

func (f *Fanout[T]) Deliver(targets []DeliveryTarget[T], item T) DeliveryResult

Deliver sends an item to all targets without spawning per-target goroutines.

func (*Fanout[T]) Stats

func (f *Fanout[T]) Stats() DeliveryStats

Stats returns cumulative fan-out counters.

type Subscriber

Subscriber is an interface for event consumers. Implementations adapt the unified event stream to specific transport mechanisms (WebSocket, SSE, MQTT, webhooks, etc.).

type Subscriber interface {
    // Send delivers an event to the subscriber.
    // Implementations should be non-blocking and handle errors gracefully.
    Send(Event) error

    // Close cleanly shuts down the subscriber.
    Close() error
}

Generated by gomarkdoc