mesh-catalog/modules/mesh-delivery/cmd/mesh-delivery/main.go

// mesh-delivery: the holder of the mesh-delivery seat (novox/hq ADR 0239, to-be 47). A Go bundle the node's
// runtime launches. It owns the delivery — one commit in one repository, from its pull request's head to
// every machine — and the delivery group above it: their states in one compiled table, kept in its own state
// on the bus, every transition said as an event, noted on the commit and shown on the pull request through
// the forge's holder. It never sends to a machine: it asks the controller, through its seat's verbs. stdout
// is the MCP channel; what this module says, it says on stderr.
package main

import (
	"encoding/json"
	"fmt"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
)

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

// listening is whether the events this owner follows reach it, in words.
type listening struct {
	mu  sync.Mutex
	now string
}

func (l *listening) set(s string) { l.mu.Lock(); l.now = s; l.mu.Unlock() }
func (l *listening) get() string  { l.mu.Lock(); defer l.mu.Unlock(); return l.now }

// The events this owner follows: the forge's pull requests, and the controller's verdicts and walks.
var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed",
	ControllerSeat + ".checked", ControllerSeat + ".plan-moved"}

func main() {
	h := &Holder{
		Store:      kvStore{},
		Controller: seatController{ask: stdio.Ask},
		Forge:      toolForge{ask: stdio.Ask},
		Emit:       func(event string, body any) error { return stdio.Emit(event, body) },
		Logf:       logf,
	}
	h.init()
	l := &listening{now: "not yet: starting"}
	go run(h, l)
	if err := stdio.Serve("", tools(h, l)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// run reads back the state, then the controller's walks, then takes the events, and keeps time — each
// retried, each failure said, never given up on quietly.
func run(h *Holder, l *listening) {
	time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it
	for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
		err := h.Load()
		if err == nil {
			break
		}
		logf("[mesh-delivery] cannot read back its deliveries yet (%v); asking again in %s", err, wait)
		time.Sleep(wait)
	}
	if err := h.Reconcile(); err != nil {
		logf("[mesh-delivery] the controller's walks cannot be read yet (%v); asked again each half minute", err)
	}
	go func() {
		tick := time.NewTicker(10 * time.Second)
		defer tick.Stop()
		last := time.Now()
		for now := range tick.C {
			if now.Sub(last) >= 30*time.Second {
				last = now
				if err := h.Reconcile(); err != nil {
					logf("[mesh-delivery] the controller's walks cannot be read: %v", err)
				}
			}
			h.Tick()
			h.Flush()
		}
	}()
	for _, pattern := range followed {
		pattern := pattern
		for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
			err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { return handle(h, e) })
			if err == nil {
				break
			}
			l.set("not yet: " + err.Error())
			logf("[mesh-delivery] not hearing %s yet (%v); asking again in %s", pattern, err, wait)
			time.Sleep(wait)
		}
	}
	l.set("listening")
	logf("[mesh-delivery] listening: %s", strings.Join(followed, ", "))
}

// handle takes one event. An event that cannot be read is said and acknowledged: read again it would fail
// again, and the controller's walks are read back by comparison anyway.
func handle(h *Holder, e stdio.Envelope) error {
	switch e.Key {
	case "gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed":
		var p PullEvent
		if err := json.Unmarshal(e.Body, &p); err != nil {
			h.Refuse(fmt.Errorf("%s is not readable: %v", e.Key, err))
			return nil
		}
		switch e.Key {
		case "gitea.pull.updated":
			h.PullUpdated(p)
		case "gitea.pull.merged":
			h.PullMerged(p)
		default:
			h.PullClosed(p)
		}
	case ControllerSeat + ".checked":
		var c CheckedEvent
		if err := json.Unmarshal(e.Body, &c); err != nil {
			h.Refuse(fmt.Errorf("a verdict is not readable: %v", err))
			return nil
		}
		h.Checked(c)
	case ControllerSeat + ".plan-moved":
		var w Walk
		if err := json.Unmarshal(e.Body, &w); err != nil {
			h.Refuse(fmt.Errorf("a walk is not readable: %v", err))
			return nil
		}
		h.WalkMoved(w)
	}
	return nil
}

func str(description string) map[string]any {
	return map[string]any{"type": "string", "description": description}
}

func strArg(a map[string]any, k string) string { s, _ := a[k].(string); return strings.TrimSpace(s) }

// intArg is a whole number given as a number or as text; absent is zero.
func intArg(a map[string]any, k string) (int, error) {
	switch v := a[k].(type) {
	case nil:
		return 0, nil
	case float64:
		if v == float64(int(v)) {
			return int(v), nil
		}
	case int:
		return v, nil
	case string:
		v = strings.TrimPrefix(strings.TrimSpace(v), "#")
		if v == "" {
			return 0, nil
		}
		if n, err := strconv.Atoi(v); err == nil {
			return n, nil
		}
	}
	return 0, fmt.Errorf("%q is not a whole number: %v", k, a[k])
}

func boolArg(a map[string]any, k string) bool {
	switch v := a[k].(type) {
	case bool:
		return v
	case string:
		return v == "true"
	}
	return false
}

// aPerson is who an act through the seat is by: the seat does not name its caller.
const aPerson = "a person, through the mesh-delivery seat"

