kami
45b5e16eff
Things arrive at Maven from eight directions — a relayed Android notification on POST /api/ambient, mail candidates from mavmaild, RSS items, changed pages from the crawler, zenmoney and wg reads from mavpoll, CalDAV events, presence probes, meeting transcripts and image descriptions. Each grew its own shape and its own log line, and nothing could answer "what came in today, from where". internal/event is that answer: a flat source-agnostic envelope (Source, Kind, EntityIDs, Title, Body, Priority, OccurredAt, Payload) plus a bounded in-memory journal. Both are pure — Publish and Normalize take `now` as a parameter, so no clock read sits on a path a replay would drive. Adopting it did not touch eight callers, because every intake path already converges on three ipc.CoreAPI methods: WriteFact, WriteNote and CaptureTask. cmd/mavend/intake.go decorates that ONE interface, so mavweb, mavcaldav, mavpoll, mavmaild and the in-core feed/crawl/capture/ vision workers publish envelopes without knowing events exist. The lone exception is cmd/mavend/mail.go, which captures through the store directly and now publishes explicitly. Nothing dispatches on an event. It is a report that something arrived, never an instruction to speak — "a feed item appeared" becoming a notification is the nag this repo refuses. Digestion may read the journal later; it will still go through internal/loop's rules and the severity/presence routing table. Read surface: ipc.MethodRecentEvents (AuthRead, daemon-cached like TickTrace — a bare store cannot serve a ring) and a read-only /events page in mavweb. Production is unchanged when nobody is watching: a nil *event.Bus makes Publish a no-op and newIntakeAPI returns the wrapped API untouched, so config.intake_journal < 0 leaves no decorator on the call path at all. The default is 512 entries; the "off unless configured" rule is for capabilities that reach out, and a bounded in-memory log of writes core already performed reaches nowhere. Verified: make build, make test (go test -race) both clean. New tests cover the envelope and ring (internal/event, 95.7%), the decorator's invariants — a failed write publishes nothing, a deduped capture publishes nothing, OccurredAt is the fact's Ts and not notice time — and the /events page including escaping of feed-supplied titles.
130 lines
3.6 KiB
Go
130 lines
3.6 KiB
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, newest first. limit <= 0 returns
|
|
// everything held. Safe on a nil receiver (returns nil).
|
|
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
|
|
}
|