145 lines
3.8 KiB
Go
145 lines
3.8 KiB
Go
package events
|
|
|
|
import (
|
|
"log/slog"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// Store retains bounded inspection events; the transport selects safe or opt-in raw previews.
|
|
type Notice struct {
|
|
Event *Event `json:"event,omitempty"`
|
|
Gap uint64 `json:"gap,omitempty"`
|
|
ResetID uint64 `json:"reset_id,omitempty"`
|
|
}
|
|
type Subscriber struct {
|
|
channel chan Notice
|
|
lost uint64
|
|
}
|
|
type Store struct {
|
|
mu sync.Mutex
|
|
history []Event
|
|
capacity int
|
|
next uint64
|
|
counts map[Kind]uint64
|
|
log *slog.Logger
|
|
nextSub uint64
|
|
subs map[uint64]*Subscriber
|
|
}
|
|
|
|
func NewStore(capacity int, log *slog.Logger) *Store {
|
|
return &Store{capacity: capacity, counts: make(map[Kind]uint64), log: log, subs: make(map[uint64]*Subscriber)}
|
|
}
|
|
func (s *Store) Add(e Event) {
|
|
s.mu.Lock()
|
|
s.next++
|
|
e.ID = s.next
|
|
e.ObservedAt = time.Now().UTC()
|
|
s.counts[e.Kind]++
|
|
if len(s.history) == s.capacity {
|
|
copy(s.history, s.history[1:])
|
|
s.history = s.history[:len(s.history)-1]
|
|
}
|
|
s.history = append(s.history, cloneEvent(e))
|
|
for _, sub := range s.subs {
|
|
if sub.lost > 0 {
|
|
select {
|
|
// Advance the cursor through the current event with the gap notice.
|
|
// Leaving the cursor before it would cause a queue-of-one viewer to
|
|
// receive a gap forever, since every notice would fill its queue.
|
|
case sub.channel <- Notice{Gap: sub.lost + 1, ResetID: e.ID}:
|
|
sub.lost = 0
|
|
continue
|
|
default:
|
|
sub.lost++
|
|
continue
|
|
}
|
|
}
|
|
copyEvent := cloneEvent(e)
|
|
select {
|
|
case sub.channel <- Notice{Event: ©Event}:
|
|
default:
|
|
sub.lost++
|
|
}
|
|
}
|
|
s.mu.Unlock()
|
|
s.log.Info(string(e.Kind), "event", e)
|
|
}
|
|
func (s *Store) Snapshot() ([]Event, map[Kind]uint64) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
history := make([]Event, len(s.history))
|
|
for i, event := range s.history {
|
|
history[i] = cloneEvent(event)
|
|
}
|
|
counts := make(map[Kind]uint64, len(s.counts))
|
|
for k, v := range s.counts {
|
|
counts[k] = v
|
|
}
|
|
return history, counts
|
|
}
|
|
|
|
// Clone pointer fields so snapshot consumers cannot mutate retained history.
|
|
func cloneEvent(e Event) Event {
|
|
if e.Scenario != nil {
|
|
result := *e.Scenario
|
|
e.Scenario = &result
|
|
}
|
|
e.Properties.CorrelationData = append([]byte(nil), e.Properties.CorrelationData...)
|
|
if e.Properties.MessageExpirySeconds != nil {
|
|
n := *e.Properties.MessageExpirySeconds
|
|
e.Properties.MessageExpirySeconds = &n
|
|
}
|
|
if e.Duplicate != nil {
|
|
n := *e.Duplicate
|
|
e.Duplicate = &n
|
|
}
|
|
if e.LatencyMS != nil {
|
|
n := *e.LatencyMS
|
|
e.LatencyMS = &n
|
|
}
|
|
return e
|
|
}
|
|
|
|
// Subscribe atomically captures bounded replay and registers a bounded live queue.
|
|
// A cursor outside retained history creates an explicit reset to the oldest event.
|
|
func (s *Store) Subscribe(after uint64, queue, maxReplay int) (replay []Event, initialGap *Notice, stream <-chan Notice, unsubscribe func()) {
|
|
s.mu.Lock()
|
|
s.nextSub++
|
|
id := s.nextSub
|
|
sub := &Subscriber{channel: make(chan Notice, queue)}
|
|
s.subs[id] = sub
|
|
if len(s.history) > 0 {
|
|
oldest := s.history[0].ID
|
|
newest := s.history[len(s.history)-1].ID
|
|
if after > newest {
|
|
initialGap = &Notice{Gap: after - newest, ResetID: newest}
|
|
after = newest
|
|
} else if after < oldest-1 {
|
|
initialGap = &Notice{Gap: oldest - 1 - after, ResetID: oldest - 1}
|
|
after = oldest - 1
|
|
}
|
|
if maxReplay > 0 && newest-after > uint64(maxReplay) {
|
|
reset := newest - uint64(maxReplay)
|
|
missed := reset - after
|
|
if initialGap == nil {
|
|
initialGap = &Notice{Gap: missed, ResetID: reset}
|
|
} else {
|
|
initialGap.Gap += missed
|
|
initialGap.ResetID = reset
|
|
}
|
|
after = reset
|
|
}
|
|
for _, event := range s.history {
|
|
if event.ID > after {
|
|
replay = append(replay, cloneEvent(event))
|
|
}
|
|
}
|
|
} else if after > 0 {
|
|
initialGap = &Notice{Gap: after, ResetID: 0}
|
|
}
|
|
s.mu.Unlock()
|
|
var once sync.Once
|
|
return replay, initialGap, sub.channel, func() { once.Do(func() { s.mu.Lock(); delete(s.subs, id); close(sub.channel); s.mu.Unlock() }) }
|
|
}
|