Maven/internal/event/bus.go

package event

import (
	"sync"
	"time"
)

// Bus — the in-memory intake journal: a bounded ring of recent Events plus
// zero or more subscribers.
//
// Two properties are load-bearing, both about not changing production
// behaviour when nobody is watching:
//
//   - A nil *Bus is a working no-op. Publish on nil returns immediately, so
//     an intake path can call b.Publish(...) unconditionally and a daemon that
//     never built a bus behaves exactly as it did before. This is what let
//     eight callers adopt the envelope without a config flag each.
//   - Publish never blocks on a subscriber and never propagates a panic from
//     one. Intake is on the request path of POST /api/ambient and of every
//     fact write; a slow or broken observer must not be able to stall or kill
//     a write that already succeeded.
//
// The ring is bounded because it is memory that nothing prunes otherwise. Its
// contents are a window, not a record: the durable consequence of an event is
// the fact, note or task the intake path wrote.
type Bus struct {
	mu   sync.Mutex
	ring []Event // len == cap once full; oldest at (next % cap)
	next int
	n    int
	subs []func(Event)
}

// DefaultCapacity — how many recent events a bus keeps. A busy day is a few
// hundred intake events (a feed poll is one per new item), so this is roughly
// "today and yesterday" at a few hundred KB.
const DefaultCapacity = 512

// NewBus returns a bus keeping the last capacity events. capacity <= 0 uses
// DefaultCapacity.
func NewBus(capacity int) *Bus {
	if capacity <= 0 {
		capacity = DefaultCapacity
	}
	return &Bus{ring: make([]Event, capacity)}
}

// Publish normalizes e, drops it if it is not Valid, appends it to the ring and
// hands it to every subscriber. Safe on a nil receiver and safe from any
// goroutine.
//
// now is passed in rather than read from the clock: the whole point of #284's
// replay is that no time.Now() sits inside a path a scenario drives.
func (b *Bus) Publish(e Event, now time.Time) {
	if b == nil {
		return
	}
	e = e.Normalize(now)
	if !e.Valid() {
		return
	}

	b.mu.Lock()
	b.ring[b.next] = e
	b.next = (b.next + 1) % len(b.ring)
	if b.n < len(b.ring) {
		b.n++
	}
	subs := make([]func(Event), len(b.subs))
	copy(subs, b.subs)
	b.mu.Unlock()

	for _, fn := range subs {
		notify(fn, e)
	}
}

// notify calls one subscriber, swallowing a panic. A test double or a page
// renderer must not be able to take down a daemon from the intake path.
func notify(fn func(Event), e Event) {
	defer func() { _ = recover() }()
	fn(e)
}

// Subscribe registers fn to be called for every subsequent event, in publish
// order. There is no unsubscribe: subscribers are wired at startup and live as
// long as the daemon. Safe on a nil receiver (the subscription is dropped,
// which is the honest outcome when there is no bus to subscribe to).
func (b *Bus) Subscribe(fn func(Event)) {
	if b == nil || fn == nil {
		return
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	b.subs = append(b.subs, fn)
}

// Recent returns up to limit events, most recently NOTICED first. limit <= 0
// returns everything held. Safe on a nil receiver (returns nil).
//
// The order is the ring's insertion order, which is NoticedAt order, and it is
// deliberately not OccurredAt order: a cold feed read publishes a week of items
// in feed order, so sorting by when things happened would put a six-day-old
// item above one that arrived before it. Readers must label the column
// accordingly.
func (b *Bus) Recent(limit int) []Event {
	if b == nil {
		return nil
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if b.n == 0 {
		return nil
	}
	if limit <= 0 || limit > b.n {
		limit = b.n
	}
	out := make([]Event, 0, limit)
	// next points one past the newest; walk backwards.
	for i := 0; i < limit; i++ {
		idx := (b.next - 1 - i + len(b.ring)*2) % len(b.ring)
		out = append(out, b.ring[idx])
	}
	return out
}

// Len reports how many events the ring currently holds. Safe on nil.
func (b *Bus) Len() int {
	if b == nil {
		return 0
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	return b.n
}