Artifact [ab65a5d447]

Artifact ab65a5d44706a4e4d99f6d5ae2ad9107216a4b3a56cd55ca0b781e087a8fe90f:

• File internal/events/log.go — part of check-in [a5c5fe771e] at 2026-08-24 21:20:37 on branch praise-ui — Event log: proof the hooks fired

  Read straight off the vocabulary stream — no table, no fold, nothing to keep in sync, so it cannot disagree with what the forge published.

  The window is the stream's window, 24 hours, and the page says so: a log that let a reader assume history would be lying about a live tail.

  Asks the consumer how much is pending before fetching. Without that a quiet repository waits out the fetch timeout, which would make 'nothing has happened yet' the slowest page on the site — 2s became 9ms.

  Linked from the webhooks settings, not given a tab: that is where the question gets asked. (user: dmestas, size: 4307) [annotate] [blame] [check-ins using] [more...]

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.
	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 {
			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
}