Artifact [b01476cc1a]

Artifact b01476cc1ae24d1f034f4cc6178f31d0345a154f123fa0d75938bdec74e839eb:

• File internal/webhook/fanout.go — part of check-in [f2ea5f5c1a] at 2026-08-24 17:13:12 on branch hooks-vocabulary — Webhooks fan out from the vocabulary, not from the ops layer

  A ticket closed by push notified nobody, because only ops queued webhooks. One consumer on quarry.evt.> turns every raised event into deliveries, so arrival path stops mattering.

  Deletes kindOf: the delivery's kind was sliced off an idempotency key. (user: dmestas, size: 4840) [annotate] [blame] [check-ins using] [more...]

package webhook

import (
	"context"
	"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
	}
	if err := d.Raise(eventOf(r, d.now())); 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 string) Event {
	str := func(k string) string {
		s, _ := r.Payload[k].(string)
		return s
	}
	if d := str("date"); d != "" {
		at = d
	}
	return Event{
		// Stable across redeliveries: the same artifact producing the
		// same event is the same delivery, whichever node folded it.
		ID:      r.Kind + ":" + str("hash"),
		Repo:    r.Repo,
		Kind:    r.Kind,
		Actor:   str("actor"),
		Summary: summaryOf(r, str),
		Link:    linkOf(r, str),
		At:      at,
	}
}

// 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)
}