package api
import (
"context"
"encoding/json"
"fmt"
"path"
"strings"
"time"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/webhook"
)
// ForgeEvent is one forge event as the event plane delivers it: what
// was raised, and where it sits in the stream every repository shares.
// Gap is the signal, not an event: the resume point (From) is older
// than the window, and the replay that follows starts at Seq.
type ForgeEvent struct {
Seq uint64
Kind string
Repo string
Payload map[string]any
At time.Time
Gap bool
From uint64
}
// EventTailPlane streams the forge's vocabulary: every repository's
// events when repo is "", else that one's, from seq from+1 (0: new
// only). A nil Deps.EventTail answers an empty stream.
type EventTailPlane interface {
TailEvents(repo string, from uint64) (<-chan ForgeEvent, func())
}
type eventsStreamIn struct {
// From resumes after this seq (Last-Event-ID, overlaid by the
// router); 0 streams new events only.
From uint64 `json:"from"`
// Kind picks events by kind: an exact kind ("ci.completed") or a
// pattern ("landing.*"). Repeat it for more; none means every kind.
Kind []string `json:"kind" validate:"max=20"`
}
type repoEventsStreamIn struct {
Repo string `json:"repo" validate:"required,max=64"`
From uint64 `json:"from"`
Kind []string `json:"kind" validate:"max=20"`
}
// eventOut is one event on the wire: quarry's own record of it, with
// the line and link its webhooks use.
type eventOut struct {
Seq uint64 `json:"seq"`
Kind string `json:"kind"`
Repo string `json:"repo"`
Actor string `json:"actor,omitempty"`
Summary string `json:"summary"`
Link string `json:"link,omitempty"`
At string `json:"at"`
// Payload is what quarry recorded about the event. Its values are
// strings: a number or a flag is written as its JSON text ("3",
// "true"), so the shape is one an API description can state.
Payload map[string]string `json:"payload,omitempty"`
}
// eventsStreamDoc is shared by both scopes of the stream.
const eventsStreamDoc = "The body is text/event-stream, not JSON: each event's `id:` line is its " +
"stream sequence and its data is {seq, kind, repo, actor, summary, link, at, payload}, where " +
"payload is what quarry recorded (a ci.completed carries runId, checkin, pipeline and status). " +
"`kind` filters by kind, exact or a pattern such as `landing.*`, and may be repeated. The first " +
"frame is `event: live`, sent once the tail is attached; nothing raised after it is missed. A " +
"client that reconnects with `Last-Event-ID: <seq>` resumes after it; a resume older than the " +
"window (a day) replays from the window's first event, preceded by an `event: gap` frame " +
"carrying {\"from\":<asked>,\"first\":<available>}. Account, team, error and settings events " +
"reach administrators only. The sequence is the FORGE'S: the ids one listener receives are a " +
"subset and gaps between them are ordinary rather than loss -- reconnecting with " +
"Last-Event-ID is the only correct way to find out what you missed."
// eventVisibleTTL is how long the forge-wide stream trusts one answer
// to "may this caller read that repository": long enough that a burst
// of one repository's events asks once, short enough that a revoked
// grant stops the flow within the minute.
const eventVisibleTTL = time.Minute
func registerEventStream(r *ops.Registry, d Deps) error {
if err := ops.Register(r, ops.Op[eventsStreamIn, eventOut]{
Name: "events.stream",
Doc: "Every forge event in the repositories the signed-in account can read, live, as " +
"Server-Sent Events. " + eventsStreamDoc,
Kind: ops.Stream,
Auth: ops.Token,
Route: "GET /api/v1/events/stream",
StreamRun: func(ctx context.Context, in eventsStreamIn, s ops.Streamer) error {
if ops.User(ctx) == "" {
return fmt.Errorf("the event stream needs a signed-in account: %w", ops.ErrInvalid)
}
return d.streamEvents(ctx, s, "", in.From, in.Kind, d.eventVisibility(ctx), d.siteAdmin(ctx))
},
}); err != nil {
return err
}
return ops.Register(r, ops.Op[repoEventsStreamIn, eventOut]{
Name: "repo.events.stream",
Caps: "o",
Doc: "One repository's forge events, live, as Server-Sent Events. " + eventsStreamDoc,
Kind: ops.Stream,
Auth: ops.Anonymous,
Route: "GET /api/v1/r/{repo}/events/stream",
StreamRun: func(ctx context.Context, in repoEventsStreamIn, s ops.Streamer) error {
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return err
}
// The router resolved this repository's letters for Caps:
// its administrator, or a site administrator, hears the
// access and settings changes.
admin := ops.CapsOf(ctx).HasAny("as") || d.siteAdmin(ctx)
return d.streamEvents(ctx, s, in.Repo, in.From, in.Kind, func(string) bool { return true }, admin)
},
})
}
// eventVisibility answers, for the forge-wide stream, whether the
// caller may read the repository an event names, asking the forge's
// own read rule once per repository per eventVisibleTTL. One stream
// owns it, so it needs no lock.
func (d Deps) eventVisibility(ctx context.Context) func(repo string) bool {
type answer struct {
ok bool
at time.Time
}
seen := map[string]answer{}
return func(repo string) bool {
if a, ok := seen[repo]; ok && time.Since(a.at) < eventVisibleTTL {
return a.ok
}
_, err := d.repoFor(ctx, repo)
seen[repo] = answer{ok: err == nil, at: time.Now()}
return err == nil
}
}
// siteAdmin says whether the caller administers the forge: a site
// token, or an account holding s at the forge. The forge-wide stream
// declares no Caps, so the router resolves no letters for it.
func (d Deps) siteAdmin(ctx context.Context) bool {
if tok, ok := ops.TokenOf(ctx); ok {
return tok.SiteAdmin()
}
who := ops.User(ctx)
return who != "" && d.CapsFor != nil && d.CapsFor(who, "").HasAny("s")
}
// streamEvents is both scopes' loop: tail, filter by kind and by what
// the caller may see, write. visible answers for a repository named in
// an event; admin says whether the caller hears adminEvent kinds.
func (d Deps) streamEvents(ctx context.Context, s ops.Streamer, repo string, from uint64,
kinds []string, visible func(repo string) bool, admin bool) error {
for _, k := range kinds {
if _, err := path.Match(k, ""); err != nil {
return fmt.Errorf("kind %q is not a pattern: %w", k, ops.ErrInvalid)
}
}
if d.EventTail == nil {
// No plane: an empty stream, not an error.
if err := s.Event("live", nil); err != nil {
return err
}
<-ctx.Done()
return nil
}
events, cancel := d.EventTail.TailEvents(repo, from)
defer cancel()
if err := s.Event("live", nil); err != nil {
return err
}
heartbeat := time.NewTicker(20 * time.Second)
defer heartbeat.Stop()
for {
select {
case <-ctx.Done():
return nil
case <-heartbeat.C:
s.Ping()
case e, open := <-events:
if !open {
return nil
}
if e.Gap {
if err := s.Event("gap", map[string]uint64{"from": e.From, "first": e.Seq}); err != nil {
return err
}
continue
}
if !kindWanted(kinds, e.Kind) || (adminEvent(e.Kind) && !admin) || !visible(e.Repo) {
continue
}
if err := s.Emit(e.Seq, eventOutOf(e)); err != nil {
return err
}
}
}
}
// kindWanted says whether kind matches one of the patterns asked for;
// no patterns means every kind.
func kindWanted(patterns []string, kind string) bool {
if len(patterns) == 0 {
return true
}
for _, p := range patterns {
if ok, _ := path.Match(p, kind); ok {
return true
}
}
return false
}
// adminEvent names the kinds a reader of a repository does not see on
// its pages: who holds which access, a team's membership, the forge's
// own failures and a repository's settings.
func adminEvent(kind string) bool {
noun, _, _ := strings.Cut(kind, ".")
switch noun {
case "account", "team", "error":
return true
}
return kind == "repo.settings"
}
// eventOutOf renders one event for the wire. The token a change came in
// on ("via") is the forge's to know, not every reader's, so it is not
// sent.
func eventOutOf(e ForgeEvent) eventOut {
payload := make(map[string]string, len(e.Payload))
for k, v := range e.Payload {
if k == "via" {
continue
}
if str, ok := v.(string); ok {
payload[k] = str
} else if b, err := json.Marshal(v); err == nil {
payload[k] = string(b)
}
}
summary, link := webhook.Describe(e.Kind, e.Repo, e.Payload)
actor, _ := e.Payload["actor"].(string)
return eventOut{
Seq: e.Seq, Kind: e.Kind, Repo: e.Repo, Actor: actor,
Summary: summary, Link: link, At: e.At.UTC().Format(time.RFC3339),
Payload: payload,
}
}