package events
import (
"encoding/json"
"errors"
"fmt"
"sync"
"time"
"github.com/nats-io/nats.go"
)
// VocabEvent is one forge event read back off the vocabulary stream,
// as Raise published it, with where it sits in the stream.
type VocabEvent struct {
// Seq is the event's position in the vocabulary stream, which every
// repository shares: a listener resumes after it.
Seq uint64
Kind string
Repo string
Payload map[string]any
At time.Time
// Gap marks the signal, not an event: the resume point asked for
// (From) is older than anything the window still holds, and the
// replay that follows starts at Seq.
Gap bool
From uint64
}
// vocabFilter is the subject a tail reads: every repository's events,
// or exactly one repository's. Exactly a noun and an action after the
// name, because repository names may contain dots and "quarry.evt.a.>"
// is also a.b's.
func vocabFilter(repo string) string {
if repo == "" {
return vocabSubject
}
return "quarry.evt." + repo + ".*.*"
}
// TailVocab delivers forge events from seq from+1 onward (from == 0
// means new events only, starting now) until cancel: every repository's
// when repo is "", else that one's. It is an ordered consumer, so it
// never drops under backpressure, for the same reason TailAttention
// does: a reconnecting listener's whole point is "everything after N".
//
// When from is older than anything the window still holds, the first
// delivery is a Gap signal and the replay follows it, because JetStream
// would otherwise clamp the start silently and the resume would look
// complete when it is not.
//
// cancel closes the returned channel once the consumer is stopped.
func (s *Server) TailVocab(repo string, from uint64) (<-chan VocabEvent, func()) {
out := make(chan VocabEvent) // unbuffered: never drops, backpressure blocks the pump
if s == nil {
close(out)
return out, func() {}
}
subj := vocabFilter(repo)
done := make(chan struct{})
var gap *VocabEvent
if from > 0 {
if first, err := s.vocabFirst(subj); err == nil && first > 0 && from+1 < first {
gap = &VocabEvent{Gap: true, From: from, Seq: first}
}
}
raw := make(chan VocabEvent)
opts := []nats.SubOpt{nats.OrderedConsumer(), nats.BindStream(VocabStream)}
if from > 0 {
opts = append(opts, nats.StartSequence(from+1))
} else {
opts = append(opts, nats.DeliverNew())
}
sub, err := s.js.Subscribe(subj, func(msg *nats.Msg) {
var r struct {
Kind string `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
if json.Unmarshal(msg.Data, &r) != nil || r.Kind == "" {
return
}
e := VocabEvent{Kind: r.Kind, Repo: r.Repo, Payload: r.Payload}
if meta, merr := msg.Metadata(); merr == nil {
e.Seq = meta.Sequence.Stream
e.At = meta.Timestamp
}
select {
case raw <- e:
case <-done:
}
}, opts...)
if err != nil {
close(out)
return out, func() {}
}
go func() {
defer close(out)
if gap != nil {
select {
case out <- *gap:
case <-done:
return
}
}
for {
select {
case <-done:
return
case e := <-raw:
select {
case out <- e:
case <-done:
return
}
}
}
}()
var once sync.Once
return out, func() {
once.Do(func() {
sub.Unsubscribe()
close(done)
})
}
}
// vocabFirst is the first sequence subj holds in the window, 0 when it
// holds none. Read through a subject-filtered view: the stream's own
// first sequence belongs to every repository at once.
func (s *Server) vocabFirst(subj string) (uint64, error) {
sub, err := s.js.PullSubscribe(subj, "",
nats.AckNone(), nats.BindStream(VocabStream),
nats.InactiveThreshold(30*time.Second))
if err != nil {
return 0, fmt.Errorf("events: vocabulary bounds: %w", err)
}
defer sub.Unsubscribe()
msgs, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
if err != nil && !errors.Is(err, nats.ErrTimeout) {
return 0, fmt.Errorf("events: vocabulary bounds: %w", err)
}
if len(msgs) == 0 {
return 0, nil
}
meta, err := msgs[0].Metadata()
if err != nil {
return 0, fmt.Errorf("events: vocabulary bounds: %w", err)
}
return meta.Sequence.Stream, nil
}