func need(a map[string]any, keys ...string) error {
	for _, k := range keys {
		if strArg(a, k) == "" {
			return fmt.Errorf("needs %q", k)
		}
	}
	return nil
}

func tools(h *Holder, l *listening) []stdio.Tool {
	seat := "mesh-delivery."
	return []stdio.Tool{
		{Name: seat + "deliveries",
			Description: "Every delivery not final, and those that ended in the last day, one line each: its id " +
				"(owner/repository@commit), its state, its group, what it waits for and since when. Narrowed by state, " +
				"repository or group.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"state": str("one state"),
				"repository": str("owner/repository"), "group": str("a group's id (its branch name)"),
				"all": map[string]any{"type": "string", "enum": []string{"true", "false"}}}},
			Run: func(a map[string]any) (any, error) {
				return h.Deliveries(strArg(a, "state"), strArg(a, "repository"), strArg(a, "group"), boolArg(a, "all")), nil
			}},
		{Name: seat + "show",
			Description: "One delivery or group whole: its delivery plan, every transition with when and why, the " +
				"machine steps of its walk, its group and its order.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("a delivery's id, or a group's")},
				"required": []string{"id"}},
			Run: func(a map[string]any) (any, error) { return h.Show(strArg(a, "id")) }},
		{Name: seat + "groups",
			Description: "Every delivery group: its members in order, why each pair is ordered, its composed check " +
				"and its state, derived from its members.",
			Input: map[string]any{"type": "object", "properties": map[string]any{
				"all": map[string]any{"type": "string", "enum": []string{"true", "false"}}}},
			Run: func(a map[string]any) (any, error) { return h.Groups(boolArg(a, "all")), nil }},
		{Name: seat + "what-if",
			Description: "The delivery plan a change would have, asked of the controller's planner and kept nowhere.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"repository": str("owner/repository"),
				"paths": str("the files it changes, comma-separated"), "base": str("the branch it merges into (default main)")},
				"required": []string{"repository", "paths"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "repository", "paths"); err != nil {
					return nil, err
				}
				return h.WhatIf(strArg(a, "repository"), strArg(a, "base"), strings.Split(strArg(a, "paths"), ","))
			}},
		{Name: seat + "checks",
			Description: "What the mesh's checks said of a pull request's head or of one commit: each of the commit's " +
				"statuses (mesh/merge-gate, mesh/repo-check, mesh/delivery, …) with its state, description and when it " +
				"was set; the merge check's full verdict as the controller said it — each layer's summary, the machine " +
				"that ran it, when, its build id and its report; and whether the branch's protection would let it " +
				"merge, every required status being success.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"repository": str("owner/repository"),
				"number": str("a pull request's number: its head is read"),
				"commit": str("a commit's sha, or the start of one, instead of a pull request")},
				"required": []string{"repository"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "repository"); err != nil {
					return nil, err
				}
				number, err := intArg(a, "number")
				if err != nil {
					return nil, err
				}
				return h.Checks(strArg(a, "repository"), number, strArg(a, "commit"))
			}},
		{Name: seat + "table",
			Description: "The state table every delivery runs by: each transition with its guard, each state's bound " +
				"and what healer H2 may do once it has passed; and the machine steps' table.",
			Run: func(map[string]any) (any, error) { return TableText(), nil }},
		{Name: seat + "stalled",
			Description: "Every delivery held past its state's bound, with the transition the table lets healer H2 take.",
			Run:         func(map[string]any) (any, error) { return h.Stalled(), nil }},
		{Name: seat + "recheck",
			Description: "Check a rejected or ready delivery again: proposed again, and its check asked. With why.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
				"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "id", "why"); err != nil {
					return nil, err
				}
				return h.Recheck(strArg(a, "id"), strArg(a, "why"), aPerson)
			}},
		{Name: seat + "release",
			Description: "A person's word that a held delivery goes on: it starts delivering. With why.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
				"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "id", "why"); err != nil {
					return nil, err
				}
				return h.Release(strArg(a, "id"), strArg(a, "why"), aPerson)
			}},
		{Name: seat + "stop",
			Description: "Stop a delivery that is not final, with why: its walk is ended through the controller, and " +
				"in a delivering group every member after it is stopped too, naming it.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
				"why": str("why, kept with the transition")}, "required": []string{"id", "why"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "id", "why"); err != nil {
					return nil, err
				}
				return h.Stop(strArg(a, "id"), strArg(a, "why"), aPerson)
			}},
		{Name: seat + "close",
			Description: "Healer H2's verb: for a delivery held past its bound, read its walk again and take the " +
				"transition the table names for that state. Refused for any other.",
			Input: map[string]any{"type": "object", "properties": map[string]any{"id": str("the delivery's id"),
				"why": str("the condition it answers")}, "required": []string{"id", "why"}},
			Run: func(a map[string]any) (any, error) {
				if err := need(a, "id", "why"); err != nil {
					return nil, err
				}
				return h.Close(strArg(a, "id"), strArg(a, "why"))
			}},
		{Name: "delivery_status",
			Description: "Whether this owner works: how many deliveries in each state, the groups, what is owed to " +
				"the forge and the bus and not yet done, walks with no delivery yet, whether its events arrive, when " +
				"the controller's walks were last read or why they could not be, and the last refusals.",
			Run: func(map[string]any) (any, error) { return h.Status(l.get()), nil }},
	}
}