llmgw/internal/cli/serverinit_notify.go

package cli

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/url"
	"slices"
	"strings"
	"time"

	"go.yaml.in/yaml/v3"

	"gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi"
	"gitea.mynodes.xyz/ilker/llmgw/internal/notify"
	"gitea.mynodes.xyz/ilker/llmgw/internal/settings"
)

// Where the cluster says what went wrong: a model parked, a node lost, a
// budget spent. notify.webhooks existed, and setup never asked for one, so a
// site learnt of a lost node from its users.

// parseWebhook reads a webhook as setup takes it: an address, with "slack:"
// or "json:" in front to say what it is sent, else what its address says.
func parseWebhook(s string) (notify.Webhook, error) {
	s = strings.TrimSpace(s)
	format := ""
	for _, f := range []string{"slack", "json"} {
		if rest, ok := strings.CutPrefix(s, f+":"); ok && !strings.HasPrefix(rest, "//") {
			format, s = f, rest
		}
	}
	u, err := url.Parse(s)
	if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {
		return notify.Webhook{}, fmt.Errorf("webhook %q: an http(s) address, with slack: or json: in front to say its format", s)
	}
	if format == "" {
		format = webhookFormat(u)
	}
	return notify.Webhook{URL: s, Format: format}, nil
}

// webhookFormat is what an address takes: Slack's own, and the incoming
// webhooks that read Slack's {"text": …} — Mattermost's and Rocket.Chat's
// /hooks/, Discord's …/slack — are sent that; anything else the event as
// JSON.
func webhookFormat(u *url.URL) string {
	switch {
	case u.Host == "hooks.slack.com",
		strings.Contains(u.Path, "/hooks/"),
		strings.HasSuffix(u.Path, "/slack"):
		return "slack"
	}
	return "json"
}

// parseWebhooks reads a list of them, by commas or as given.
func parseWebhooks(list []string) ([]notify.Webhook, error) {
	var out []notify.Webhook
	for _, item := range list {
		for _, s := range strings.Split(item, ",") {
			if strings.TrimSpace(s) == "" {
				continue
			}
			w, err := parseWebhook(s)
			if err != nil {
				return nil, err
			}
			out = append(out, w)
		}
	}
	return out, nil
}

// webhookEntry is one of notify.webhooks as the settings document has it.
type webhookEntry struct {
	URL        string   `yaml:"url"`
	Format     string   `yaml:"format,omitempty"`
	Events     []string `yaml:"events,omitempty,flow"`
	SecretFile string   `yaml:"secret_file,omitempty"`
}

// withWebhooks adds these webhooks to a settings document's notify.webhooks,
// and says whether that changed it: one already there, by address, is left
// as it is, with whatever events and secret it was given.
func withWebhooks(doc *settings.Doc, add []notify.Webhook) (bool, error) {
	path := []string{"notify", "webhooks"}
	var list []webhookEntry
	if n := doc.Node(path); n != nil {
		if err := n.Decode(&list); err != nil {
			return false, fmt.Errorf("notify.webhooks in the cluster's settings: %w", err)
		}
	}
	changed := false
	for _, w := range add {
		if !slices.ContainsFunc(list, func(e webhookEntry) bool { return e.URL == w.URL }) {
			list = append(list, webhookEntry{URL: w.URL, Format: w.Format})
			changed = true
		}
	}
	if !changed {
		return false, nil
	}
	b, err := yaml.Marshal(list)
	if err != nil {
		return false, err
	}
	return true, doc.SetYAML(path, string(b))
}

// ensureWebhooks puts the webhooks given into the cluster's settings and
// sends each a test, once the control plane runs on them. Said, not fatal: a
// cluster without its alerts still serves.
func (in *serverInit) ensureWebhooks(ctx context.Context) error {
	if len(in.webhooks) == 0 {
		return nil
	}
	hooks, err := parseWebhooks(in.webhooks)
	if err != nil {
		return err
	}
	c := in.adminAPI()
	var cur adminapi.SettingsView
	if err := c.get(ctx, "/api/v1/settings", &cur); err != nil {
		return err
	}
	doc, err := settings.FromBytes("settings", []byte(cur.Body))
	if err != nil {
		return err
	}
	changed, err := withWebhooks(doc, hooks)
	if err != nil {
		return err
	}
	if changed {
		b, err := doc.Bytes()
		if err != nil {
			return err
		}
		req, _ := json.Marshal(adminapi.SettingsRequest{Base: cur.Version, Body: string(b), Note: "notification webhooks, from setup"})
		if err := c.put(ctx, "/api/v1/settings", req, &cur); err != nil {
			return err
		}
		in.done("notification webhooks in the cluster's settings (version %d)", cur.Version)
	} else {
		in.kept("notification webhooks (already in the cluster's settings)")
	}

	// The control planes start again on the change; the test is sent by
	// one that knows every webhook asked for.
	var res []notify.Result
	err = waitFor(ctx, 2*time.Minute, "the control plane to run on its webhooks", func() (bool, error) {
		res = nil
		if c.post(ctx, "/api/v1/notify/test", nil, &res) != nil {
			return false, nil
		}
		for _, w := range hooks {
			// Answered by host: the path of a webhook is its secret.
			if !slices.ContainsFunc(res, func(r notify.Result) bool { return r.URL == notify.Redact(w.URL) }) {
				return false, nil
			}
		}
		return true, nil
	})
	if err != nil {
		return err
	}
	var failed []string
	for _, r := range res {
		if !slices.ContainsFunc(hooks, func(w notify.Webhook) bool { return notify.Redact(w.URL) == r.URL }) {
			continue
		}
		if r.Err != "" {
			failed = append(failed, r.URL+": "+r.Err)
			continue
		}
		in.done("a test notification reached %s", r.URL)
	}
	if len(failed) > 0 {
		return errors.New("a test notification did not arrive: " + strings.Join(failed, "; ") +
			" (kept in the settings: 'llmgw ctl notify test' tries again)")
	}
	return nil
}