event

Documentation

Overview

Package event implements the pub/sub bus and hook dispatch.

Index

• Constants
• Variables
• type Bus
  • func NewBus(cfg BusConfig, opts ...BusOption) *Bus
  • func (b *Bus) Close() error
  • func (b *Bus) IsPreHookEvent(_ string) bool
  • func (b *Bus) Publish(ctx context.Context, event domain.SystemEvent) error
  • func (b *Bus) RegisterPostHook(config HookConfig, hook PostHook)
  • func (b *Bus) RegisterPreHook(config HookConfig, hook PreHook)
  • func (b *Bus) RegisterPreHookType(_ string)
  • func (b *Bus) Subscribe(id string, topics ...string) (*Subscriber, error)
  • func (b *Bus) SubscriberCount() int
  • func (b *Bus) Unsubscribe(id string)
• type BusConfig
• type BusOption
  • func WithDropHandler(h DropHandler) BusOption
• type DropHandler
• type HookConfig
• type PostHook
• type PreHook
• type Publisher
• type Subscriber

Constants

const DefaultHookTimeout = 30 * time.Second

DefaultHookTimeout is the default timeout for hook execution.

Variables

var ErrBusClosed = error(busClosed{})

ErrBusClosed is returned when attempting to subscribe to a closed bus.

var ErrRetryLater = error(retryLater{})

ErrRetryLater is returned when dispatch fails due to slow subscribers. The publisher should retry publishing the event after a short delay.

Types

type Bus

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

Bus implements a channel-based pub/sub system with topic routing, synchronous pre-hooks, and asynchronous post-hooks.

Transaction boundary: the bus persists events outside the caller's transaction. This is intentional — if event persistence were wrapped in the caller's transaction and that transaction rolled back, we would have already dispatched the event (triggering side effects like adapter calls, worktree operations, etc.) but lost the record of it. The bus is the source of truth for what was dispatched; callers must treat bus.Publish() as idempotent if they need at-least-once delivery guarantees.

For synchronous validation that can abort an operation before it proceeds, use the worktree.HookRegistry instead.

func NewBus

func NewBus(cfg BusConfig, opts ...BusOption) *Bus

NewBus creates a new event bus with the given options.

func (*Bus) Close

func (b *Bus) Close() error

Close shuts down the bus, closing all subscriber channels.

func (*Bus) IsPreHookEvent

func (b *Bus) IsPreHookEvent(_ string) bool

IsPreHookEvent reports whether the given event type is a pre-hook event.

Deprecated: pre-hook event types were removed in the async event-bus migration. This method always returns false. Hooks now run after persistence and cannot abort database writes — use worktree.HookRegistry for synchronous validation.

func (*Bus) Publish

func (b *Bus) Publish(ctx context.Context, event domain.SystemEvent) error

Publish persists the event and dispatches it to matching subscribers.

• Persist event to repository
• Run pre-hooks synchronously (can abort dispatch but not persistence)
• Dispatch to matching subscribers
• Run post-hooks asynchronously

func (*Bus) RegisterPostHook

func (b *Bus) RegisterPostHook(config HookConfig, hook PostHook)

RegisterPostHook registers an asynchronous post-hook. Post-hooks are called after event dispatch with the configured timeout.

func (*Bus) RegisterPreHook

func (b *Bus) RegisterPreHook(config HookConfig, hook PreHook)

RegisterPreHook registers a synchronous pre-hook. Pre-hooks are called in registration order, after persistence but before dispatch.

For pre-creation validation (e.g., blocking worktree creation), use worktree.HookRegistry instead — it can abort the operation before persistence.

If a pre-hook returns an error, dispatch is aborted but the event remains persisted (it already happened).

Note: When a pre-hook times out, the goroutine running the hook continues executing if the hook function does not respect context cancellation. Go cannot forcefully kill goroutines. Hook implementations should check ctx.Done() and return promptly to avoid goroutine leaks.

func (*Bus) RegisterPreHookType

func (b *Bus) RegisterPreHookType(_ string)

RegisterPreHookType is a no-op.

Deprecated: pre-hook event types were removed in the async event-bus migration. This method does nothing. For synchronous validation that runs before state transitions, use worktree.HookRegistry instead.

func (*Bus) Subscribe

func (b *Bus) Subscribe(id string, topics ...string) (*Subscriber, error)

Subscribe creates a new subscriber for the given topics. If no topics are specified, the subscriber receives all events. Returns ErrBusClosed if the bus has been closed.

func (*Bus) SubscriberCount

func (b *Bus) SubscriberCount() int

SubscriberCount returns the number of active subscribers.

func (*Bus) Unsubscribe

func (b *Bus) Unsubscribe(id string)

Unsubscribe removes a subscriber.

type BusConfig

type BusConfig struct {
	EventRepo repository.EventRepository
}

BusConfig configures the event bus.

type BusOption

type BusOption func(*Bus)

BusOption configures the event bus.

func WithDropHandler

func WithDropHandler(h DropHandler) BusOption

WithDropHandler returns a BusOption that sets the drop handler. When set, dropped events call the handler instead of returning ErrRetryLater.

type DropHandler

type DropHandler func(subscriberID string, event domain.SystemEvent)

DropHandler is called when an event is dropped due to a slow subscriber. The handler typically logs a warning or enqueues a toast notification.

type HookConfig

type HookConfig struct {
	Name    string
	Timeout time.Duration // 0 means DefaultHookTimeout
}

HookConfig configures a hook's behavior.

type PostHook

type PostHook func(ctx context.Context, event domain.SystemEvent) error

PostHook is an asynchronous hook called after event dispatch. Errors are logged but do not affect event delivery.

type PreHook

type PreHook func(ctx context.Context, event domain.SystemEvent) error

PreHook is a synchronous hook called before event dispatch. If it returns an error, the event is aborted and not dispatched to subscribers.

type Publisher

type Publisher interface {
	Publish(ctx context.Context, event domain.SystemEvent) error
}

Publisher is the interface for publishing events.

type Subscriber

type Subscriber struct {
	ID     string
	Topics map[string]bool // event types to subscribe to (empty = all)
	C      chan domain.SystemEvent
}

Subscriber receives events matching their subscribed topics.