package webhook
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"strings"
"time"
"github.com/nats-io/nats.go"
)
// fanoutDurable consumes the vocabulary stream. One consumer, because
// the fan-out to subscriptions happens after: a repo with nine endpoints
// is nine stream messages on QUARRYHOOK, not nine reads of QUARRYEVT.
const (
fanoutDurable = "hook-fanout"
vocabSubject = "quarry.evt.>"
)
// raised is one vocabulary event as internal/events publishes it.
type raised struct {
Kind string `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
// Fanout turns the vocabulary into webhook deliveries.
//
// This is where Principle 6 stops depending on how a change ARRIVED. An
// op used to queue its own webhook, so a ticket closed through the forge
// notified subscribers and the same ticket closed by a push notified
// nobody — the endpoint's owner could not have known the difference
// existed. Both now raise the same event on the same subject, and this
// consumer is the only thing that turns one into a delivery.
//
// It also ends the guessing. The old path labelled a delivery by slicing
// a noun off the front of an idempotency key, which made the event's
// TYPE a property of how somebody spelled a transaction id. The subject
// carries the type now, and the type is the vocabulary's.
func (d *Dispatcher) Fanout(ctx context.Context) error {
if d == nil {
return nil
}
// DeliverNew: the vocabulary stream holds a day, and a first start
// must not read that day as a backlog of undelivered webhooks.
sub, err := d.js.PullSubscribe(vocabSubject, fanoutDurable,
nats.DeliverNew(), nats.AckWait(30*time.Second))
if err != nil {
return fmt.Errorf("webhook: fanout subscribe: %w", err)
}
go func() {
for ctx.Err() == nil {
msgs, ferr := sub.Fetch(32, nats.MaxWait(2*time.Second))
if ferr != nil {
continue // timeout is the common case
}
for _, m := range msgs {
d.fanout(m)
}
}
}()
return nil
}
func (d *Dispatcher) fanout(m *nats.Msg) {
var r raised
if err := json.Unmarshal(m.Data, &r); err != nil {
// Undecodable: retrying cannot help.
log.Printf("webhook: fanout: bad event on %s: %v", m.Subject, err)
_ = m.Ack()
return
}
// The vocabulary message's stream position keys an event that has
// no key of its own: a redelivery of the message reads the same
// position, and no other message does.
var fallback string
if md, err := m.Metadata(); err == nil {
fallback = fmt.Sprintf("seq-%d", md.Sequence.Stream)
}
if err := d.Raise(eventOf(r, d.now(), fallback)); err != nil {
log.Printf("webhook: fanout: queue %s: %v", r.Kind, err)
_ = m.Nak()
return
}
_ = m.Ack()
}
// eventOf renders one vocabulary event as the outbound shape.
func eventOf(r raised, at, fallback string) Event {
str := func(k string) string {
s, _ := r.Payload[k].(string)
return s
}
if d := str("date"); d != "" {
at = d
}
return Event{
ID: r.Kind + ":" + keyOf(str, fallback),
Repo: r.Repo,
Kind: r.Kind,
Actor: str("actor"),
Summary: summaryOf(r, str),
Link: linkOf(r, str),
At: at,
}
}
// keyOf names the one event a delivery is about. JetStream drops a
// second message with the same id inside its duplicate window, so two
// DIFFERENT events must never share a key — and a redelivery of the
// same one should.
func keyOf(str func(string) string, fallback string) string {
// Stable across redeliveries: the same artifact producing the
// same event is the same delivery, whichever node folded it.
if h := str("hash"); h != "" {
return h
}
// CI events name a run, not an artifact. The status is part of the
// key because one run raises ci.started once per status it moves
// through.
if run := str("runId"); run != "" {
return run + ":" + str("status")
}
if fallback != "" {
return fallback
}
// Nothing names this event, so nothing may make it a duplicate.
b := make([]byte, 8)
_, _ = rand.Read(b)
return "gen-" + hex.EncodeToString(b)
}
// summaryOf is one human line. The event already carries the author's
// own words where there are any; the rest is the vocabulary read aloud,
// which is deliberately dull — a subscriber parses the kind, and a
// person reading a delivery log wants to know what it was.
func summaryOf(r raised, str func(string) string) string {
if s := str("summary"); s != "" {
return s
}
noun, action, _ := strings.Cut(r.Kind, ".")
switch noun {
case "branch":
return "branch " + str("branch") + " " + action
case "tag":
return "tag " + str("tag") + " " + action
case "ticket":
return "ticket " + short(str("ticket")) + " " + action
case "wiki":
return "wiki page " + str("page") + " " + action
}
return r.Kind
}
// linkOf is where a person clicks. An event whose subject quarry has no
// page for gets no link rather than a broken one.
func linkOf(r raised, str func(string) string) string {
base := "/r/" + r.Repo
noun, _, _ := strings.Cut(r.Kind, ".")
switch noun {
case "checkin", "branch":
return base + "/checkin/" + short(str("hash"))
case "tag":
if t := str("target"); t != "" {
return base + "/checkin/" + short(t)
}
case "ticket":
return base + "/ticket/" + short(str("ticket"))
case "wiki":
return base + "/wiki/" + str("page")
case "technote":
return base + "/technote/" + short(str("note"))
case "forum":
return base + "/forum/" + short(str("thread"))
case "landing":
id := str("landing")
if id == "" {
id = str("thread")
}
return base + "/landing/" + short(id)
}
return ""
}
// short is quarry's own link length for a hash: long enough to be
// unambiguous in a store, short enough to read.
func short(s string) string {
if len(s) > 16 {
return s[:16]
}
return s
}
func (d *Dispatcher) now() string {
if d.Now != nil {
return d.Now()
}
return time.Now().UTC().Format(time.RFC3339)
}