Sandbar/internal/sandbar/notify.go

package sandbar

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"net/smtp"
	"net/url"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"git.veillax.com/Veillax/reef/netguard"
)

// Notify is the ntfy-shaped part: anything can publish a message to a topic with
// an HTTP request, and anything can subscribe to it (SSE, newline-delimited JSON,
// or the web UI). Alerts and incidents publish here too, so a phone with the ntfy
// app subscribed to "alerts" is a working pager with no other setup.

type Message struct {
	ID       string   `json:"id"`
	Time     int64    `json:"time"`
	Event    string   `json:"event"` // always "message"; the ntfy apps ignore lines without it
	Topic    string   `json:"topic"`
	Title    string   `json:"title,omitempty"`
	Message  string   `json:"message"`
	Priority int      `json:"priority"`
	Tags     []string `json:"tags,omitempty"`
	Alert    string   `json:"alert,omitempty"` // id of the on-call alert this is about; only on live delivery and push, not stored
}

var topicRe = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`)

// reservedTopics are first path segments that belong to Sandbar itself.
var reservedTopics = map[string]bool{"api": true, "status": true, "hb": true, "metrics": true, "static": true, "healthz": true}

func validTopic(t string) bool { return topicRe.MatchString(t) && !reservedTopics[t] }

// hub fans published messages out to live subscribers.
type hub struct {
	mu   sync.Mutex
	subs map[string]map[chan Message]bool
}

func newHub() *hub { return &hub{subs: map[string]map[chan Message]bool{}} }

func (h *hub) subscribe(topic string) (chan Message, func()) {
	ch := make(chan Message, 64)
	h.mu.Lock()
	if h.subs[topic] == nil {
		h.subs[topic] = map[chan Message]bool{}
	}
	h.subs[topic][ch] = true
	h.mu.Unlock()
	return ch, func() {
		h.mu.Lock()
		delete(h.subs[topic], ch)
		h.mu.Unlock()
	}
}

func (h *hub) publish(m Message) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for ch := range h.subs[m.Topic] {
		select {
		case ch <- m:
		default: // a stalled subscriber drops messages instead of blocking publishers
		}
	}
}

func parsePriority(s string) int {
	switch strings.ToLower(strings.TrimSpace(s)) {
	case "5", "max", "urgent":
		return 5
	case "4", "high":
		return 4
	case "2", "low":
		return 2
	case "1", "min":
		return 1
	}
	return 3
}

func parseTags(s string) []string {
	var out []string
	for _, t := range strings.Split(s, ",") {
		if t = strings.TrimSpace(t); t != "" {
			out = append(out, t)
		}
	}
	return out
}

// Publish stores a message and delivers it to live subscribers.
func (s *Server) Publish(topic, title, body string, prio int, tags []string) (Message, error) {
	return s.publish(topic, title, body, prio, tags, "", "")
}

func (s *Server) publish(topic, title, body string, prio int, tags []string, alertID, status string) (Message, error) {
	if !validTopic(topic) {
		return Message{}, errors.New("a topic is 1-64 letters, digits, - or _")
	}
	m := Message{ID: newID(), Topic: topic, Time: s.now().Unix(), Title: title, Message: body, Priority: prio, Tags: tags}
	tj, _ := json.Marshal(tags)
	if _, err := s.st.db.Exec(`INSERT INTO messages(id,topic,time,title,body,priority,tags) VALUES(?,?,?,?,?,?,?)`,
		m.ID, m.Topic, m.Time, m.Title, m.Message, m.Priority, string(tj)); err != nil {
		return m, err
	}
	m.Alert = alertID
	s.hub.publish(m)
	s.pushMessage(m, status)
	return m, nil
}

// MessagesSince returns stored messages for a topic after the given unix time, oldest first.
func (s *Server) MessagesSince(topic string, since int64, limit int) []Message {
	out := []Message{}
	rows, err := s.st.db.Query(`SELECT id,time,title,body,priority,tags FROM messages WHERE topic=? AND time>=? ORDER BY time, rowid LIMIT ?`, topic, since, limit)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		m := Message{Topic: topic}
		var tj string
		if rows.Scan(&m.ID, &m.Time, &m.Title, &m.Message, &m.Priority, &tj) == nil {
			json.Unmarshal([]byte(tj), &m.Tags)
			out = append(out, m)
		}
	}
	return out
}

// ---------- outputs: email and webhooks ----------

// Channel is a place broadcast notifications go: an email address list or a webhook.
type Channel struct {
	ID           string `json:"id"`
	Name         string `json:"name"`
	Kind         string `json:"kind"`   // "email" or "webhook"
	Target       string `json:"target"` // addresses, or a URL (secret: URLs often carry tokens)
	Format       string `json:"format"` // webhook body: json, slack, discord or ntfy
	AllowPrivate bool   `json:"allowPrivate"`
	Enabled      bool   `json:"enabled"`
}

// Note is what every notification is built from.
type Note struct {
	Status   string `json:"status"` // firing, resolved, acked, test, declared, update
	Kind     string `json:"kind"`   // alert, monitor, incident
	Title    string `json:"title"`
	Message  string `json:"message"`
	Severity string `json:"severity,omitempty"`
	Link     string `json:"link,omitempty"`
	Time     int64  `json:"time"`
	Alert    string `json:"alert,omitempty"` // the alert's id, for alerts
}

func (n Note) subject() string {
	switch n.Status {
	case "resolved":
		return "[Sandbar] Resolved: " + n.Title
	case "test":
		return "[Sandbar] Test notification"
	case "acked":
		return "[Sandbar] Acknowledged: " + n.Title
	}
	return "[Sandbar] " + strings.ToUpper(n.Kind[:1]) + n.Kind[1:] + ": " + n.Title
}

func (n Note) text() string {
	if n.Status == "test" {
		return "This is a test from Sandbar. If you can read this, the channel works."
	}
	t := n.Message
	if n.Link != "" {
		t += "\n" + n.Link
	}
	return t
}

func (n Note) priority() int {
	switch {
	case n.Status == "resolved" || n.Status == "acked" || n.Status == "test":
		return 3
	case n.Severity == "critical":
		return 5
	}
	return 4
}

func (n Note) tags() []string {
	switch n.Status {
	case "resolved":
		return []string{"white_check_mark"}
	case "acked":
		return []string{"eyes"}
	case "test":
		return []string{"test_tube"}
	}
	return []string{"rotating_light"}
}

func webhookBody(format string, n Note) (body []byte, ctype string, hdr map[string]string) {
	text := n.subject() + "\n" + n.text()
	switch format {
	case "slack":
		body, _ = json.Marshal(map[string]string{"text": text})
	case "discord":
		body, _ = json.Marshal(map[string]string{"content": text})
	case "ntfy":
		return []byte(n.text()), "text/plain", map[string]string{
			"Title": n.subject(), "Priority": strconv.Itoa(n.priority()), "Tags": strings.Join(n.tags(), ","),
		}
	default:
		body, _ = json.Marshal(n)
	}
	return body, "application/json", nil
}

func (s *Server) sendChannel(ch Channel, n Note) error {
	switch ch.Kind {
	case "email":
		var errs []error
		for _, to := range strings.Split(s.st.open(ch.Target), ",") {
			if to = strings.TrimSpace(to); to != "" {
				if err := s.Mail(to, n.subject(), n.text()); err != nil {
					errs = append(errs, err)
				}
			}
		}
		return errors.Join(errs...)
	case "webhook":
		body, ctype, hdr := webhookBody(ch.Format, n)
		req, err := http.NewRequest("POST", s.st.open(ch.Target), bytes.NewReader(body))
		if err != nil {
			return err
		}
		req.Header.Set("Content-Type", ctype)
		for k, v := range hdr {
			req.Header.Set(k, v)
		}
		resp, err := s.httpClient(ch.AllowPrivate).Do(req)
		if err != nil {
			return err
		}
		resp.Body.Close()
		if resp.StatusCode >= 300 {
			return fmt.Errorf("the webhook answered %s", resp.Status)
		}
		return nil
	}
	return errors.New("unknown channel kind")
}

func (s *Server) httpClient(allowPrivate bool) *http.Client {
	if s.opts.HTTPClient != nil {
		return s.opts.HTTPClient
	}
	c := netguard.Client(allowPrivate)
	c.Timeout = 15 * time.Second
	return c
}

// Mail sends one plain-text email through the configured SMTP server.
func (s *Server) Mail(to, subject, body string) error {
	if s.opts.Mailer != nil {
		return s.opts.Mailer(to, subject, body)
	}
	c := s.opts.SMTP
	if c.Host == "" {
		return errors.New("no SMTP server configured (SANDBAR_SMTP_HOST)")
	}
	if strings.ContainsAny(to+subject, "\r\n") {
		return errors.New("invalid header")
	}
	port := c.Port
	if port == "" {
		port = "587"
	}
	msg := "From: " + c.From + "\r\nTo: " + to + "\r\nSubject: " + subject +
		"\r\nDate: " + s.now().Format(time.RFC1123Z) + "\r\nContent-Type: text/plain; charset=utf-8\r\n\r\n" + body + "\r\n"
	var auth smtp.Auth
	if c.User != "" {
		auth = smtp.PlainAuth("", c.User, c.Password, c.Host)
	}
	return smtp.SendMail(c.Host+":"+port, auth, c.From, []string{to}, []byte(msg))
}

// broadcast sends a note to the topic named by the source and every enabled channel.
func (s *Server) broadcast(n Note, topics ...string) {
	for _, t := range topics {
		if _, err := s.publish(t, n.subject(), n.text(), n.priority(), n.tags(), n.Alert, n.Status); err != nil {
			log.Printf("publish %s: %v", t, err)
		}
	}
	chans, _ := listOf[Channel](s.st, "channel")
	for _, ch := range chans {
		if !ch.Enabled {
			continue
		}
		ch := ch
		go func() {
			if err := s.sendChannel(ch, n); err != nil {
				log.Printf("channel %q: %v", ch.Name, err)
			}
		}()
	}
}

// validateChannel normalises ch.
func validateChannel(ch *Channel) error {
	ch.Name = strings.TrimSpace(ch.Name)
	ch.Target = strings.TrimSpace(ch.Target)
	switch ch.Kind {
	case "email":
		for _, a := range strings.Split(ch.Target, ",") {
			if a = strings.TrimSpace(a); !strings.Contains(a, "@") || strings.ContainsAny(a, " \r\n<>") {
				return errors.New("enter one or more email addresses, separated by commas")
			}
		}
		ch.Format, ch.AllowPrivate = "", false
	case "webhook":
		u, err := url.Parse(ch.Target)
		if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
			return errors.New("the webhook must be an http:// or https:// URL")
		}
		switch ch.Format {
		case "json", "slack", "discord", "ntfy":
		case "":
			ch.Format = "json"
		default:
			return errors.New("format must be json, slack, discord or ntfy")
		}
	default:
		return errors.New("a channel is an email or a webhook")
	}
	if ch.Name == "" {
		ch.Name = ch.Kind
	}
	return nil
}