package events
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/nats-io/nats.go/jetstream"
"craftdesign.group/quarry/internal/evt"
)
// Logged is one raised event, as the event log shows it.
type Logged struct {
Kind evt.Kind
Repo string
Actor string
Summary string
At time.Time
Subject string
}
// scanCap bounds how many stream messages one Recent call will read.
//
// The stream is every repository's events interleaved, and JetStream
// cannot be read backwards, so answering "the last N for THIS repo"
// means walking forward through a filtered view. The cap stops a busy
// fleet from turning one page render into an unbounded read; a repo past
// it shows its most recent events within the window that was scanned,
// which is the honest degradation — never a wrong list, sometimes a
// short one.
const scanCap = 4096
// Recent returns a repository's most recent events, newest first.
//
// Read straight off the vocabulary stream: no table, no fold, nothing to
// keep in sync. That is the whole reason this can exist cheaply — the
// stream already holds a day of every event with its kind, actor and
// payload, so the log is a VIEW of what the forge raised rather than a
// second copy of it that could disagree.
//
// The window is therefore the stream's window: 24 hours (§2.3.6). An
// event log that claimed to be history would be lying about a live tail,
// so callers must say what they are showing.
func (s *Server) Recent(repo string, limit int) ([]Logged, error) {
if s == nil || repo == "" || limit <= 0 {
return nil, nil
}
js, err := jetstream.New(s.nc)
if err != nil {
return nil, fmt.Errorf("events: jetstream: %w", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
stream, err := js.Stream(ctx, VocabStream)
if err != nil {
if errors.Is(err, jetstream.ErrStreamNotFound) {
return nil, nil
}
return nil, fmt.Errorf("events: stream: %w", err)
}
// An ordered ephemeral consumer: no durable to name, no state left
// behind, and it cannot interfere with the fan-out's own consumer.
// Exactly a noun and an action after the name: repository names may
// contain dots, and "quarry.evt.a.>" is also a.b's — private, maybe.
cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{
FilterSubjects: []string{"quarry.evt." + repo + ".*.*"},
DeliverPolicy: jetstream.DeliverAllPolicy,
})
if err != nil {
return nil, fmt.Errorf("events: consumer: %w", err)
}
// Ask how much there is before asking for it. A Fetch on a repository
// with no events waits out its whole timeout to return nothing, which
// would make "nothing has happened yet" the slowest page on the site.
info, err := cons.Info(ctx)
if err != nil {
return nil, fmt.Errorf("events: consumer info: %w", err)
}
pending := int(info.NumPending)
if pending == 0 {
return nil, nil
}
want := pending
if want > scanCap {
want = scanCap
}
// A ring: the newest `limit` seen, without holding the whole window.
ring := make([]Logged, 0, limit)
seen := 0
batch, err := cons.Fetch(want, jetstream.FetchMaxWait(2*time.Second))
if err != nil {
return nil, fmt.Errorf("events: fetch: %w", err)
}
for m := range batch.Messages() {
seen++
if e, ok := loggedOf(m); ok && e.Repo == repo {
if len(ring) == limit {
ring = ring[1:]
}
ring = append(ring, e)
}
if seen >= want {
break
}
}
if err := batch.Error(); err != nil {
return nil, fmt.Errorf("events: fetch: %w", err)
}
// Newest first: the ring is oldest-first as read.
out := make([]Logged, 0, len(ring))
for i := len(ring) - 1; i >= 0; i-- {
out = append(out, ring[i])
}
return out, nil
}
// loggedOf decodes one stream message. An undecodable message is skipped
// rather than shown as a blank row: a log that renders garbage teaches a
// reader to stop trusting it.
func loggedOf(m jetstream.Msg) (Logged, bool) {
var r struct {
Kind evt.Kind `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
if err := json.Unmarshal(m.Data(), &r); err != nil || r.Kind == "" {
return Logged{}, false
}
str := func(k string) string {
v, _ := r.Payload[k].(string)
return v
}
e := Logged{Kind: r.Kind, Repo: r.Repo, Actor: str("actor"),
Summary: str("summary"), Subject: m.Subject()}
if md, err := m.Metadata(); err == nil {
e.At = md.Timestamp
}
return e, true
}