// robot.go is RFC 0011 §10's activity robot: fossil's one good chat
// idea — the timeline talking in the channel — done on quarry's own
// event stream.
//
// One consumer of the vocabulary stream serves every repository. A
// repository opts in by storing event patterns (`chat.robot`, webhook
// pattern syntax), and each matching event becomes one compact
// `Kind: robot` line in that repository's channel.
//
// Robot lines are the channel talking about the repository, not a
// person addressing anyone: they never count as unread (see Unread) and
// they never mention anybody — Post resolves no mentions, so nothing a
// robot line quotes can trigger a notification.
package chat
import (
"context"
"encoding/json"
"errors"
"fmt"
"log"
"strconv"
"strings"
"time"
"github.com/nats-io/nats.go"
"craftdesign.group/quarry/internal/evt"
)
// robotDurable is the FLEET's durable on the vocabulary stream, not a
// per-node one: every node binds this same name, so an event is
// announced in its channel once however many nodes are up. One
// consumer, not one per repository either — the fan-out to channels
// happens after the read, so a forge with ninety opted-in repositories
// still reads the stream once.
const robotDurable = "chat-robot"
// robotBackoff paces the loop after a fetch fails for a reason that is
// not "nothing arrived". A timeout returns after its own wait and is
// the ordinary case; a broken connection or a deleted consumer returns
// AT ONCE, and retrying that as fast as the CPU allows is a spin that
// fills the log and starves the node.
const robotBackoff = 2 * time.Second
// vocabSubject is RFC 0005's subject space. Named here rather than
// imported so this package stays a leaf on the event plane: it consumes
// a subject, not a Go type from another package.
const vocabSubject = "quarry.evt.>"
// Robot turns vocabulary events into channel lines.
type Robot struct {
room *Room
js nats.JetStreamContext
// Patterns answers one repository's robot event patterns, empty
// when its robot is off. Read per event rather than cached: an
// administrator turning the robot on should not have to wait for a
// restart.
Patterns func(repo string) string
// Code maps a repository NAME to its project code — the channel
// key. Events carry the name; channels are keyed by the code, which
// is stable across renames.
Code func(repo string) (string, bool)
// Number resolves the !N of a landing or the #N of a ticket
// (noun "landing" or "ticket") by its id, 0 when none is known.
// Optional: without it, or before the thing is numbered, the line
// keeps the short hash and is resolved when it is shown.
Number func(repo, noun, id string) int
}
// NewRobot returns the consumer. patterns and code are required; a nil
// either way makes Start a no-op, which is how a forge with no OLTP
// simply runs no robot.
func NewRobot(room *Room, js nats.JetStreamContext,
patterns func(repo string) string, code func(string) (string, bool)) *Robot {
return &Robot{room: room, js: js, Patterns: patterns, Code: code}
}
// vocabEvent is one raised event as internal/events publishes it.
type vocabEvent struct {
Kind string `json:"kind"`
Repo string `json:"repo"`
Payload map[string]any `json:"payload"`
}
// Start begins consuming until ctx is done.
//
// DeliverNew: the vocabulary stream holds a day, and a first start must
// not read that day as a backlog to announce — a robot switched on this
// morning replaying yesterday into the channel is the exact failure
// that teaches people to mute it.
func (r *Robot) Start(ctx context.Context) error {
if r == nil || r.room == nil || r.js == nil || r.Patterns == nil || r.Code == nil {
return nil
}
sub, err := r.js.PullSubscribe(vocabSubject, robotDurable,
nats.DeliverNew(), nats.AckWait(30*time.Second))
if err != nil {
return fmt.Errorf("chat: robot subscribe: %w", err)
}
go func() {
for ctx.Err() == nil {
msgs, ferr := sub.Fetch(32, nats.MaxWait(2*time.Second))
if errors.Is(ferr, nats.ErrTimeout) {
continue // nothing was raised in that window
}
if ferr != nil {
log.Printf("chat: robot: fetch: %v", ferr)
select {
case <-ctx.Done():
return
case <-time.After(robotBackoff):
}
continue
}
for _, m := range msgs {
r.announce(m)
}
}
}()
return nil
}
func (r *Robot) announce(m *nats.Msg) {
var e vocabEvent
if err := json.Unmarshal(m.Data, &e); err != nil {
// Undecodable: retrying cannot help.
log.Printf("chat: robot: bad event on %s: %v", m.Subject, err)
_ = m.Ack()
return
}
_ = m.Ack()
if !r.Wants(e.Repo, e.Kind) {
return
}
code, ok := r.Code(e.Repo)
if !ok {
return
}
payload := e.Payload
if noun, _, _ := strings.Cut(e.Kind, "."); r.Number != nil && (noun == "landing" || noun == "ticket") {
id, _ := payload[noun].(string)
if id == "" && noun == "landing" {
id, _ = payload["thread"].(string)
}
if n := r.Number(e.Repo, noun, id); n > 0 {
payload = make(map[string]any, len(e.Payload)+1)
for k, v := range e.Payload {
payload[k] = v
}
payload["number"] = n
}
}
line := RobotLine(e.Kind, payload)
if line == "" {
return
}
if _, err := r.room.Post(code, Message{
Kind: KindRobot, Origin: "robot", User: "robot", Text: line,
}); err != nil {
log.Printf("chat: robot: post to %s: %v", e.Repo, err)
}
}
// Wants reports whether this repository's robot announces this kind. An
// empty pattern is off — never "everything", which is the direction a
// mistake here must not fail in.
func (r *Robot) Wants(repo, kind string) bool {
pats := strings.Fields(r.Patterns(repo))
if len(pats) == 0 {
return false
}
return evt.Kind(kind).MatchesAny(pats)
}
// RobotLine renders one event as the channel line for it: one line,
// past tense, naming the thing and whoever did it. Exported because it
// is the robot's whole vocabulary and a test asserting the shape should
// read the same function the channel does.
//
// A landing or ticket whose number the payload carries ("number") is
// named by it, !N or #N, bare: the reader's icon says what it is. With
// no number it is "landing <hash>", which display resolves later.
//
// An event this has nothing useful to say about returns "", and the
// robot stays quiet rather than announcing a kind read aloud.
func RobotLine(kind string, payload map[string]any) string {
str := func(k string) string {
s, _ := payload[k].(string)
return s
}
noun, action, _ := strings.Cut(kind, ".")
named := func(word, sigil, id string) string {
switch n := payload["number"].(type) {
case int:
if n > 0 {
return sigil + strconv.Itoa(n)
}
case float64:
if n > 0 {
return sigil + strconv.Itoa(int(n))
}
}
return word + " " + short(id)
}
by := func() string {
if a := str("actor"); a != "" {
return " by " + a
}
return ""
}
// The comment's own words where there are any: the robot's job is
// to carry the repository's voice, not to paraphrase it.
tail := func() string {
if s := str("summary"); s != "" {
return " — " + s
}
return ""
}
switch noun {
case "checkin":
where := ""
if b := str("branch"); b != "" {
where = " on " + b
}
return "check-in " + short(str("hash")) + where + by() + tail()
case "branch":
return "branch " + str("branch") + " " + action + by()
case "tag":
return "tag " + str("tag") + " " + action + by()
case "landing":
id := str("landing")
if id == "" {
id = str("thread")
}
return named("landing", "!", id) + " " + action + by() + tail()
case "ticket":
return named("ticket", "#", str("ticket")) + " " + action + by() + tail()
case "wiki":
return "wiki page " + str("page") + " " + action + by()
case "technote":
return "note " + short(str("note")) + " " + action + by() + tail()
case "forum":
return "forum thread " + short(str("thread")) + " " + action + by() + tail()
case "ci":
state := str("state")
if state == "" {
state = action
}
return "CI " + state + " on " + short(str("hash")) + tail()
case "mirror":
return "mirror " + action
case "uv":
// One action for both, as fossil has one card for both; the
// payload says which it was.
if deleted, _ := payload["deleted"].(bool); deleted {
return "unversioned file " + str("name") + " removed" + by()
}
return "unversioned file " + str("name") + " published" + by()
}
return ""
}
// short is quarry's own link length for a hash in prose: long enough to
// be unambiguous in a store, short enough to read in a chat line.
func short(s string) string {
if len(s) > 12 {
return s[:12]
}
return s
}